17 - 综合实战:从零构建一个 AI 研究助手
一句话总结:把前 16 篇学到的所有概念(Agent、Tools、Knowledge、Memory、Structured Output、Team)串成一个完整项目,从零搭出一个真正能用的 AI 研究助手。
我们要做什么
这是整个系列的最后一篇,也是毕业设计。我们要把之前学的所有东西组装到一起,构建一个完整的 AI 研究助手。它能做这些事:
- 搜索网络 – 用 DuckDuckGo 工具获取最新信息(第 2 篇:Tools)
- 知识库检索 – 存储和搜索已有的研究资料(第 7 篇:Knowledge/RAG)
- 记住用户偏好 – 跨会话记住你关心什么领域(第 6 篇:Memory)
- 结构化输出 – 生成格式统一的研究报告(第 3 篇:Structured Output)
- 多 Agent 协作 – 研究员搜集信息、分析师提炼洞察(第 10 篇:Team)
如果你是从第 1 篇跟过来的,到这里你已经掌握了 Agno 的核心能力。这一篇只是把积木搭到一起,没有新概念,全是你见过的老朋友。
安装依赖
pip install agno openai duckduckgo-search chromadb
第一步:定义输出结构
研究助手最终要产出一份研究报告。用 Pydantic 把报告格式定义清楚,Agent 就会严格按这个结构返回数据,拿到手直接能用。
from typing import List
from pydantic import BaseModel, Field
class ResearchReport(BaseModel):
topic: str = Field(..., description="研究主题")
summary: str = Field(..., description="核心摘要,3-5句话")
key_findings: List[str] = Field(..., description="关键发现,3-5条")
sources: List[str] = Field(..., description="信息来源")
recommendation: str = Field(..., description="建议和下一步行动")
confidence: float = Field(..., ge=0, le=1, description="结论可信度 0-1")
每个字段的 description 就是写给 LLM 的说明书。描述越精确,LLM 填得越靠谱。这个我们在第 3 篇已经聊过了。
第二步:搭建基础设施
一个正经的应用需要数据库、知识库和记忆系统。这些都是前面单独学过的,现在组装到一起。
from agno.db.sqlite import SqliteDb
from agno.knowledge.knowledge import Knowledge
from agno.knowledge.embedder.openai import OpenAIEmbedder
from agno.vectordb.chroma import ChromaDb
from agno.vectordb.search import SearchType
from agno.memory import MemoryManager
from agno.models.openai import OpenAIChat
# 共用的数据库和模型
db = SqliteDb(db_file="tmp/research_assistant.db")
model = OpenAIChat(id="gpt-4o-mini")
# 知识库:存储和检索研究资料
knowledge = Knowledge(
name="research_kb",
vector_db=ChromaDb(
name="research",
collection="research",
path="tmp/chromadb",
persistent_client=True,
search_type=SearchType.hybrid,
embedder=OpenAIEmbedder(id="text-embedding-3-small"),
),
max_results=5,
contents_db=db,
)
# 记忆管理器:记住用户的研究偏好
memory_manager = MemoryManager(
model=model,
db=db,
additional_instructions="记录用户的研究兴趣领域和偏好的报告风格",
)
三个核心组件:
- SqliteDb – 存对话历史、会话状态、用户记忆。生产环境换 PostgreSQL,接口一样
- Knowledge + ChromaDb – 向量知识库,用 hybrid 搜索(关键词 + 语义混合),检索效果最好
- MemoryManager – 从对话中提取用户偏好,跨会话记住”这个用户关心什么”
第三步:创建专精 Agent
一个研究员负责搜信息,一个分析师负责提炼洞察。专人专事,各司其职。
from agno.agent import Agent
from agno.tools.duckduckgo import DuckDuckGoTools
# Agent 1:网络研究员
researcher = Agent(
name="网络研究员",
role="搜索网络获取最新信息",
model=model,
tools=[DuckDuckGoTools()],
instructions="""\
你是一个网络研究员。
1. 搜索与主题相关的最新信息
2. 收集多个来源的数据
3. 整理成结构化的研究笔记
只收集信息,不做分析判断。""",
)
# Agent 2:知识分析师
analyst = Agent(
name="知识分析师",
role="分析信息并提取洞察",
model=model,
knowledge=knowledge,
search_knowledge=True,
instructions="""\
你是一个知识分析师。
1. 分析研究员提供的信息
2. 结合知识库中的已有知识
3. 提取关键洞察和趋势
4. 评估信息的可信度
客观分析,用数据说话。""",
)
几个关键设计决策:
- 研究员有 tools(DuckDuckGo),分析师有 knowledge – 能力互补,不重叠
- 每个 Agent 的 instructions 明确职责边界 – 研究员”只收集不分析”,分析师”客观分析”
- role 很重要 – Team 的领导者根据 role 来决定把什么任务分给谁
第四步:组建研究团队
把两个 Agent 放进 Team,加上输出结构、记忆和存储。
from agno.team.team import Team
research_team = Team(
name="AI研究助手",
model=model,
members=[researcher, analyst],
instructions="""\
你是研究团队的负责人。
1. 先让研究员搜索相关信息
2. 再让分析师进行深度分析
3. 综合两者的结果,生成研究报告
报告要客观、有数据支撑。""",
output_schema=ResearchReport,
db=db,
memory_manager=memory_manager,
enable_agentic_memory=True,
add_history_to_context=True,
num_history_runs=3,
markdown=True,
)
这个 Team 身上集成了我们学过的大部分概念:
| 配置 | 对应的知识点 | 作用 |
|---|---|---|
members |
第 10 篇 Team | 成员列表,领导者负责分配任务 |
output_schema |
第 3 篇 Structured Output | 强制输出 ResearchReport 格式 |
db |
第 5 篇 Storage | 持久化对话历史 |
memory_manager |
第 6 篇 Memory | 跨会话记住用户偏好 |
enable_agentic_memory |
第 6 篇 Memory | 让 Team 自己决定何时存记忆 |
add_history_to_context |
第 5 篇 Storage | 上下文中带入历史对话 |
instructions |
第 4 篇 Instructions | 指导领导者的协调策略 |
第五步:跑起来
if __name__ == "__main__":
user_id = "researcher_01"
# 第一次研究任务
response = research_team.run(
"研究一下2024年大语言模型的最新进展",
user_id=user_id,
)
report: ResearchReport = response.content
print(f"主题: {report.topic}")
print(f"摘要: {report.summary}")
print(f"可信度: {report.confidence:.0%}")
print("\n关键发现:")
for finding in report.key_findings:
print(f" - {finding}")
print(f"\n建议: {report.recommendation}")
print("\n信息来源:")
for source in report.sources:
print(f" - {source}")
因为用了 output_schema=ResearchReport,response.content 直接就是一个 Pydantic 对象,字段该是什么类型就是什么类型。不用解析 JSON,不用切字符串。
如果你想要流式输出看过程,也可以换成 print_response:
research_team.print_response(
"研究一下2024年大语言模型的最新进展",
user_id=user_id,
stream=True,
)
完整代码
把上面的步骤合到一个文件里,直接可以跑:
from typing import List
from pydantic import BaseModel, Field
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.db.sqlite import SqliteDb
from agno.knowledge.knowledge import Knowledge
from agno.knowledge.embedder.openai import OpenAIEmbedder
from agno.vectordb.chroma import ChromaDb
from agno.vectordb.search import SearchType
from agno.memory import MemoryManager
from agno.team.team import Team
from agno.tools.duckduckgo import DuckDuckGoTools
# --- 输出结构 ---
class ResearchReport(BaseModel):
topic: str = Field(..., description="研究主题")
summary: str = Field(..., description="核心摘要,3-5句话")
key_findings: List[str] = Field(..., description="关键发现,3-5条")
sources: List[str] = Field(..., description="信息来源")
recommendation: str = Field(..., description="建议和下一步行动")
confidence: float = Field(..., ge=0, le=1, description="结论可信度 0-1")
# --- 基础设施 ---
db = SqliteDb(db_file="tmp/research_assistant.db")
model = OpenAIChat(id="gpt-4o-mini")
knowledge = Knowledge(
name="research_kb",
vector_db=ChromaDb(
name="research",
collection="research",
path="tmp/chromadb",
persistent_client=True,
search_type=SearchType.hybrid,
embedder=OpenAIEmbedder(id="text-embedding-3-small"),
),
max_results=5,
contents_db=db,
)
memory_manager = MemoryManager(
model=model,
db=db,
additional_instructions="记录用户的研究兴趣领域和偏好的报告风格",
)
# --- 专精 Agent ---
researcher = Agent(
name="网络研究员",
role="搜索网络获取最新信息",
model=model,
tools=[DuckDuckGoTools()],
instructions="""\
你是一个网络研究员。
1. 搜索与主题相关的最新信息
2. 收集多个来源的数据
3. 整理成结构化的研究笔记
只收集信息,不做分析判断。""",
)
analyst = Agent(
name="知识分析师",
role="分析信息并提取洞察",
model=model,
knowledge=knowledge,
search_knowledge=True,
instructions="""\
你是一个知识分析师。
1. 分析研究员提供的信息
2. 结合知识库中的已有知识
3. 提取关键洞察和趋势
4. 评估信息的可信度
客观分析,用数据说话。""",
)
# --- 研究团队 ---
research_team = Team(
name="AI研究助手",
model=model,
members=[researcher, analyst],
instructions="""\
你是研究团队的负责人。
1. 先让研究员搜索相关信息
2. 再让分析师进行深度分析
3. 综合两者的结果,生成研究报告
报告要客观、有数据支撑。""",
output_schema=ResearchReport,
db=db,
memory_manager=memory_manager,
enable_agentic_memory=True,
add_history_to_context=True,
num_history_runs=3,
markdown=True,
)
# --- 运行 ---
if __name__ == "__main__":
user_id = "researcher_01"
response = research_team.run(
"研究一下2024年大语言模型的最新进展",
user_id=user_id,
)
report: ResearchReport = response.content
print(f"主题: {report.topic}")
print(f"摘要: {report.summary}")
print(f"可信度: {report.confidence:.0%}")
print("\n关键发现:")
for finding in report.key_findings:
print(f" - {finding}")
print(f"\n建议: {report.recommendation}")
print("\n信息来源:")
for source in report.sources:
print(f" - {source}")
整体架构
跑起来之后,信息是这么流动的:
用户
|
v
研究团队 (Team Leader)
|
+------ 委派 ------+
| |
v v
网络研究员 知识分析师
(DuckDuckGo) (Knowledge/RAG)
| |
+------ 汇总 ------+
|
v
Team Leader 综合 -> ResearchReport (结构化输出)
|
+-> Memory (记住用户偏好)
+-> Storage (保存对话历史)
每个箭头都对应一个我们学过的概念。没有魔法,就是积木搭出来的。
扩展思路
这个项目是一个起点,你可以根据需要往上加东西:
加 Guardrails(第 8 篇) – 校验研究报告的质量,比如 confidence 太低就打回重做:
from agno.guardrails import GuardrailResult
def check_report_quality(report: ResearchReport) -> GuardrailResult:
if report.confidence < 0.3:
return GuardrailResult(pass_guardrail=False, reason="可信度太低,需要更多信息")
if len(report.key_findings) < 2:
return GuardrailResult(pass_guardrail=False, reason="关键发现太少")
return GuardrailResult(pass_guardrail=True)
加 Hooks(第 9 篇) – 在研究开始前记录日志,研究完成后发通知。
换成 Workflow(第 12 篇) – 如果研究流程更复杂(比如先做初步调研、再让用户确认方向、然后深入研究),用 Workflow 来编排多个阶段。
加 Reasoning(第 13 篇) – 让分析师在回答前先”想一想”,推理过程更透明。
加 Evals(第 16 篇) – 自动评测研究报告的质量,持续优化 Agent 的表现。
这些都不需要重写代码,只需要在现有结构上加参数或者加组件。这就是 Agno 的设计哲学 – 渐进式复杂度。
系列总结
17 篇教程,从第一行 agent.print_response("你好") 走到了一个完整的多 Agent 研究助手。回头看看,我们到底学了什么。
知识地图
| 篇目 | 主题 | 核心概念 |
|---|---|---|
| 01 | Hello Agent | Agent 基本创建和运行 |
| 02 | Tools | 给 Agent 装工具,让它能做事 |
| 03 | Structured Output | 用 Pydantic 控制输出格式 |
| 04 | Instructions | 通过指令塑造 Agent 的行为 |
| 05 | Storage | 对话历史持久化和 Session 管理 |
| 06 | Memory | 跨会话记住用户偏好 |
| 07 | Knowledge/RAG | 给 Agent 灌入你的数据 |
| 08 | Guardrails | 输入输出校验,防止跑偏 |
| 09 | Hooks | 生命周期钩子,精细控制流程 |
| 10 | Teams | 多 Agent 协作的四种模式 |
| 11 | Teams 进阶 | 知识共享、记忆传递、嵌套团队 |
| 12 | Workflows | 多步骤流程编排 |
| 13 | Reasoning | 让 Agent 先思考再回答 |
| 14 | Learning | Agent 从反馈中学习进步 |
| 15 | Multi-Model | 多模型切换和对比 |
| 16 | Evals | 自动化评测 Agent 表现 |
| 17 | 综合实战 | 全部概念组装成完整项目 |
核心概念速查
- Agent – 最基本的单元,一个模型 + 指令 + 工具的组合
- Tools – Agent 的能力扩展,搜索、计算、调 API 都靠它
- Output Schema – 结构化输出,让 Agent 的回复可编程处理
- Instructions – Agent 的行为指南,好指令等于好员工
- Storage – 对话历史持久化,Session 内的记忆
- Memory – 跨 Session 的长期记忆,记住用户是谁
- Knowledge – 向量知识库 + RAG,让 Agent 基于你的数据回答
- Guardrails – 输入输出的安全网,该拦的拦
- Hooks – 生命周期钩子,在关键节点插入自定义逻辑
- Team – 多 Agent 协作,coordinate/route/broadcast/tasks 四种模式
- Workflow – 多阶段流程编排,比 Team 更灵活
- Reasoning – 先想后答,推理过程可见
- Learning – 从用户反馈中持续优化
- Evals – 自动化评测,量化 Agent 表现
一些实战建议
从简单开始。 不要上来就搞 Team + Workflow + Knowledge 全家桶。先用一个 Agent 解决问题,发现单个 Agent 搞不定了再加复杂度。绝大多数场景,一个配置良好的 Agent 就够用了。
Instructions 是最重要的调优手段。 很多人一上来就想换更大的模型、加更多工具。其实最该花时间的是写好 instructions。好的指令能让 gpt-4o-mini 干出 gpt-4o 的效果。
结构化输出用起来。 一旦 Agent 的输出要被程序处理(存数据库、渲染 UI、喂给下一个 Agent),就用 output_schema。别去解析字符串,那是自找苦吃。
生产环境用 PostgreSQL。 SQLite 适合开发和学习,生产环境换成 PostgreSQL,接口完全一样,只换连接配置就行。
多看 cookbook。 Agno 的 cookbook/ 目录里有 1500+ 个示例,覆盖了各种场景。遇到问题先去翻翻,大概率有人已经写好了。
最后
这 17 篇教程覆盖了 Agno 框架的核心能力。但框架只是工具,真正有价值的是你用它解决的问题。
去构建你自己的 Agent 吧。从一个简单的开始。