企业官网建设流程全解析

写这个系列第三篇之前，我把后台留言翻了一遍，问得最多的不是“怎么调API”，而是“Agent一复杂就翻车怎么办”：多轮工具调用互相覆盖、上下文一长就失忆、本地能跑的生产环境就崩。这三类问题几乎每个做过Agent的人都会撞上。这篇就专门把硬骨头啃掉——先说Agent主流架构怎么选，再把工具调用、记忆系统、多Agent协作的落地细节讲透，最后给出从Jupyter脚本到生产服务的完整部署方案。内容偏进阶，但我会把每一步的“为什么”也交代清楚，方便不同基础的读者都能顺着思路复现。

1. Agent主流架构：单轮、ReAct、Plan-and-Execute，别再只会堆提示词

很多人做Agent的第一反应是写一段很长的System Prompt，把功能全部塞进去，然后让模型直接返回结果。这种做法本质上还是“单轮LLM调用”，不是Agent。真正的Agent至少要具备“感知-决策-行动-观察”的闭环能力。先把这个闭环拆清楚，才能理解为什么后面几个工程化的设计如此重要。

1.1 ReAct循环：最基础也最容易被玩坏的架构

ReAct（Reasoning + Acting）是Agent最经典的实现方式：模型先推理，决定调什么工具，拿到工具结果后再推理，再调下一个工具，直到认为任务完成。这个循环写出来非常短，但真正让它稳定的细节都在循环外面。

下面是一个功能完整的ReAct简化版，我尽量把必要的骨架都保留：

import json from openai import OpenAI client = OpenAI(api_key="YOUR_API_KEY") TOOLS = [ { "type": "function", "function": { "name": "web_search", "description": "搜索公开信息，返回网页标题、摘要和链接", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "搜索关键词，建议使用名词短语"} }, "required": ["query"] } } } ] def run_agent(user_prompt: str, max_steps: int = 8): messages = [{"role": "user", "content": user_prompt}] for step in range(max_steps): resp = client.chat.completions.create( model="gpt-4o", messages=messages, tools=TOOLS, tool_choice="auto", ) msg = resp.choices[0].message # 模型不再请求调用工具，说明它认为可以给出最终答案了 if not msg.tool_calls: return msg.content # 把模型的“请求动作”追加到对话里 messages.append(msg) # 逐个执行工具调用 for tc in msg.tool_calls: args = json.loads(tc.function.arguments) try: result = web_search(args["query"]) observation = {"status": "ok", "data": result} except Exception as exc: # 工具出错时，把错误信息当作观察结果还给模型，让它自己修正 observation = {"status": "error", "message": str(exc)} messages.append({ "role": "tool", "tool_call_id": tc.id, "content": json.dumps(observation, ensure_ascii=False) }) return "达到最大工具调用步数，任务未完成"

这段代码里有几个必须注意的坑：

第一，tool_call_id必须和模型返回的tool_call一一对应，拼错了API直接报400。所以工具结果要按tool_calls里的顺序逐条回填，不能批量套一个ID。

第二，工具调用次数必须有上限。我见过很多线上事故，都是Agent在一个错误分支里反复调用同一个工具，把成本烧爆了。max_steps不是可选项，是必需品。

第三，工具参数用json.loads解析后，一定要跑一遍字段校验。模型偶尔会在JSON里塞进多余字段，或者把数字写成字符串，直接传给底层工具会很危险，后面我会专门讲校验方案。

第四，工具的报错信息要原样返回给模型，而不是自己忍住。很多新手喜欢在except里打印日志就完事，结果模型不知道工具失败了，继续用错误假设往下编，产出漏洞百出的结论。把{"status": "error", "message": ...}作为观察结果传回去，模型才有机会自我纠正。这是ReAct日志里最能体现“Agent感”的一环。

1.2 Plan-and-Execute：复杂任务的稳定器

ReAct适合“走一步看一步”的动态任务，但遇到那种步骤多、容错低的活，比如“调研行业现状并生成结构化报告”，纯ReAct很容易东一榔头西一棒子。这时候更适合先做计划，再执行，也就是Plan-and-Execute架构：第一轮让模型把任务拆成一个有序的子任务清单，然后按清单逐个执行，每完成一个子任务就把结果汇总到最终目标里。

我实践中常用的简化流程是：

