AgentMask / src /agents /summarizer_agent.py
b2230765034
stage1: multi-agent basic implementation + tests
c34b33b
"""
Summarizer Agent Module
========================
Agent responsible for summarizing documents and content.
"""
from dataclasses import dataclass, field
from typing import Any
from .base import BaseAgent
@dataclass
class SummarizerAgent(BaseAgent):
"""
Agent that summarizes documents and content.
Takes a list of documents/text and produces a concise summary.
"""
role: str = "summarizer"
tools: list[str] = field(default_factory=lambda: ["text_analysis", "summarization"])
async def run(self, input: dict[str, Any]) -> dict[str, Any]:
"""
Summarize the provided documents.
Args:
input: Dictionary with 'documents' key containing list of documents to summarize
Returns:
Dictionary with summary
"""
documents = input.get("documents", [])
self.log(f"Summarizing {len(documents)} documents")
summary = self._generate_summary(documents)
return {
"agent": "summarizer",
"summary": summary,
"document_count": len(documents)
}
def _generate_summary(self, documents: list[dict | str]) -> str:
"""
Generate a summary from the provided documents.
Args:
documents: List of documents (can be dicts with 'snippet'/'text' or strings)
Returns:
Summary string
"""
if not documents:
return "No documents provided for summarization."
# Extract text content from documents
texts = []
for doc in documents:
if isinstance(doc, dict):
# Try common text fields
text = doc.get("snippet") or doc.get("text") or doc.get("content") or ""
if doc.get("title"):
text = f"{doc['title']}: {text}"
texts.append(text)
elif isinstance(doc, str):
texts.append(doc)
# Generate summary (simplified - in production, use LLM)
combined = " ".join(texts)
# Create a simulated intelligent summary
if len(combined) > 200:
key_sentences = combined[:200] + "..."
else:
key_sentences = combined
summary = (
f"Summary of {len(documents)} documents: "
f"{key_sentences} "
f"[Analysis based on {len(combined)} characters of source text]"
)
return summary