第 43 章 · Agno 综合实战：Research Agent

把 Part 10 合起来

全书从一个想查竞品价格的跨境电商卖家开始。前 34 章，我们用 pi 把 Agent 的模式一个个讲透；Part 10 前七章，我们用 Agno 把这些模式一个个落地——工具、Knowledge、Storage/Memory/State、Team、Workflow、AgentOS、治理。

现在是把它们合起来的时候。这一章不引入任何新的 Agno 概念，只回答一个问题：

当你要真正做一个属于自己的、生产可用的 Agno 应用时，该怎么把这些能力组装起来？

这正是主书第 35 章“Building on the Harness”在 Agno 里的对应版本。主书那一章讲的是“在 pi 之上，从四个扩展面注入业务”；这一章讲的是“用 Agno 的电池组件，从单 Agent 一步步长成一个完整应用”。落地路径不同，但组装的思路完全一致——从最薄的能跑版本起步，按真实复杂度逐层加,而不是一上来就堆满所有组件。

我们要造的还是那个卖家的助手,目标能力：

用工具搜索公开信息、查询业务数据（第 37 章 / 主书第 3-4 章）。
用 Knowledge 回答公司资料和合规政策（第 38 章 / 主书第 11 章）。
用 Storage 恢复会话（第 39 章 / 主书第 12 章）。
用 Memory 记住用户长期偏好（第 39 章 / 主书第 10 章）。
用 Team 做多角色并行研究（第 40 章 / 主书第 15、17、18 章）。
用 Workflow 管固定的周报发布流程（第 40 章 / 主书第 17 章）。
用 AgentOS/API 变成可交互、可观测的服务（第 41 章 / 主书第 22-24 章）。
用治理、评测、上线清单约束风险（第 42 章 / 主书第 25-28 章、附录 D/E）。

第一阶段：单 Agent 起步

不要一上来就 Team + Workflow + AgentOS。先从一个能把“工具、会话、输出口径”跑通的单 Agent 开始——这是最容易验证、最容易调试的形态。

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.storage.sqlite import SqliteStorage
from agno.tools.duckduckgo import DuckDuckGoTools

from tools.sales import query_sales  # 只读销量工具，见第 42 章

research_agent = Agent(
    name="CrossBorderResearchAgent",
    model=OpenAIChat(id="gpt-4o"),
    user_id="seller_001",
    session_id="weekly_research",
    # Storage：会话持久化，跨进程恢复"聊过什么"（第 39 章）
    storage=SqliteStorage(
        table_name="research_sessions",
        db_file="tmp/research.db",
    ),
    tools=[DuckDuckGoTools(), query_sales],
    instructions=[
        "你是跨境电商研究助手。",
        "涉及最新公开信息时先用搜索工具。",
        "涉及自家销量时调用只读销量工具，绝不编造数字。",
        "输出先给结论，再列依据，最后给建议动作。",
    ],
    add_history_to_messages=True,   # 把历史带进上下文
    num_history_responses=5,        # 只带最近 5 轮，控上下文成本
    show_tool_calls=True,
    markdown=True,
)

research_agent.print_response(
    "最近三个月我的德国站 SKU-123 卖得怎么样，竞品有没有降价？",
    stream=True,
)

第一阶段只追求一件事：能把工具、会话、输出口径跑通。这里已经隐含了两个主书原则——num_history_responses=5 是第 9 章的上下文预算控制（别把全部历史无脑塞进去），“绝不编造数字”配只读工具是第 4 章的工具优先于臆测。跑顺了再往上加。

第二阶段：加入 Knowledge

现在给它接上公司资料——合规政策、选品手册、物流规则、平台条款。这是第 38 章的 RAG 能力。关键不是“能答”，而是答案要带来源、能追溯：

from agno.knowledge.pdf_url import PDFUrlKnowledgeBase
from agno.vectordb.lancedb import LanceDb, SearchType
from agno.embedder.openai import OpenAIEmbedder

# 构建带 hybrid search 的知识库（第 38 章）
policy_knowledge = PDFUrlKnowledgeBase(
    urls=["https://example.com/eu-compliance.pdf",
          "https://example.com/logistics-rules.pdf"],
    vector_db=LanceDb(
        table_name="policies",
        uri="tmp/lancedb",
        search_type=SearchType.hybrid,          # 关键词 + 向量混合检索
        embedder=OpenAIEmbedder(id="text-embedding-3-small"),
    ),
)
policy_knowledge.load(recreate=False)  # 首次构建索引；已存在则跳过

