ARRG-自动研究报告生成器
一个多智能体系统，用于使用专门的人工智能代理生成综合研究报告，这些代理仅通过 A2A（代理到代理）协议v1.0 并使用 MCP（模型上下文协议）2025-11-25 所有工具调用的规范。
状态
ARRG目前进展顺利，但仍在进行中。到目前为止，我只测试了Tetrate 提供者，使用 claude-haiku-4-5示例报告见 example_reports 目录。
概述
ARRG使用五个专业代理共同制作高质量的研究报告：
- 规划代理：创建具有大纲和方法的结构化研究计划
- 研究代理：根据研究问题收集信息和来源（使用MCP工具）
- 分析代理：将研究数据综合为见解和发现
- 写作代理：将分析转化为精美的专业报告
- QA代理：审查和验证报告的质量和准确性
协议架构
ARRG使用两个互补的协议，将关注点清晰地分开：
A2A协议v1.0——代理间通信
所有代理之间的通信都使用 A2A协议 独家：
- 代理卡 --每个代理都通过
AgentCard具有技能、支持的输入/输出模式和提供者元数据（根据A2A发现规范） - 任务 --代理人之间交换的工作单位。任务遵循A2A生命周期状态机：
submitted → working → completed/failed（附加说明：canceled,input_required,rejected,auth_required) - 消息 --任务内的沟通使用
Message带有a的对象role（用户/代理）并键入Parts:
- TextPart --纯文本内容 - DataPart --结构化JSON数据（研究计划、分析结果等） - FilePart --MIME类型的二进制文件内容
- 人工制品 --已完成任务附带的可交付成果（报告、质量保证结果等）
- 任务状态 --使用时间戳和状态消息跟踪状态转换
编排器为工作流的每个阶段创建A2A任务，并通过以下方式向代理发送消息 process_task()代理处理任务，转换状态，并返回已完成的任务和工件。
MCP 2025-11-25-工具调用
所有工具调用都遵循 MCP规范:
- 工具发现 --工具已在
MCPToolRegistry使用JSON模式输入定义 - 工具调用 --LLM发起的工具调用通过以下方式执行
MCPToolCall/MCPToolResult - 内置工具 —
web_search用于互联网研究，可扩展到更多工具 - JSON-RPC 2.0 --MCP通信使用标准JSON-RPC通过stdio传输
为什么是两个协议？ A2A手柄 *代理人对代理人* 通信（任务委托、结果、工件），而MCP处理 *代理到工具* 通信（搜索、文件I/O、API）。它们是互补的——代理通过A2A接收工作，并使用MCP工具来完成它。
建筑
┌─────────────────────────────────────────────────────┐
│ Orchestrator │
│ (creates A2A Tasks, routes Messages) │
└─────────┬───────┬───────┬───────┬───────┬───────────┘
│ A2A │ A2A │ A2A │ A2A │ A2A
▼ ▼ ▼ ▼ ▼
┌─────────┐ ┌─────┐ ┌────────┐ ┌───────┐ ┌────┐
│Planning │ │Rsrch│ │Analysis│ │Writing│ │ QA │
│ Agent │ │Agent│ │ Agent │ │ Agent │ │Agnt│
└─────────┘ └──┬──┘ └────────┘ └───────┘ └────┘
│ MCP
▼
┌─────────┐
│MCP Tools│
│(search) │
└─────────┘工作流程
- 规划 --编排器通过A2A任务发送主题→ 规划代理返回研究计划
- 研究 --通过A2A任务发送的研究问题→ 研究代理使用MCP
web_searchtool → 返回调查结果 - 分析 --通过A2A任务发送的研究数据→ Analysis Agent返回综合见解
- 写作 --通过A2A任务发送分析→ 写入代理返回格式化的报告
- 问答 --通过A2A任务发送的报告→ QA代理返回质量评估
- 修订循环 --如果QA拒绝，则将报告连同反馈一起发送回Writing Agent（最多重试2次）
A2A任务生命周期
每个工作流阶段都遵循A2A任务状态机：
submitted → working → completed
→ failed
→ input_required (for revision requests)编排器跟踪所有任务及其状态转换，维护完整的消息历史记录以供调试和审计。
A2A数据结构
代理卡
from arrg.a2a import AgentCard, AgentSkill, AgentProvider, AgentCapabilities
card = AgentCard(
name="Research Agent",
description="Gathers information from web sources",
url="local://research-agent",
provider=AgentProvider(organization="ARRG"),
capabilities=AgentCapabilities(streaming=True),
skills=[
AgentSkill(
id="web_research",
name="Web Research",
description="Search and synthesize web sources",
tags=["research", "search"],
)
],
input_modes=["application/json"],
output_modes=["application/json"],
)任务和消息
from arrg.a2a import Task, TaskState, Message, MessageRole, TextPart, DataPart
# Create a task
task = Task()
# Add a user message with text and structured data
message = Message.create_user_message(
text="Research the impact of AI on healthcare",
data={"topic": "AI in healthcare", "questions": ["What are the benefits?"]},
sender="orchestrator",
task_id=task.id,
)
task.add_message(message)
# Update task state
task.update_state(TaskState.WORKING, message="Agent processing...")
task.update_state(TaskState.COMPLETED, message="Research complete")人工制品
from arrg.a2a import Artifact
# Create a data artifact
artifact = Artifact.create_data_artifact(
data={"findings": [...], "sources": [...]},
name="research_results",
description="Research findings for AI in healthcare",
)
task.add_artifact(artifact)安装
# Clone the repository
git clone
cd arrg
# Install in development mode
pip install -e .
# Run the dashboard
python -m arrg dashboard
# Or generate a report from CLI
python -m arrg cli --topic "Your Research Topic" --api-key YOUR_API_KEY用法
流线型仪表板
python -m arrg dashboard
# or
streamlit run arrg/ui/dashboard.py命令行界面
python -m arrg cli --topic "The Impact of AI on Healthcare" --api-key YOUR_KEY --model claude-haiku-4-5程序化
from arrg import Orchestrator
from pathlib import Path
orchestrator = Orchestrator(
api_key="your-api-key",
provider_endpoint="Tetrate",
models={"planning": "claude-haiku-4-5", "research": "claude-haiku-4-5", ...},
workspace_dir=Path("./workspace"),
)
result = orchestrator.generate_report("Your Research Topic")
if result["status"] == "success":
report = result["report"]
print(report["title"])
print(report["full_text"])项目结构
arrg/
├── a2a/ # A2A Protocol v1.0 implementation
│ ├── __init__.py # Package exports
│ ├── agent_card.py # AgentCard, AgentSkill, AgentProvider, AgentCapabilities
│ ├── task.py # Task, TaskState, TaskStatus
│ ├── message.py # Message, MessageRole, TextPart, DataPart, FilePart
│ └── artifact.py # Artifact with typed Parts
├── agents/ # Agent implementations
│ ├── base.py # BaseAgent (abstract) - A2A + MCP integration
│ ├── planning.py # PlanningAgent - research plan generation
│ ├── research.py # ResearchAgent - web research via MCP tools
│ ├── analysis.py # AnalysisAgent - data synthesis
│ ├── writing.py # WritingAgent - report composition + revision
│ └── qa.py # QAAgent - quality validation
├── core/
│ └── orchestrator.py # Workflow orchestrator (A2A task coordination)
├── mcp/ # MCP 2025-11-25 implementation
│ ├── client.py # MCP client (JSON-RPC/stdio)
│ ├── server.py # MCP server
│ ├── schema.py # MCP tool schemas
│ └── tools.py # Built-in tools (web_search)
├── protocol/ # Backward-compatible re-exports from a2a/
│ ├── __init__.py # Re-exports A2A types + SharedWorkspace
│ ├── message.py # Deprecated shim → arrg.a2a
│ └── workspace.py # SharedWorkspace (key-value artifact storage)
├── ui/
│ └── dashboard.py # Streamlit dashboard
├── utils/
│ └── llm_client.py # LLM API client (OpenAI/Anthropic)
├── __init__.py # Package exports
└── __main__.py # CLI entry point关键设计决策
A2A全代理通信协议
代理之间的每次交互都使用A2A协议数据结构：
- 无自定义消息类型 --所有消息使用
Message随着TextPart/DataPart而不是自定义枚举 - 以任务为中心的工作流程 --每个阶段都会产生一个A2A
Task具有适当的状态转换 - 输出工件 --代理交付成果表示为A2A
Artifact物体 - 发现代理卡 --每个代理通过
AgentCard有技能
刀具调用MCP（补充）
MCP专门用于工具调用（网络搜索等），它不处理代理到代理的通信。这种干净的分离遵循了两种方案的预期设计：
- A2A =代理人如何相互交谈
- 主控程序 =代理如何使用工具
大型工件的共享工作区
大数据（研究结果、完整报告）存储在 SharedWorkspace 并且由A2A消息中的键引用。这避免了在消息中直接传递大型有效载荷，同时保持了A2A协议的合规性——工作区密钥作为 DataPart A2A以内 Message.
扩展ARRG
添加新的MCP工具
在中注册新工具 arrg/mcp/tools.py:
from arrg.mcp import MCPTool, MCPToolCall, MCPToolResult, TextContent
def my_tool_executor(call: MCPToolCall) -> MCPToolResult:
"""Execute the tool and return MCP-compliant result."""
result_text = f"Executed with args: {call.arguments}"
return MCPToolResult(
call_id=call.call_id,
content=[TextContent(text=result_text)],
is_error=False,
)
# In MCPToolRegistry._register_builtin_tools():
self.register_tool(
MCPTool(
name="my_tool",
description="Description of my tool",
inputSchema={
"type": "object",
"properties": {
"param1": {"type": "string", "description": "First parameter"},
},
"required": ["param1"],
},
),
executor=my_tool_executor,
)添加新代理
- 创建一个新的代理类，继承自
BaseAgent - 实施
get_capabilities()返回技能描述列表 - 实施
process_task(task: Task) -> Task遵循A2A任务生命周期 - 将代理添加到编排器工作流中
- 创建一个
AgentSkill对于代理提供的每种功能
故障排除
常见问题
API密钥错误：
- 确保在侧栏中正确输入API密钥
- 验证密钥是否可以访问所选型号
型号不可用：
- 检查您的提供商是否支持所选型号
- 从下拉列表中尝试其他型号
工作区错误：
- 确保
./workspace目录可写 - 如果工作区包含损坏的数据，请清除它
导入错误：
- 验证软件包是否已安装：
pip install -e . - 检查Python版本：
python --version（需要3.12+）
MCP工具错误：
- 检查工具注册表初始化：
python -c "from arrg.mcp import get_tool_registry; r = get_tool_registry(); print(r.list_tools())" - 验证工具架构：
python -c "from arrg.mcp import get_tool_registry; r = get_tool_registry(); print([t.name for t in r.list_tools()])"
许可证
\[您的许可证在这里\]
贡献
欢迎投稿！拜托：
- 分叉存储库
- 创建要素分支
- 进行更改
- 提交拉取请求
支持
对于问题和疑问：
- 在GitHub上打开一个问题
- 检查文档
- 审查PRD（产品要求文件）
路线图
未来的增强功能：
- \[\]用于真正网络化A2A通信的HTTP JSON-RPC传输
- \[\]A2A代理卡送达
/.well-known/agent-card.json - \[\]通过A2A进行SSE流式实时任务更新
- \[\]A2A推送通知
- \[\]将模拟工具执行器替换为实际实现（web搜索API、文件I/O等）
- \[\]支持多主题批处理
- \[\]先进的引文和参考文献管理
- \[\]通过MCP客户端连接与外部研究数据库集成
- \[\]自定义代理插件
- \[\]报告模板和样式选项
- \[\]团队工作流程的协作功能
- \[\]用于编程访问的API端点
- \[\]MCP服务器发现和多服务器工具聚合