第 26 章 · 实战二：多源研究助手 Team
本章目标：构建一条"多源检索 → 协作撰写 → 逐条核查"的研究流水线——研究员 A 负责公网搜索、研究员 B 负责本地文档库检索，撰写员汇总成报告，事实核查员对关键论断逐条复核并标注置信度，最终产出带引用来源的
research-report.md。
26.1 需求与团队设计
写一份可信的研究报告，难点不在"写"，而在素材的广度与论断的可信度。我们把这两个关注点拆给不同角色：
text
研究主题："边缘 AI 推理框架 2025 年的技术格局"
│
▼
┌──────────────────────────────────────────────┐
│ 研究 Team（coordinate 模式） │
│ │
│ 领导者：拆解子任务、分派、汇总 │
│ ├── 研究员 A：DuckDuckGo 公网检索 │
│ │ （新闻、博客、官方发布） │
│ └── 研究员 B：本地 Knowledge 检索 │
│ （arXiv 论文摘要库，LanceDb 向量） │
└──────────────────────────────────────────────┘
│ 素材包（带来源标注）
▼
撰写员 Agent ──▶ 报告草稿
│
▼
事实核查 Agent ──▶ 对每个关键论断标注置信度
│
▼
research-report.md（含引用来源列表）为什么用 Team（coordinate）而不是线性 Workflow？
- 两个研究员的检索互不依赖，coordinate 模式下领导者会并发委派，总耗时约等于最慢的那个成员；线性 Workflow 只能串行等待；
- 检索结果的质量不可预知，需要领导者根据素材情况决定是否追加委派（比如"公网结果太少，让 B 再查一次论文库"）——这种动态性正是 coordinate 的强项；
- 后半段"撰写 → 核查"是强顺序依赖，放进 Team 里反而多余，所以拆成独立的两个 Agent 串联执行。
安装依赖：
bash
pip install agno duckduckgo-search "agno[lancedb]" pypdf python-dotenv26.2 准备模型与本地论文知识库
统一使用 DeepSeek 作为底层模型；研究员 B 检索的是预先灌入 LanceDb 的 arXiv 论文摘要库：
python
# research_common.py —— 团队共享的基础设施
import os
from agno.knowledge.knowledge import Knowledge
from agno.knowledge.embedder.openai import OpenAIEmbedder
from agno.models.openai import OpenAIChat
from agno.vectordb.lancedb import LanceDb, SearchType
def make_model() -> OpenAIChat:
"""所有角色共用同一个三方模型配置，便于统一换模型。"""
return OpenAIChat(
id="deepseek-chat",
api_key=os.getenv("DEEPSEEK_API_KEY"),
base_url="https://api.deepseek.com/v1",
temperature=0.3, # 研究场景要求克制，降低发散
)
# 研究员 B 的弹药库：arXiv 论文摘要向量库
paper_kb = Knowledge(
vector_db=LanceDb(
table_name="arxiv_abstracts",
uri="tmp/lancedb",
search_type=SearchType.vector,
embedder=OpenAIEmbedder(
id="Qwen/Qwen3-Embedding-8B",
api_key=os.getenv("EMBEDDING_API_KEY"),
base_url="https://api.siliconflow.cn/v1",
),
)
)
# 首次运行时灌入本地论文 PDF 目录（skip_if_exists 保证幂等）
# paper_kb.insert(path="docs/papers/", skip_if_exists=True)素材从哪来
insert(path="docs/papers/") 会自动识别目录下的 PDF 并切块入库。没有现成论文也没关系——把任意领域资料放进去即可，本章演示的是检索能力而非数据本身。
26.3 两名研究员：工具互补
研究员 A 用 DuckDuckGo 工具查公网；研究员 B 挂载 Knowledge 由模型自主决定何时检索。两者输出格式统一约定为「发现 + 来源」，方便后续引用：
python
# researchers.py —— 互不重叠的两路检索
from agno.agent import Agent
from agno.tools.duckduckgo import DuckDuckGoTools
from research_common import make_model, paper_kb
researcher_web = Agent(
id="researcher-web",
name="研究员A-公网",
role="通过搜索引擎收集最新动态、产品发布与行业新闻。",
model=make_model(),
tools=[DuckDuckGoTools(news=True, fixed_max_results=8)],
instructions=[
"围绕主题做 2~3 组不同关键词的搜索，覆盖面优先于深度。",
"每条发现必须附带来源 URL 与大致日期。",
"区分'官方发布'与'自媒体观点'，分别标注。",
"最终输出编号列表，每行格式：[发现] —— [来源URL] —— [日期]",
],
markdown=True,
)
researcher_local = Agent(
id="researcher-local",
name="研究员B-文献库",
role="从本地论文知识库中检索学术观点与技术对比。",
model=make_model(),
knowledge=paper_kb,
search_knowledge=True, # 模型自行决定何时触发检索
instructions=[
"至少执行 2 次不同角度的知识库检索（如按技术路线、按性能指标）。",
"引用论文时注明论文名称或文件名作为来源。",
"知识库中没有的内容明确说明，不要编造论点。",
"最终输出编号列表，每行格式：[发现] —— [论文/来源名]",
],
markdown=True,
)注意两名成员的 instructions 都强制了结构化输出——这是后面核查环节能逐条定位论断的前提。
26.4 研究 Team：coordinate 模式组装
python
# research_team.py
from agno.team import Team, TeamMode
from researchers import researcher_web, researcher_local
from research_common import make_model
research_team = Team(
id="research-team",
name="研究小组",
mode=TeamMode.coordinate,
model=make_model(),
members=[researcher_web, researcher_local],
instructions=[
"收到主题后，先拆解成'公网动态'和'文献观点'两类子任务。",
"将子任务并行委派给对应研究员，等待双方返回。",
"若某一方素材少于 3 条，追加一轮更具体的关键词委派。",
"汇总时去除重复信息，合并同类观点，形成结构化素材包。",
"素材包中保留每条信息的原始来源标注。",
],
markdown=True,
)
if __name__ == "__main__":
research_team.print_response(
"边缘 AI 推理框架 2025 年的技术格局：主流方案、性能对比与趋势",
stream_intermediate_steps=True, # 打印各成员的中间过程，便于观察协作
)stream_intermediate_steps=True 会把领导者的分派决策和每个成员的产出流式打印出来——第一次跑这条流水线时强烈建议打开，你能直观看到 coordinate 模式的委派过程。
26.5 撰写员与事实核查员
撰写员只吃素材包写报告；核查员拿到的则是报告草稿，逐条复核后输出修订版：
python
# writer.py —— 撰写员：只吃素材包写报告
from agno.agent import Agent
from research_common import make_model
writer = Agent(
id="report-writer",
name="撰写员",
role="基于研究素材包撰写结构化研究报告。",
model=make_model(),
instructions=[
"报告结构：执行摘要 / 技术格局综述 / 关键对比 / 趋势研判。",
"每个关键论断后用 [S1][S2]… 编号标记其支撑来源。",
"对立观点要如实呈现，不得为行文流畅而抹平分歧。",
"文末附《引用来源》章节，列出全部编号对应的 URL 或论文名。",
],
markdown=True,
)核查员拿到的则是报告草稿，逐条复核后输出修订版：
python
# checker.py —— 事实核查员：逐条复核并标注置信度
from agno.agent import Agent
from research_common import make_model
fact_checker = Agent(
id="fact-checker",
name="事实核查员",
role="对报告中的关键论断逐条复核，标注置信度。",
model=make_model(),
instructions=[
"提取报告中所有带 [Sn] 编号的论断，逐一检查："
"论断表述是否超出素材支持范围？数字/时间/名称是否与素材一致？",
"对每条论断输出：✅ 支持 / ⚠️ 部分支持 / ❌ 缺乏依据，并附一句理由。",
"整体给出高/中/低三档置信度结论。",
"直接输出修订后的完整报告：无依据的论断删除或改写为'待验证'，"
"并在文末追加《核查记录》附录。",
],
markdown=True,
)核查员的输出规范（✅/⚠️/❌ 三态）比笼统的"已核实"更有工程价值——低置信度论断在交付前可以被单独处理。
26.6 串起整条流水线并落盘报告
python
# run_research.py —— 入口脚本
import datetime
from pathlib import Path
from research_team import research_team
from writer_and_checker import writer, fact_checker
TOPIC = "边缘 AI 推理框架 2025 年的技术格局"
# 第一棒：Team 多源检索，得到素材包
material_pack = research_team.run(f"研究主题：{TOPIC}").content
# 第二棒：撰写草稿
draft = writer.run(
f"研究主题：{TOPIC}\n\n以下是研究素材包：\n{material_pack}"
).content
# 第三棒：逐条核查并修订
final = fact_checker.run(
f"请核查并修订以下研究报告草稿：\n\n{draft}"
).content
# 落盘：带日期归档，便于追溯历史版本
out_dir = Path("reports"); out_dir.mkdir(exist_ok=True)
today = datetime.date.today().isoformat()
report_path = out_dir / f"research-report-{today}.md"
report_path.write_text(final, encoding="utf-8")
print(f"报告已生成: {report_path}")跑完后打开 reports/research-report-<日期>.md：正文里的 [S1][S2] 编号能在《引用来源》里找到出处，《核查记录》附录则告诉你哪些论断被降级成了"待验证"。
成本提示
这条流水线全程使用 deepseek-chat，单次完整运行的 token 消耗主要来自 Team 的多轮协调与长素材注入。若预算敏感，可把核查员换成更小的模型——核对类任务对推理能力的要求低于撰写。
本章小结
- Team 选型：互不依赖的多路检索选
TeamMode.coordinate（可并发委派、可动态补查）；强顺序依赖的后段用独立 Agent 串联； - 结构化中间产物：研究员输出统一的「发现 + 来源」格式，是下游撰写与核查能自动衔接的关键；
- 职责分离：撰写员只管表达，核查员只管验证——同一份草稿的两个视角比一个 Agent 自查可靠得多；
- 置信度分级：✅/⚠️/❌ 三态核查记录让报告的交付质量可以被量化管理。
🧪 随堂测验
点击你认为正确的选项。答错时会展示正确答案与原因解析。
1. 为什么两名研究员的检索任务适合用 Team（coordinate）而不是线性 Workflow 串联？
2. 研究员的 instructions 强制输出「[发现] —— [来源]」格式的最主要目的是？
3. 事实核查员对每条论断输出 ✅/⚠️/❌ 三态结果，相比只回答"已核实"的优势是？
4. 关于本流水线的成本优化，下列做法最合理的是？
🛠️ 动手实践
- 给研究员 A 增加 RSS 订阅源读取工具（如
agno.tools.rss.RssTools），对比加入后报告的时效性变化。 - 把核查员的输出接入一个新 Agent"修订执行者"，让它只负责按核查意见改写报告，实现"核查"与"修订"彻底分离。
- 为流水线增加成本统计：分别在三个阶段前后打印模型用量，找出 token 消耗最大的环节并尝试优化。
下一章我们把视角从"研究"切回"服务"：把客服知识库 Agent 真正推上生产线——第 27 章：客服知识库 Agent 上线 AgentOS。