research_agent = Agent(
    name="CrossBorderResearchAgent",
    model=OpenAIChat(id="gpt-4o"),
    knowledge=policy_knowledge,
    tools=[DuckDuckGoTools(), query_sales],
    instructions=[
        "回答合规、物流、平台规则问题时，必须优先检索知识库。",
        "资料不足时明确说明没有找到依据，不要靠常识硬答。",
        "涉及政策结论时必须列出来源文档。",
    ],
    add_references=True,   # 把检索到的片段作为引用带进回答
    markdown=True,
)

SearchType.hybrid 混合了关键词与向量检索——对付“欧盟 CE 认证”这种既有专有名词又需语义匹配的合规查询效果最好。add_references=True 让回答带上依据片段，这是主书第 11 章反复强调的：RAG 的价值不只在检索到，更在能追溯到哪一份文档的哪一段。 这一步做完，你就能跑第一类评测——问一个合规问题，检查它是否命中正确文档、是否引用了正确来源。

第三阶段：加入 Memory

第三层，加入用户长期偏好。这个卖家总希望报告简短、只关心德国市场、主攻玩具品类——这些是跨会话稳定的事实，属于 Memory（第 39 章 / 主书第 10 章）：

from agno.memory.v2.memory import Memory
from agno.memory.v2.db.sqlite import SqliteMemoryDb

user_memory = Memory(
    db=SqliteMemoryDb(table_name="user_memories", db_file="tmp/memory.db"),
)

research_agent = Agent(
    name="CrossBorderResearchAgent",
    model=OpenAIChat(id="gpt-4o"),
    knowledge=policy_knowledge,
    tools=[DuckDuckGoTools(), query_sales],
    memory=user_memory,
    enable_user_memories=True,        # 自动从对话中提炼用户偏好
    enable_session_summaries=True,    # 自动生成会话摘要
    add_references=True,
    markdown=True,
)

这里有一条第 39 章反复强调、也最容易踩的边界：不要把“本次调研命中的网页片段”写进 Memory。 网页片段是本次任务的临时材料，属于 Knowledge/Context；Memory 只装“这个用户是谁、他长期偏好什么”这类跨会话稳定的事实。四个概念的分工务必记牢——Storage=聊过什么，Memory=用户是谁，State=进行到哪，Context=本次额外参考。 混淆它们，是 Agno 应用最常见的架构错误。

第四阶段：拆成 Team

当单 Agent 开始工具过多、角色过杂——既要搜网页、又要查销量、又要读合规、还要写报告——就该拆成 Team（第 40 章 / 主书第 16 章的编排、第 17/18 章的多角色协作）。每个成员只拿自己需要的工具，这既更安全（第 42 章的最小权限），也更好评测：

from agno.team import Team
from agno.models.openai import OpenAIChat

web_researcher = Agent(
    name="WebResearcher",
    role="搜索公开市场与竞品信息",
    model=OpenAIChat(id="gpt-4o"),
    tools=[DuckDuckGoTools()],
    instructions=["只负责检索公开信息，给出带链接的原始发现。"],
)

compliance_analyst = Agent(
    name="ComplianceAnalyst",
    role="查询公司知识库与政策依据",
    model=OpenAIChat(id="gpt-4o"),
    knowledge=policy_knowledge,
    instructions=["只回答有知识库依据的合规问题，必须列来源。"],
)

sales_analyst = Agent(
    name="SalesAnalyst",
    role="调用只读销量工具分析自家表现",
    model=OpenAIChat(id="gpt-4o-mini"),   # 简单任务用便宜模型（第 42 章成本）
    tools=[query_sales],
    instructions=["只做销量数据的取数与趋势描述，不编造。"],
)

editor = Agent(
    name="Editor",
    role="把各方发现汇总成结构化报告",
    model=OpenAIChat(id="gpt-4o-mini"),
    instructions=["结论先行，再列依据，最后给建议动作。"],
)

research_team = Team(
    name="CrossBorderResearchTeam",
    mode="coordinate",   # 由 Team 协调各成员分工与汇总（第 40 章）
    model=OpenAIChat(id="gpt-4o"),
    members=[web_researcher, compliance_analyst, sales_analyst, editor],
    instructions=[
        "根据问题把子任务分给合适的成员，最后由 Editor 汇总。",
        "合规结论必须来自 ComplianceAnalyst 的带来源回答。",
    ],
    show_tool_calls=True,
    markdown=True,
)

research_team.print_response(
    "给我一份德国站玩具品类的本周竞品与合规简报", stream=True,
)

mode="coordinate" 是关键——Team 的协调者负责把问题拆成子任务、分派给合适成员、再收敛成一份报告。这正是主书第 18 章“发散调研、收敛综合”的模式：多个研究员并行探索，一个 Editor 汇总。注意 sales_analyst 和 editor 用了 gpt-4o-mini——取数和格式化不需要顶配模型，这是第 42 章成本分层的直接应用。

