Capstone：构建一个研究与报告智能体
构建一个会搜索、综合并产出结构化报告的完整智能体。
- 理解并应用本课涵盖的核心概念
你将构建什么
这个 capstone 把 Agentic AI 课程里的一切——多智能体编排、工具使用、记忆、安全与评估——汇聚成一个连贯的项目。
研究与报告智能体以一个研究问题作为输入，产出一个带来源引用的结构化 markdown 报告。它是研究者、分析师或技术写作者真正会用的那种工具。
输入：「RAG 与微调（fine-tuning）在企业 LLM 应用中的权衡是什么？」
输出： 一份结构化报告，含执行摘要、详细分析章节、权衡表、建议与引用来源——在 2 分钟内产出。
系统架构
系统依次使用四个专门智能体：
User Question
|
v
[Planner Agent]
Breaks the question into 4-6 specific sub-questions
|
v
[Research Agent]
Searches the web for each sub-question (can run in parallel)
|
v
[Synthesis Agent]
Combines all research findings into coherent prose
|
v
[Formatter Agent]
Structures the content into a polished markdown report
|
v
Final Report (with sources, sections, and executive summary)
每个智能体只有单一职责。这让系统比单一庞然大物般的智能体更易调试、改进与扩展。
第一部分：Planner 智能体
Planner 的职责是把一个宽泛的研究问题分解成具体的、可搜索的子问题。这一点很关键，因为「RAG 与微调的权衡是什么」对一个单一搜索来说太宽泛——你会得到散乱的结果。把它拆成子问题，能产出聚焦、高质量的研究。
import anthropic
import json
client = anthropic.Anthropic()
PLANNER_SYSTEM = """You are a research planning specialist. Your job is to decompose
a broad research question into 4-6 specific sub-questions that together cover the topic completely.
Each sub-question should:
- Be specific enough to search for directly
- Cover a distinct aspect of the main question
- Together provide comprehensive coverage of the topic
Return a JSON object with:
{
"main_question": "the original question",
"sub_questions": ["question 1", "question 2", ...],
"key_concepts": ["concept 1", "concept 2", ...],
"report_sections": ["suggested section titles for the final report"]
}"""
def run_planner(research_question: str) -> dict:
"""Break a research question into a research plan."""
response = client.messages.create(
model="claude-opus-4-5",
max_tokens=1024,
system=PLANNER_SYSTEM,
messages=[{
"role": "user",
"content": f"Create a research plan for: {research_question}"
}]
)
try:
# Extract JSON from response
text = response.content[0].text
# Find JSON block
start = text.find('{')
end = text.rfind('}') + 1
return json.loads(text[start:end])
except (json.JSONDecodeError, ValueError):
# Fallback if JSON parsing fails
return {
"main_question": research_question,
"sub_questions": [research_question],
"key_concepts": [],
"report_sections": ["Overview", "Analysis", "Conclusion"]
}
# Test the planner
plan = run_planner("What are the trade-offs between RAG and fine-tuning for enterprise LLM applications?")
print(json.dumps(plan, indent=2))
对我们的测试问题，planner 应当产出类似这样的子问题：
- 什么是 RAG（检索增强生成），它如何工作？
- 什么是 LLM 微调，何时使用它？
- RAG 与微调在成本与基础设施要求上各是什么？
- RAG 与微调在知识新鲜度与准确性上如何比较？
- 延迟与可扩展性上的权衡是什么？
- 企业通常选哪种方案，为什么？
第二部分：Research 智能体
Research 智能体接收一个单一子问题，并找到相关、准确的信息。对每个子问题，它执行 1-3 次定向搜索并返回结构化的发现。
import requests
import time
# Simple web search using DuckDuckGo's instant answer API
def web_search(query: str, max_results: int = 3) -> list[dict]:
"""Search the web and return clean results."""
try:
# Using DuckDuckGo HTML scraping (for production, use Serper or Tavily API)
headers = {"User-Agent": "Mozilla/5.0 (research bot)"}
params = {"q": query, "format": "json"}
response = requests.get(
"https://api.duckduckgo.com/",
params=params,
headers=headers,
timeout=10
)
data = response.json()
results = []
# Abstract (main result)
if data.get("Abstract"):
results.append({
"source": data.get("AbstractURL", "DuckDuckGo"),
"title": data.get("Heading", query),
"content": data["Abstract"]
})
# Related topics
for topic in data.get("RelatedTopics", [])[:max_results - 1]:
if isinstance(topic, dict) and topic.get("Text"):
results.append({
"source": topic.get("FirstURL", ""),
"title": topic.get("Text", "")[:50],
"content": topic.get("Text", "")
})
return results if results else [{"source": "", "title": query, "content": "No results found"}]
except Exception as e:
return [{"source": "", "title": "Search failed", "content": str(e)}]
RESEARCH_TOOLS = [
{
"name": "search_web",
"description": """Search the web for information about a specific topic.
Use for current information, technical comparisons, and expert opinions.
Returns: list of results with source URL, title, and content snippet.""",
"input_schema": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "Specific search query"}
},
"required": ["query"]
}
}
]
RESEARCH_SYSTEM = """You are a research analyst. Given a specific research question,
search for accurate and relevant information.
Your output should be structured research notes containing:
- Key facts and data points (with sources)
- Expert opinions or consensus views
- Any important nuances or contradictions found
- 2-4 credible sources
Be factual and specific. Avoid vague generalities."""
def run_research_agent(sub_question: str) -> dict:
"""Research a single sub-question. Returns findings dict."""
messages = [{"role": "user", "content": f"Research this question: {sub_question}"}]
sources = []
for _ in range(5):
response = client.messages.create(
model="claude-opus-4-5",
max_tokens=1024,
system=RESEARCH_SYSTEM,
tools=RESEARCH_TOOLS,
messages=messages
)
if response.stop_reason == "end_turn":
return {
"question": sub_question,
"findings": response.content[0].text,
"sources": sources
}
if response.stop_reason == "tool_use":
messages.append({"role": "assistant", "content": response.content})
tool_results = []
for block in response.content:
if block.type == "tool_use":
results = web_search(block.input["query"])
# Collect sources for citation
sources.extend([r["source"] for r in results if r.get("source")])
formatted = "\n\n".join([
f"Source: {r['source']}\n{r['content']}"
for r in results
])
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": formatted
})
messages.append({"role": "user", "content": tool_results})
return {"question": sub_question, "findings": "Research incomplete", "sources": sources}
第三部分：并行运行研究
对我们 planner 生成的六个子问题，我们可以并行运行研究以节省时间：
import concurrent.futures
from typing import List
def research_all_questions(sub_questions: List[str]) -> List[dict]:
"""Run research agents for all sub-questions in parallel."""
print(f"Researching {len(sub_questions)} sub-questions in parallel...")
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
# Submit all research tasks
future_to_question = {
executor.submit(run_research_agent, q): q
for q in sub_questions
}
results = []
for future in concurrent.futures.as_completed(future_to_question):
question = future_to_question[future]
try:
result = future.result(timeout=60)
results.append(result)
print(f" Completed: {question[:60]}...")
except Exception as e:
print(f" Failed: {question[:60]}... ({e})")
results.append({
"question": question,
"findings": f"Research failed: {str(e)}",
"sources": []
})
return results
使用 3 个并行 worker 在速度与限流之间取得平衡。有 6 个子问题时，这能把总研究时间从约 90 秒（顺序）降到约 30 秒（并行）。
第四部分：Synthesis 智能体
Synthesis 智能体接收所有研究发现，并把它们编织成连贯的散文。它的挑战是避免重复，并让行文自然流畅。
SYNTHESIS_SYSTEM = """You are an expert technical writer and analyst. You receive research
findings from multiple sources and synthesize them into clear, coherent prose.
Guidelines:
- Integrate findings across sub-questions into flowing paragraphs, not bullet dumps
- Identify and highlight areas of consensus and disagreement
- Use specific facts and data points from the research
- Write for a senior technical audience — no hand-holding, no padding
- Maintain a neutral, analytical tone
- Flag anywhere the research found contradictions or uncertainty"""
def run_synthesis_agent(plan: dict, research_results: List[dict]) -> str:
"""Synthesize all research findings into coherent analysis."""
# Format research for the synthesis agent
research_text = ""
all_sources = []
for result in research_results:
research_text += f"\n\n### Sub-question: {result['question']}\n"
research_text += result['findings']
all_sources.extend(result.get('sources', []))
# Deduplicate sources
unique_sources = list(dict.fromkeys(s for s in all_sources if s))
prompt = f"""Main research question: {plan['main_question']}
Suggested report sections: {', '.join(plan.get('report_sections', []))}
Research findings:
{research_text}
Write a comprehensive synthesis that covers all aspects of the research.
This synthesis will be used as the basis for the final report."""
response = client.messages.create(
model="claude-opus-4-5",
max_tokens=3000,
system=SYNTHESIS_SYSTEM,
messages=[{"role": "user", "content": prompt}]
)
return response.content[0].text, unique_sources
第五部分：Formatter 智能体
Formatter 智能体接收综合结果，并产出一份带恰当结构、执行摘要与格式化引用的精美 markdown 报告。
FORMATTER_SYSTEM = """You are a report formatter. Convert research synthesis into a
polished, professional markdown report.
Required report structure:
# [Title]
## Executive Summary
(2-3 paragraphs, standalone summary of key findings and recommendation)
## [Section 1 Title]
(prose content)
## [Section 2 Title]
(prose content)
... (additional sections)
## Recommendation
(clear, actionable recommendation based on the research)
## Sources
(formatted citation list)
---
*Report generated by Research Agent | [Date]*
Keep all technical content from the synthesis. The formatting should make the
content more readable, not change its substance."""
def run_formatter_agent(synthesis: str, sources: List[str], research_question: str) -> str:
"""Format synthesis into a polished markdown report."""
# Format sources as a numbered list
source_list = "\n".join([
f"{i+1}. {source}"
for i, source in enumerate(sources[:10]) # Limit to 10 sources
]) if sources else "Web research (sources available on request)"
prompt = f"""Research question: {research_question}
Synthesis:
{synthesis}
Available sources:
{source_list}
Format this into a polished markdown report following the required structure."""
response = client.messages.create(
model="claude-opus-4-5",
max_tokens=4096,
system=FORMATTER_SYSTEM,
messages=[{"role": "user", "content": prompt}]
)
return response.content[0].text
第六部分：编排器（Orchestrator）
现在把所有东西串起来：
from datetime import datetime
def run_research_report_agent(research_question: str) -> dict:
"""
Full pipeline: Plan → Research → Synthesize → Format → Report
Returns the final report and metadata.
"""
start_time = datetime.now()
print(f"\n{'='*70}")
print(f"Research Report Agent")
print(f"Question: {research_question}")
print(f"{'='*70}\n")
# Step 1: Plan
print("[1/4] Planning research structure...")
plan = run_planner(research_question)
print(f"Generated {len(plan['sub_questions'])} sub-questions")
for i, q in enumerate(plan['sub_questions'], 1):
print(f" {i}. {q}")
# Step 2: Research (parallel)
print(f"\n[2/4] Researching {len(plan['sub_questions'])} sub-questions...")
research_results = research_all_questions(plan['sub_questions'])
successful = sum(1 for r in research_results if "failed" not in r['findings'].lower())
print(f"Research complete: {successful}/{len(research_results)} successful")
# Step 3: Synthesize
print("\n[3/4] Synthesizing findings...")
synthesis, sources = run_synthesis_agent(plan, research_results)
print(f"Synthesis complete ({len(synthesis)} chars)")
# Step 4: Format
print("\n[4/4] Formatting final report...")
report = run_formatter_agent(synthesis, sources, research_question)
elapsed = (datetime.now() - start_time).total_seconds()
print(f"\nReport complete in {elapsed:.1f} seconds")
print(f"Report length: {len(report)} characters")
return {
"question": research_question,
"report": report,
"plan": plan,
"sources": sources,
"elapsed_seconds": elapsed,
"timestamp": start_time.isoformat()
}
第七部分：运行这个 Capstone
if __name__ == "__main__":
result = run_research_report_agent(
"What are the trade-offs between RAG and fine-tuning for enterprise LLM applications?"
)
# Save the report
with open("rag_vs_finetune_report.md", "w") as f:
f.write(result["report"])
print("\n" + "="*70)
print("FINAL REPORT")
print("="*70)
print(result["report"])
预期的输出结构
在我们的测试问题上运行，会产出具备如下结构的报告：
# RAG vs Fine-Tuning for Enterprise LLM Applications: A Comparative Analysis
## Executive Summary
Enterprise organizations adopting LLMs face a fundamental architectural choice...
[2-3 paragraphs covering the key trade-offs and bottom-line recommendation]
## Understanding RAG and Fine-Tuning
RAG (Retrieval-Augmented Generation) augments LLM responses by retrieving...
Fine-tuning modifies the model's weights through additional training on...
## Cost and Infrastructure Requirements
RAG typically requires a vector database (Pinecone, Weaviate, ChromaDB)...
Fine-tuning costs are front-loaded: a full fine-tune of a 7B parameter model...
## Knowledge Freshness and Accuracy
RAG's key advantage is knowledge currency — the retrieval index can be updated...
Fine-tuned models bake knowledge into weights, creating a staleness problem...
## Latency and Scalability
RAG adds retrieval latency (50-200ms typical) to each inference call...
Fine-tuned models have no retrieval overhead but require serving a custom model...
## Enterprise Adoption Patterns
Based on available data, enterprises with dynamic knowledge bases prefer RAG...
## Recommendation
For most enterprise use cases, start with RAG. It has lower upfront cost...
Consider fine-tuning when: you need consistent output format/style, the task...
## Sources
1. https://arxiv.org/abs/2312.10997
2. https://huggingface.co/blog/rag-vs-fine-tuning
...
---
*Report generated by Research Agent | 2026-06-06*
扩展系统
一旦基础管道能跑起来，这些扩展会显著增加价值：
添加引用校验： 在包含一个来源前，抓取并校验 URL 是活的：
def verify_source(url: str) -> bool:
try:
response = requests.head(url, timeout=5)
return response.status_code == 200
except:
return False
添加批判步骤： 在格式化前，跑一个「critic 智能体」，识别缺口或薄弱声明：
CRITIC_SYSTEM = """Review this research synthesis. Identify: (1) claims that lack evidence,
(2) important aspects of the topic that weren't covered, (3) any logical inconsistencies."""
添加导出格式： 除了 markdown，formatter 还能产出 PDF、HTML 或 Notion 格式。
添加评估： 用第 6 课（Lesson 6）的评估框架自动给每份报告打分。
你构建内容的总结
这个 capstone 项目组装了课程里的每个概念：
- 多智能体架构（Planner、Research、Synthesis、Formatter）
- 工具使用（把网页搜索包装成智能体工具）
- 并行执行（并发研究子问题）
- 真实 API 集成（带错误处理的 DuckDuckGo 搜索）
- 安全（来源限制、输出大小限制、超时处理）
- 结构化输出（formatter 产出一致的 markdown）
结果是一个系统，能在约 60-90 秒内把一个研究问题变成一份精美、带引用的报告——而人类研究者要花 2-4 小时。
这个架构可自然扩展：添加更多研究工具（arXiv、新闻 API、内部数据库）、在不触碰其他智能体的前提下改进任一智能体的提示、在综合步骤为超长报告加并行、或用为该任务微调过的专门模型替换任一组件。
生产级智能体系统就是这样构建的：从一个清晰的管道开始，把每个阶段实现为一个聚焦的智能体，端到端测试管道，然后基于质量最弱处逐步改进组件。
#AI#tutorial#superml#agent#multi-agent#capstone#report