plan_prompt = """请把用户任务拆成不超过5个步骤的子任务清单。 每个子任务必须给出：步骤说明、是否需要调用工具、期望输出。 只输出JSON数组，不要多余解释。 """ def plan_and_execute(user_prompt: str): plan_text = client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": plan_prompt + "\n\n" + user_prompt}], response_format={"type": "json_object"}, ) plan = json.loads(plan_text.choices[0].message.content) context = [] for step in plan["steps"]: # 每个子任务都可以套一个ReAct子循环 step_result = run_agent( f"当前任务：{step['description']}\n已知上下文：{context}\n子任务期望输出：{step['expected_output']}" ) context.append({"step": step["description"], "result": step_result}) # 最后汇总所有子任务结果，生成最终输出 final_prompt = f"用户原始需求：{user_prompt}\n\n各步骤结果：\n{json.dumps(context, ensure_ascii=False)}\n\n请综合整理成最终交付内容。" return client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": final_prompt}] ).choices[0].message.content

这套做法的优势非常明显：每个子任务的上下文范围被严格限定，不会出现某一步跑偏后污染全局；中途宕机或模型输出格式失控时，可以断点续跑；子任务之间天然可以拆给不同模型、不同并发度去执行。

代价是总Token消耗明显上升，且计划一旦制定后不太容易根据中途信息动态调整。如果任务对实时性要求高、前面几步的结果可能否定整个计划，Plan-and-Execute就会很僵化。

1.3 架构选型决策表：没有银弹，按场景配比

我在实际项目里很少只用一种架构，通常是混合的：先Plan定骨架，子步骤用ReAct去执行，执行到一半发现新信息就触发“重新计划”。下面这个表是我选择架构时的基本判断逻辑：

多说一句，很多框架把ReAct吹得神乎其神，但真实场景里，40%的需求用“单轮Function Call + 一个完善的后处理函数”就能解决，根本不需要Agent循环。不要为了“Agent”而Agent，架构越简单，线上越好维护。这也是我前面强调“别只会堆提示词”的原因——把任务拆成“需要用Agent的部分”和“不需要用Agent的部分”，比让模型把所有事都干完更稳定。

2. 工具调用工程化：写清SOP、管好异常、守住安全边界

Agent的能力上限，很大程度取决于你能给它多少靠谱的“手”。而手的好坏，不在函数名字好不好听，而在工具描述、参数定义和错误处理这三层工程细节。

2.1 工具描述要像给实习生写SOP，而不是给老同事写备注

模型不熟悉你的业务，它只是一个“精于字面匹配的决策者”。工具描述写得越模糊，模型越容易用错参数。我见过一个典型的反例：某团队封装了一个发邮件的工具，描述写的是“发送邮件”，参数是to、content，结果模型经常把收件人姓名传进to，因为描述里没说明格式。

正确做法是把单位、格式、边界、甚至一个示例都写进去：

{ "type": "function", "function": { "name": "send_email", "description": "发送一封文本格式邮件。收件人必须是完整的邮箱地址，多个收件人用逗号分隔。禁止用收件人姓名代替邮箱地址。", "parameters": { "type": "object", "properties": { "to": { "type": "string", "description": "收件人邮箱，示例：alice@example.com,bob@example.com" }, "subject": {"type": "string", "description": "邮件主题，不超过100个字"}, "content": {"type": "string", "description": "邮件正文，纯文本"} }, "required": ["to", "subject", "content"] } } }

这样的描述本质上是一份SOP：明确“怎么做”、明确“不能怎么做”、给了一个示范。不要小看这几行字，它往往能把工具调用成功率从70%拉到95%以上。

2.2 用Pydantic统一管理工具定义与入参校验

手写JSON Schema又一个很大的问题：模型传参和实际函数签名一旦对不上，只能在运行时崩溃。我的做法是：用一个工具注册器，函数参数直接定义成Pydantic模型，然后自动生成JSON Schema。这样函数签名、校验规则、Schema描述三处不会出现不一致。

from pydantic import BaseModel, EmailStr, Field from typing import Any, Callable class SendEmailInput(BaseModel): to: str = Field(description="收件人邮箱，多个用逗号分隔") subject: str = Field(description="邮件主题") content: str = Field(description="邮件正文") class ToolRegistry: def __init__(self): self._tools = {} self._schemas = [] def register(self, schema_cls: type[BaseModel]): def decorator(func: Callable): name = func.__name__ result_cls = self._to_json_schema(schema_cls) result_cls["name"] = name result_cls["description"] = func.__doc__ or "" self._tools[name] = (schema_cls, func) self._schemas.append({"type": "function", "function": result_cls}) return func return decorator def call(self, name: str, args: dict): schema_cls, func = self._tools[name] parsed = schema_cls(**args) # 校验失败会抛异常 return func(**parsed.model_dump()) def _to_json_schema(self, schema_cls): schema = schema_cls.model_json_schema() parameters = {k: v for k, v in schema.items() if k != "title"} return {"parameters": parameters}