第五阶段：用 Workflow 管周报流程

Team 擅长开放式协作，但周报不是开放聊天，而是一条固定流程——每一步顺序确定、其中一步必须人审、失败要能从断点恢复。这种确定性流程用 Workflow 表达更合适（第 40 章 / 主书第 17 章）：

from agno.workflow import Workflow, Step

weekly_report = Workflow(
    name="WeeklyReportWorkflow",
    steps=[
        Step(name="collect_market",   agent=web_researcher),      # 1 收集市场动态
        Step(name="query_sales",      agent=sales_analyst),       # 2 查询自家销量
        Step(name="check_compliance", agent=compliance_analyst),  # 3 检索合规资料
        Step(name="draft_report",     agent=editor),              # 4 生成报告草稿
        Step(name="human_review",     agent=review_agent),        # 5 人工审核（人审门）
        Step(name="publish",          agent=publish_agent),       # 6 发布或归档
    ],
    storage=SqliteStorage(
        table_name="weekly_report_runs",
        db_file="tmp/workflow.db",     # 步骤级持久化，支持从断点恢复
    ),
)

weekly_report.print_response(input="生成 2026 年第 27 周德国站玩具周报")

Team 与 Workflow 的分工是第 40 章的核心决策：流程开放、需要智能分派用 Team；流程固定、要顺序/人审/可恢复用 Workflow。 实践中两者常嵌套——这里第 1-4 步的研究协作本身可以交给上面的 research_team，而 Workflow 负责整体的顺序、第 5 步的人审暂停（publish_agent 里的改价/发布工具带 requires_confirmation=True，接第 42 章）、以及失败后从 weekly_report_runs 表恢复。

第六阶段：接入服务层

当项目从“我本地跑”进入“团队一起用”，就补服务层——用 Agno 的 AgentOS（第 41 章 / 主书第 22-24 章）：

from agno.os import AgentOS

# 把 Team 和 Workflow 一起挂上 AgentOS，得到 API + Playground + trace
agent_os = AgentOS(
    teams=[research_team],
    workflows=[weekly_report],
)
app = agent_os.get_app()   # 标准 FastAPI 应用，可用 uvicorn 起服务

AgentOS 免费给你四样东西：API（创建会话、发消息、查任务状态）、Playground/UI（调试和前端入口）、trace（记录模型/工具/Knowledge/Team/Workflow 每一步）、以及会话与状态的可恢复性。你要在它之上补的，是第 42 章讲的控制面：auth.py 把请求绑定到 user_id/tenant_id 并做租户隔离，tracing.py 把成本按维度归因，写操作接后台队列。

有一个反直觉但重要的经验：服务层不是最后“包一层接口”就完事。 它会反向影响你前面的设计——一旦要多用户并发，你就得回头确认 Storage 的 session_id 是否按用户隔离、State 是否线程安全、工具返回值的错误语义是否结构化（前端要能区分“没找到”和“没权限”）。所以真实项目里，服务层的约束往往会让你重构前五个阶段的某些决定。这也是为什么我们强调逐层长出来——每加一层，都可能倒逼上一层做对。

上线门槛

组装完不等于能上线。用这张表做最终检查——它就是第 42 章控制面在这个项目上的逐项落地：

类别	必须满足	对应章
Tools	只读默认，写操作人审，错误结构化	36 / 41
Knowledge	引用来源、索引版本化、RAG 评测通过	37
Storage	会话可恢复、可删除、可审计	38
Memory	用户可控、敏感信息过滤、不混入网页片段	38
Team	成员角色清楚、工具最小化	39
Workflow	步骤可恢复、可暂停、可重试	39
Observability	trace 覆盖模型、工具、RAG、成本	40
Budget	单 run/用户/租户/后台任务都有上限	41
Evaluation	核心任务、高风险样本、真实回放都跑过	41 / 附录 D
Rollback	prompt、模型、工具、索引、workflow 都能回退	41 / 附录 E

动手看看

按上面的六个阶段,自己搭一遍——但每次只加一层，加完立刻跑一次。特别做这个实验：在第四阶段（Team）搭好后，故意给 sales_analyst 也装上 DuckDuckGoTools，然后问一个纯销量问题，观察它会不会“越权”去搜网页而不是查数据库。你会亲眼看到“每个成员只拿自己需要的工具”这条最小权限原则为什么重要——工具给多了，模型就会在不该用的地方乱用。再把多余工具去掉，对比行为的变化。

反向映射回主书

