Skip to content

WIP: In memory processing using AttachmentProcessor #81

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
satwikkansal wants to merge 1 commit into
base: master
Choose a base branch
from
Open

WIP: In memory processing using AttachmentProcessor #81

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 23 additions & 8 deletions mxtoai/agents/email_agent.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,7 @@

# Import citation management and web search tools
from mxtoai.scripts.report_formatter import ReportFormatter
from mxtoai.scripts.visual_qa import azure_visualizer
from mxtoai.scripts.visual_qa import azure_visualizer_from_content
from mxtoai.tools.attachment_processing_tool import AttachmentProcessingTool
from mxtoai.tools.citation_aware_visit_tool import CitationAwareVisitTool
from mxtoai.tools.deep_research_tool import DeepResearchTool
Expand Down Expand Up @@ -94,7 +94,12 @@ class EmailAgent:
"""

def __init__(
self, email_request: EmailRequest, attachment_dir: str = "email_attachments", verbose: bool = False, enable_deep_research: bool = False, attachment_info: list[dict] | None = None
self,
email_request: EmailRequest,
attachment_dir: str = "email_attachments",
verbose: bool = False,
enable_deep_research: bool = False,
attachment_info: list[dict] | None = None,
):
"""
Initialize the email agent with tools for different operations.
Expand Down Expand Up @@ -150,7 +155,7 @@ def __init__(
self.scheduled_tasks_tool,
self.delete_scheduled_tasks_tool,
self.references_generator_tool,
azure_visualizer,
azure_visualizer_from_content,
]