这套结构有三个隐形的收益：一是新增工具只需要写一个函数、一个Schema类，不容易漏；二是模型调用时即使传了多余字段，Pydantic默认会忽略，不会污染底层逻辑；三是校验失败抛出的错误能被上一章的ReAct循环捕获并反馈给模型，形成自我修正闭环。

2.3 工具错误处理与安全边界：让模型试错，但别让它乱来

工具调用错误处理的核心心法只有一条：把错误当作观察结果，而不是程序异常。模型是一个“推理器”，你需要给它完整的感知信息，它才能做出正确决策。工具失败了、参数非法、外部服务超时，都要想办法转成结构化消息回填给模型。

但安全边界绝对不能靠模型自觉。LLM不是可信执行环境，绝不能因为“模型判断没问题”就直接放行写操作。我自己的项目里定了这么几条红线：

• 所有写操作（发邮件、改数据库、删文件）必须登记白名单，动态参数一律走模板校验。
• 凡是会真实影响外部系统的动作，Agent只能产出“拟执行内容”，由人工审核后触发。
• 工具结果里涉及隐私、密钥的字段，在回传给模型之前要做好脱敏。
• 给每个工具设置独立的执行超时，避免一个慢接口拖垮整个Agent循环。

举个制造过事故的例子：某Agent在回复用户时被诱导调用“删除项目”工具，因为工具描述正好只写了“根据参数删除项目”，模型也没意识到底层影响。现在我会在描述里加上“本操作不可逆，调用前必须确认用户明确输入了项目名且只涉及测试环境”，同时在工具函数内部强制二次确认参数包含特定标识。这种“描述约束+代码防御”的双保险，比单纯指望模型守规矩可靠得多。

3. 记忆系统：短期上下文、长期向量库、结构化事实三件套

Agent一长就“失忆”，几乎是所有人的共同痛点。它的根源很简单：LLM的上下文窗口有限，而真实业务里的历史信息是无限增长的。所以记忆系统要解决的问题不是“能不能记”，而是“该记什么、忘什么、从哪找回”。

3.1 三种记忆的分工

我把Agent的记忆拆成三类，和人类记忆做类比：

很多人把“记忆”和“上下文”混为一谈，这是认知上的误区。上下文只是当前任务的一次性草稿纸，草稿纸写满就该归档；长期记忆才是Agent持久价值的来源。比如一个售后支持Agent，它真正需要长期记住的不是每一条客流记录，而是“这个客户是VIP，偏好邮件联系，上次沟通中明确拒绝过电话推销”这类结构化事实。

3.2 上下文窗口管理的三种策略

工作记忆不可能无限制增长，我的日常处理方案有三种，按任务复杂度组合使用：

策略一：滑动窗口。只保留系统提示、最近N轮对话、以及当前正在处理的工具结果。最省Token，但会丢早期关键信息。

策略二：摘要压缩。当消息总数超过阈值时，把旧消息丢给模型生成一段摘要，替换掉原始消息。损失细节，但保住主线。

策略三：检索增强。每次收到用户新消息之前，先从长期记忆里检索和当前意图最相关的片段，拼接到上下文里。这是最“Agent”的做法，适合跨会话场景。

一个比较通用的伪代码如下：

def build_context(user_message, session_history, memory_store, max_context_tokens=8000): # 1. 取最近若干条历史 recent = session_history[-4:] # 2. 从长期记忆里检索与本次意图相关的片段 related_memories = memory_store.search(user_message, top_k=3) # 3. 拼装：系统提示 + 相关记忆 + 最近历史 + 当前消息 return { "system": SYSTEM_PROMPT, "memory": related_memories, "history": recent, "current": user_message }

这里有个重要细节：检索必须在“构造上下文”之前完成，因为你要用“当前用户消息”去检索记忆。而历史消息本身不需要全部参与检索，否则既费Token又容易带回噪声。我的经验是，检索时把“结构化事实”单独抽出来存储，和“对话痕迹”分开——对话痕迹适合用向量检索，结构化事实更适合用SQL精确查询。

3.3 长期记忆落地：向量库和结构化存储各司其职