Part 10 学完，最有价值的收获是这张反向映射表——它证明主书的每个架构控制点，在 Agno 里都有对应落点。哪天你换到第三个框架，也能立刻认出这些点：

Agno 项目能力	主书章节
Agent + instructions	第 1-3 章
Tools + MCP	第 3-4 章
response_model 结构化输出	第 6 章
Knowledge / RAG	第 11 章
Storage / Memory / State / Context	第 8-11 章、附录 B
Team（coordinate）	第 15、17、18 章
Workflow	第 16、39 章
AgentOS / API / trace	第 22-24、32 章
governance / security / HITL	第 25-28 章
eval / ops	附录 D、E

实战中的几个坑

坑一：一上来就把所有组件堆满。

现象：第一版就写 Team + Workflow + Memory + AgentOS，跑不通时完全不知道错在哪一层。
原因：跳过了“单 Agent 能跑”的地基，把调试难度叠加到了最大。
对策：严格按六阶段逐层加，每加一层跑一次；上一层稳了再上下一层。

坑二：把网页检索片段写进 Memory。

现象：Memory 越攒越大，还把上周某次调研的过时竞品价当成用户“偏好”反复带进上下文。
原因：混淆了 Memory（用户是谁）和 Knowledge/Context（本次材料）。
对策：Memory 只存跨会话稳定的用户事实，临时材料走 Knowledge/Context，牢记四概念分工。

坑三：该用 Workflow 的固定流程硬塞给 Team。

现象：周报流程用 Team 开放协作，结果每次步骤顺序都不一样，人审步骤时有时无，失败没法从断点恢复。
原因：把确定性流程当成了开放式协作。
对策：固定顺序 + 人审 + 可恢复的流程用 Workflow；开放式智能分派才用 Team，两者可嵌套。

坑四：把服务层当成最后“包接口”。

现象：前五阶段都单用户跑通，上 AgentOS 后多用户并发一来，会话串号、State 被并发改乱。
原因：Storage 的 session 隔离、State 的并发安全没在前面设计好，服务层无法事后补救。
对策：接服务层前先想清并发与多租户，必要时回头重构 Storage/State——服务层的约束要反向驱动前面的设计。

Agno vs 主书 pi 做法

组装维度	Agno 怎么做	pi（harness）怎么做	各自适合谁
整体心智	用电池组件逐层拼装（Agent→Team→Workflow→AgentOS）	在 harness 之上从四个扩展面注入业务	Agno 快出完整应用，pi 深度定制运行时
业务工具	自定义 Python 函数 + 内置工具 + MCP	`registerTool` 注册 + MCP	都接 MCP 复用；Agno 更贴近普通函数
领域方法论	写进 instructions / 成员 role	封装成 `SKILL.md` 技能	pi 的技能更可复用、可版本化
多 Agent	Team（coordinate） + Workflow	Orchestration（研究/辩论/流水线）	模式一致，抽象层次不同
服务化	AgentOS 一键出 API/Playground	RPC 远程驱动 + 后台 run	Agno 开箱即用，pi 融入既有 harness
运行时通用能力	框架内建（Storage/Memory/trace）	harness 内建（循环/压缩/持久化/多 provider）	都“免费继承”，只是继承的东西边界不同

差异归结成一句：Agno 是“电池全含、逐层拼装出应用”，pi 是“在 harness 之上、从扩展面注入业务”。 但你注意到没有——两栏的每一行，做的都是同一件事：接业务工具、沉淀方法论、组织多 Agent、服务化、继承通用运行时。这就是全书最想让你记住的：模式是一套，框架只是它的不同实现。

小结

Agno 适合把主书的架构模式快速组合成完整 Agent 应用，这一章对应主书第 35 章的“组装”高度。
综合项目要从单 Agent 起步，按真实复杂度逐层加入 Knowledge、Memory、Team、Workflow、AgentOS——每加一层跑一次，别一次堆满。
Team 管开放式智能协作，Workflow 管固定可恢复流程，两者常嵌套；四概念（Storage/Memory/State/Context）分工不能混。
生产化不是最后一步，而是贯穿工具、数据、会话、权限、观测、评测、回滚——服务层的约束还会反向驱动前面的设计。
Part 10 的最终目的，是让你换一个框架也能认出同一套架构控制点——反向映射表证明：Agno 的每个组件都对得上主书的某个模式。

到这里，全书形成了一个闭环：前 34 章用 pi 看懂 Harness 的原理，Part 10 用 Agno 验证同一套模式在应用层的落地。从那个想查竞品价格的卖家，到一个组件齐备、治理到位、能上生产的跨境电商 Research Agent——框架会变，模式不会。你学到的，是任何 harness、任何框架都通用的那套东西。