# Add all available search tools
Expand Down Expand Up @@ -485,7 +490,9 @@ def _create_task_template(
f"Process this email according to the '{handle}' instruction type.\n",
email_context,
distilled_section,
RESEARCH_GUIDELINES["mandatory"] if deep_research_mandatory and self.enable_deep_research else RESEARCH_GUIDELINES["optional"],
RESEARCH_GUIDELINES["mandatory"]
if deep_research_mandatory and self.enable_deep_research
else RESEARCH_GUIDELINES["optional"],
attachment_task,
handle_specific_template,
output_template,
Expand Down Expand Up @@ -666,7 +673,11 @@ def _process_agent_result(
logger.info(f"Scheduled task created successfully with ID: {tool_output['task_id']}")
else:
error_msg = tool_output.get("message", "Scheduled task creation failed")
error_type = "Scheduled Task Limit Exceeded" if tool_output.get("error") == "Task limit exceeded" else "Scheduled Task Error"
error_type = (
"Scheduled Task Limit Exceeded"
if tool_output.get("error") == "Task limit exceeded"
else "Scheduled Task Error"
)
errors_list.append(ProcessingError(message=error_type, details=error_msg))
if tool_output.get("error") == "Task limit exceeded":
logger.warning(f"Scheduled task limit exceeded: {error_msg}")
Expand Down Expand Up @@ -939,7 +950,9 @@ def limited_forward(*args, **kwargs):
current_active_count = count_active_tasks_for_user(session, user_email)

if current_active_count >= max_calls:
logger.warning(f"Scheduled task limit reached ({max_calls} active tasks per email). User has {current_active_count} active tasks.")
logger.warning(
f"Scheduled task limit reached ({max_calls} active tasks per email). User has {current_active_count} active tasks."
)
return {
"success": False,
"error": "Task limit exceeded",
Expand All @@ -954,7 +967,6 @@ def limited_forward(*args, **kwargs):
# Call the original method
return original_forward(*args, **kwargs)


# Replace the forward method
base_tool.forward = limited_forward

Expand All @@ -974,9 +986,12 @@ def _finalize_response_with_citations(self, content: str) -> str:
if self.context.has_citations():
# Check if content already contains a References or Sources section to avoid duplication
import re

existing_references_pattern = r"(^|\n)#{1,3}\s*(References|Sources|Bibliography)\s*$"
if re.search(existing_references_pattern, content, re.MULTILINE | re.IGNORECASE):
logger.warning("Content already contains a References/Sources section - skipping automatic references to avoid duplication")
logger.warning(
"Content already contains a References/Sources section - skipping automatic references to avoid duplication"
)
return content

references_section = self.context.get_references_section()
Expand Down
27 changes: 9 additions & 18 deletions mxtoai/request_context.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,7 +64,7 @@ def load_attachment(self, filename: str, file_path: str, content_type: str, size
"filename": filename,
"contentType": content_type,
"size": size,
"original_path": file_path
"original_path": file_path,
}
logger.debug(f"Loaded attachment into memory: {filename} ({size} bytes)")
return True
Expand All @@ -90,7 +90,7 @@ def has_attachment(self, filename: str) -> bool:


class CitationManager:
"""Per-request citation manager without threading complexity."""
"""Per-request citation manager."""

def __init__(self):
self._citations = CitationCollection()
Expand Down Expand Up @@ -125,7 +125,7 @@ def add_web_source(self, url: str, title: str, description: str | None = None, *
url=url,
date_accessed=datetime.now(timezone.utc).strftime("%Y-%m-%d"),
source_type="web",
description=description
description=description,
)

self._citations.add_source(source)
Expand All @@ -150,7 +150,7 @@ def add_attachment_source(self, filename: str, description: str | None = None) -
filename=filename,
date_accessed=datetime.now(timezone.utc).strftime("%Y-%m-%d"),
source_type="attachment",
description=description or "processed attachment"
description=description or "processed attachment",
)

self._citations.add_source(source)
Expand All @@ -170,7 +170,7 @@ def add_api_source(self, title: str, description: str | None = None) -> str:
title=sanitized_title,
date_accessed=datetime.now(timezone.utc).strftime("%Y-%m-%d"),
source_type="api",
description=description or "API data"
description=description or "API data",
)

self._citations.add_source(source)
Expand All @@ -179,8 +179,7 @@ def add_api_source(self, title: str, description: str | None = None) -> str:
def get_citations(self) -> CitationCollection:
"""Get a copy of the current citations."""
return CitationCollection(
sources=self._citations.sources.copy(),
references_section=self._citations.references_section
sources=self._citations.sources.copy(), references_section=self._citations.references_section
)

def generate_references_section(self) -> str:
Expand All @@ -189,12 +188,7 @@ def generate_references_section(self) -> str:
return ""

# Categorize sources
source_categories = {
"visited": [],
"search": [],
"attachment": [],
"api": []
}
source_categories = {"visited": [], "search": [], "attachment": [], "api": []}

for source in self._citations.sources:
if source.source_type == "web":
Expand All @@ -213,7 +207,7 @@ def generate_references_section(self) -> str:
("visited", "#### Visited Pages", lambda s: f"{s.id}. [{s.title}]({s.url})"),
("search", "#### Search Results", lambda s: f"{s.id}. [{s.title}]({s.url})"),
("attachment", "#### Attachments", lambda s: f"{s.id}. {s.filename}"),
("api", "#### Data Sources", lambda s: f"{s.id}. {s.title}")
("api", "#### Data Sources", lambda s: f"{s.id}. {s.title}"),
]

for category, header, formatter in section_configs:
Expand Down Expand Up @@ -265,10 +259,7 @@ def _load_attachments(self, attachment_info: list[dict]):
"""Load attachments from disk into memory store."""
for info in attachment_info:
success = self.attachment_service.load_attachment(
filename=info["filename"],
file_path=info["path"],
content_type=info["type"],
size=info["size"]
filename=info["filename"], file_path=info["path"], content_type=info["type"], size=info["size"]
)
if success:
logger.debug(f"Successfully loaded attachment: {info['filename']}")
Expand Down
152 changes: 0 additions & 152 deletions mxtoai/scripts/email_processor.py
View file
Open in desktop

This file was deleted.

Loading
Loading