长期记忆我通常做成两层：第一层是向量库，存对话摘要和文本信息；第二层是关系型数据库，存可量化的用户事实、项目状态、任务偏好。原因很简单，向量检索擅长“模糊匹配”，但“这个客户上次买的套餐是什么”这种精确查询用向量去做又慢又不准。

以Chroma为例，一个轻量的长期记忆实现：

import chromadb from chromadb.utils import embedding_functions client = chromadb.PersistentClient(path="./memory_store") col = client.get_or_create_collection( name="conversation_memories", embedding_function=embedding_functions.DefaultEmbeddingFunction() ) def save_memory(session_id: str, content: str, meta: dict): col.add( ids=[f"{session_id}-{hash(content)}"], documents=[content], metadatas=[{"session_id": session_id, **meta}] ) def search_memory(query: str, top_k: int = 3): results = col.query(query_texts=[query], n_results=top_k) return [doc for doc in results["documents"][0]]

写入记忆的时机很关键。我的习惯是：每个Agent循环结束后，把“结论性内容”而不是“过程性内容”写入长期记忆。比如写报告Agent，过程里搜了20个网页，只需要抽取核心结论存向量；中间那些“某链接打开失败”“搜索关键词重试两次”之类，写进去只会污染检索结果。

另外一个容易被忽视的坑：长期记忆写入时要带时间戳和来源。否则当任务涉及“最近一周的数据”时，Agent从记忆库里翻出三个月前的旧结论当新事实用，后果很严重。所以每次save_memory我都会在meta里加created_at（ISO格式字符串），检索时也会在提示词里明确“优先采用最新时间戳的记忆”。

4. 多Agent协作：监督者、流水线、市场式三种模式怎么落地

单Agent的能力有上限，但盲目堆多Agent只会让系统更脆弱。多Agent协作的本质是“合理的任务拆解 + 清晰的消息协议 + 可追踪的执行链路”。把这三件事做好，比Agent数量重要得多。

4.1 三种主流协作模式

我归纳了三种可落地的协作模式，分别对应不同场景：

监督者模式（Orchestrator-Worker）：一个中枢Agent负责任务规划、派发和结果汇总，多个Worker分别执行子任务。适合大多数生产场景。优点是职责清晰、好管控，缺点是中枢Agent可能成为性能瓶颈。

流水线模式（Pipeline）：任务按照固定顺序依次经过多个Agent，每个Agent只处理特定环节。适合流程固定的任务，比如“清洗数据 → 特征分析 → 生成图表 → 撰写结论”。优点是单点逻辑简单，缺点是链路中某环节失败会阻断整个流程。

市场/黑板模式（Blackboard）：多个Agent共享一个任务面板，谁有能力认领谁做。适合开放式探索场景，比如代码评审中让多个“专家视角”Agent各自发言。控制难度最高，实际项目慎用。

我自己的原则：能不用多Agent就不用。只有当一个任务确实由多个职责差异巨大的子任务组成，并且这些子任务能并行或必须隔离上下文时，才考虑拆Agent。拆出来的好处是“上下文隔离+专注度提升”，代价是“消息复制+多轮Token膨胀”。

4.2 消息协议与任务编排

多Agent之间通信，切忌直接传“字符串”。否则你根本查不清一个结论是哪个Agent、哪一轮、基于哪些上下文生产的。我常用的消息结构是一个Dataclass：

from dataclasses import dataclass, field from datetime import datetime from typing import Any @dataclass class AgentMessage: msg_id: str task_id: str sender: str receiver: str payload: dict created_at: str = field(default_factory=lambda: datetime.utcnow().isoformat())

每个字段都有它的用途：msg_id用于全局追踪，task_id串起一轮完整任务，sender/receiver让链路清晰，payload承载结构化数据。这样在日志系统里，只要按task_id过滤，就能看到整个Agent协作的完整链路——排查问题的时候，这条链路等于救命稻草。

然后是一个简化版监督者模式的编排骨架：

def orchestrator(user_request: str): workers = {"researcher": call_researcher, "analyst": call_analyst, "writer": call_writer} plan = generate_plan(user_request) # 返回子任务列表 results = {} for subtask in plan: worker_name = subtask["worker"] msg = AgentMessage( msg_id=f"msg_{subtask['step_id']}", task_id=f"task_{uuid4().hex[:8]}", sender="orchestrator", receiver=worker_name, payload={"request": subtask["description"], "deps": results} ) results[worker_name] = workers[worker_name](msg) return assemble_output(results)

编排中最容易踩的坑是死循环。比如一个Worker发现自己缺资料，向另一个Worker请求帮助，另一个Worker又反过来请求，两个Agent来回发消息出不来。我通常会给整个编排流程设一个全局步骤上限，超过上限直接终止并返回“任务过于复杂，需要人工介入”。这个“人工介入”不是丢人，多Agent系统本身就该有fallback到人的路径。

4.3 多Agent系统的成本与稳定性

如果说单Agent耗费的Token已经让你心疼，那么多Agent能让你心疼到麻木。每多一个Agent，意味着多一轮模型调用、多一次上下文组装、多一份重复的系统提示词。这些成本都是线性的，但问题排查难度是指数上升的。

成本上，我最常用三种手段：

• 模型分级：简单Worker用便宜小模型，只有规划、汇总这种复杂任务才上最强模型。
• 结果缓存：Worker输出如果无状态、可哈希，就按输入内容哈希做结果缓存，同子任务不重复执行。
• 延迟调用：Orchestrator不是每次都把全部子任务发出去，而是“按需拉取”，等前置结果真的被用到时再触发下一步。

稳定性上，最重要的一点是：每个Worker的输出都要做“格式校验”，而不是直接信它。我的Worker函数结尾都有统一的validate_output，检查返回JSON是否符合预定义的输出Schema，不符合就重试一次，重试再失败就标记为失败，让Orchestrator决定是跳过还是找人类。

5. 部署与服务化：状态持久化、并发控制和可观测性一个都不能少

从“我笔记本上能跑”到“线上能稳定扛流量”，中间差的不是代码量，而是几个工程意识。这一章我把最关键的三个问题讲透。

5.1 项目结构：别再让一切逻辑都堆在Jupyter里

一个可以部署的Agent项目，目录结构至少要让人一眼看出“入口、工具、Agent、存储”四层：

agent_project/ ├── main.py # FastAPI入口 / CLI入口 ├── agent/ │ ├── orchestrator.py # 编排逻辑 │ ├── memory.py # 长期记忆读写 │ └── loop.py # ReAct循环 ├── tools/ │ ├── registry.py # 工具注册器 │ ├── search.py │ └── database.py ├── schemas/ │ └── models.py # Pydantic数据结构 ├── storage/ │ ├── vector_store.py │ └── session_store.py └── config.py # 配置项集中管理

这个结构的意义在于：工具、Agent逻辑、存储三者解耦。替换一个向量库，或者换一个模型底座，不需要动Agent循环代码。很多项目做到中期重构，就是因为前期把所有函数都写在一个脚本里，改一行配置都要全局搜。

API层我喜欢用FastAPI，因为异步支持和Pydantic集成天然适合Agent服务：

from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class ChatRequest(BaseModel): session_id: str message: str stream: bool = False @app.post("/agent/chat") def chat(req: ChatRequest): # 从存储加载session状态，构建上下文，执行Agent循环，写回状态 return run_agent_session(req.session_id, req.message, req.stream)

如果你是Django项目，也可以直接把Agent逻辑封装成一个service层，放在视图函数里调用，本质一样，只是把API层从FastAPI换成Django view。

5.2 状态持久化与并发控制

最典型的部署事故是：把session状态放在Python进程的全局字典里。一旦开了多Worker进程或负载均衡，用户请求打到不同进程，上下文直接错乱。正确做法是把会话状态放到外部存储。

我这里有一套现成的持久化方案：

• 短期工作记忆（半个小时内会频繁读写）放Redis，用session_id做key，存最近的messages列表，TTL设为30分钟。
• 长期记忆放向量库加关系型数据库，跨session保留。
• 任务锁：对同一个session_id的并发请求加分布式锁，避免多个请求同时读写同一个Agent会话。

并发控制常被忽略。用户连按两次发送，两个请求同时操作同一个session，会出现消息顺序错乱、工具结果张冠李戴。用redis.setnx做锁是最省事的：

import redis r = redis.Redis(host="localhost", port=6379) lock_key = f"agent_lock:{session_id}" acquired = r.set(lock_key, "1", nx=True, ex=60) if not acquired: return {"error": "上一轮任务还在处理中，请稍候"}

这个逻辑虽然简单，但能挡住绝大多数并发引起的脏读脏写。

5.3 可观测性与成本控制

部署之后，你迟早会遇到同一个问题：“Agent上一轮干了什么？为什么结论这么奇怪？”没有可观测性，你只能对着屏幕发呆。

我最低要求的可观测性有四条，缺一不可：

• 结构化日志：每条Agent日志至少包含task_id、session_id、step、tool_name、tokens_used、latency_ms。
• 调用追踪：把Agent内部的每次LLM调用、工具调用都打点，能按task_id串成树状链路。
• 结果评测集：准备一组固定的测试用例，每次改完Agent逻辑都要跑一遍，防止“改一处坏全局”。
• 成本报表：按session_id或task_id汇总Token消耗，每天出报表。不看成本，你永远不知道多Agent到底烧了多少钱。

Token计数这块顺带一提：LLM按token计费的原理是把文本切成子词，中文一个字通常对应1到2个token，一段千字中文大概在1500-2000 token左右。所以一个多Agent任务跑下来，几十万token是稀松平常的事。这也是为什么我前面建议模型分级、结果缓存——成本控制不是抠门，是让Agent能活到明天。

评测集很容易被忽略，但它比日志还重要。我建议固定场景不少于20条，覆盖“单独调用工具”“多轮工具联动”“错误恢复”“上下文重写”四类典型情况。每次改Prompt、改工具描述，都跑一遍这20条，看工具调用准确率和最终任务完成率的变化。这比人工随机测试靠谱得多。

6. 实战串联：一个自动研究报告Agent的完整骨架

最后一个章节，我把前面讲的架构、工具、记忆、多Agent协作、部署一次性串起来，做一个自动研究报告生成Agent。这个Agent的需求很典型：用户丢过来一个话题，它要自动搜集公开资料、分析要点、生成一篇带结论的报告。

6.1 系统组成与职责划分

整个系统分成四层：

• Orchestrator（监督者）：接收用户话题，把它拆成三个子任务，依次派发给三个Worker。
• Researcher Worker：负责搜集公开网页资料，并把每次搜索得到的结论写入长期记忆。
• Analyst Worker：读取长期记忆中的搜索结果，提炼关键观点、时间线、争议点。
• Writer Worker：基于Analyst输出，生成结构化研究报告，报告格式定义为Pydantic模型。

数据流非常简单：user_request → orchestrator → plan → worker_chain → final_report。但每个Worker内部都可以有自己的ReAct子循环，比如Researcher会连续搜索多个关键词。

6.2 核心数据结构

from pydantic import BaseModel, Field from typing import List class ResearchPlan(BaseModel): topic: str = Field(description="研究报告主题") subtopics: List[str] = Field(description="需要调研的子话题列表") class ReportSection(BaseModel): title: str content: str key_facts: List[str] class FinalReport(BaseModel): topic: str summary: str sections: List[ReportSection] disclaimer: str = "本报告由AI Agent自动生成，重要决策请人工复核。"

强制让Writer输出结构化的FinalReport有两个好处：一是后续入库、渲染、比对都方便，二是结构约束能让模型“回答得更像报告而不是流水账”。

6.3 把记忆和工具串进来

系统运行到这个阶段，长期记忆存储的完整链路是这样的：

def run_researcher(topic: str): search_query = f"{topic} 最新进展 2025" for _ in range(3): result = call_web_search(search_query) if result.status == "ok": save_memory( session_id=topic, content=result.snippet, meta={"source": result.url, "created_at": now()} ) return summarize_search_result(result) return {"error": "连续三次搜索失败"}

这里的save_memory直接把检索回来的公开信息片段存入向量库。Analyst启动时会用search_memory(topic)拉取关联片段，跨过“必须把所有原始数据塞进上下文”的笨办法。

6.4 部署形态和扩展方向

这个Agent部署成HTTP服务后，用户只需要提交一个topic，轮询等待报告生成。生产环境可能会加一个任务队列，用一个后台worker消费消息，因为研究报告生成耗时较长，不适合HTTP请求同步等待。

往后的扩展方向也很明确：一是给Researcher增加更多信息源工具；二是在Analyst和Writer之间加一层人工审核，让Agent先产出草稿再由人确认；三是把FinalReport结构从Markdown升级成带最新时间戳的持久化数据，支持按主题回查历史报告。每一步扩展都是在前面的骨架上做加法，不需要推翻重写。

这篇写到这里，我自己最深的体会是：Agent开发真正难的不是让它“跑起来”，而是让它在多变、易错、有成本压力的现实环境里“可靠地跑”。把架构选型想清楚、工具边界守好、记忆分好层、协作协议定明白、部署可观测，剩下的就是反复用评测集打磨Prompt和工具描述。这套方法论是我踩了无数坑之后沉淀下来的，照着做，至少能少走一半弯路。