先交代个背景:这段时间我在两个项目里分别做了两种形态的 LLM 应用,一类是 DeepAgents——把大模型从“回答问题”变成“完成任务”的深度智能体;另一类是 DeepResearch——让模型自动完成资料检索、信息筛选、阅读归纳、报告生成全流程的深度研究智能体。两条路线拼在一起,基本就是一个能对外交付的智能应用骨架。这篇文章不打算停留在概念层,我会直接拆代码:DeepAgents 的任务规划、工具调用、多智能体协作怎么落地,DeepResearch 的查询重写、检索阅读、报告合成每一步为什么要这么写,并给出一套你可以直接拿来改的精简实现。适合已经跑通基础 LLM 调用、想往 Agent 方向进阶的开发者。
1. 智能体不是“API 串一串”:从三步闭环开始
很多人的第一个 Agent 是从“给模型加一个工具”开始的:把几个 API 封装成函数,塞进 Prompt,让模型自己选。这不算错,但你会发现它撑不过三个问题:多步骤任务怎么拆?中间步骤失败怎么办?模型绕圈子怎么打断?这三个问题本质上指向同一个答案——需要一个规划-执行-反思的闭环。
1.1 固定流程和动态规划的差别
固定流程像流水线:步骤 A 做完做步骤 B,每一步的结果都按预设格式传递。优点是稳定、好调试;缺点是一旦任务里出现“如果这一步查不到数据,就换一种查询方式”这类分支,代码会瞬间膨胀到不可维护。
动态规划则相反:模型每一步都基于当前状态决定下一步动作,动作集合是预先定义好的(调用工具、检索记忆、输出回答),但动作顺序完全由模型自己决定。这才是 Agent 和普通 Chain 的分水岭——控制流从代码里转移到了模型的推理里。
1.2 一个最简可运行的 Agent 循环
我直接给一版可以跑起来的最小循环,所有代码都只依赖一个llm_complete(messages)函数,你可以把它的底层实现换乘任何一家模型厂商的 SDK,不影响后续逻辑:
SYSTEM_PROMPT = """你是一个智能体。你有以下工具可用: - search_web(query): 搜索网络并返回结果片段 - read_url(url): 读取指定网页的正文内容 - finish(answer): 结束任务并给出最终答案 每次输出必须是一个 JSON 对象,格式为: {"type": "tool", "name": "<工具名>", "args": {...}} 或 {"type": "finish", "answer": "<最终答案>"} """ def run_agent(query: str, tools: dict, max_steps: int = 8) -> dict: messages = [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": query}, ] for step in range(max_steps): resp = llm_complete(messages) action = parse_action(resp) if action["type"] == "finish": return {"answer": action["answer"], "steps": step + 1} if action["type"] == "tool": if action["name"] not in tools: messages.append({ "role": "user", "content": f"工具 {action['name']} 不存在,请重新选择。", }) continue try: tool_result = tools[action["name"]](**action["args"]) except Exception as e: tool_result = f"工具调用失败: {e}" messages.append({"role": "assistant", "content": resp}) messages.append({ "role": "user", "content": f"工具 {action['name']} 返回下列结果,请根据结果决定下一步:\n{tool_result}", }) continue # 模型输出了非法结构,给一次纠错机会 messages.append({ "role": "user", "content": "输出格式不对,请严格按照 JSON 格式输出。", }) return {"error": "max steps exceeded", "trace": messages}这段代码里有几个容易出错但我故意保留的细节:
第一,工具结果我是作为"user"角色塞回对话的,而不是用"tool"角色。原因很简单:很多模型厂商的 SDK 对"tool"角色有严格的配对要求,尤其是 function call 格式不统一时,最容易报错的就是这里。用"user"角色承载工具结果,通用性最好,只要在文本里写清楚“这是工具返回结果”,模型完全能理解。
第二,parse_action一定要做容错。模型偶尔会在 JSON 前后加 Markdown 代码块标记,或者把单引号当双引号用。我的做法是先尝试json.loads,失败后用正则提取最外层大括号再解析一次,再不行才走到“重新输出”分支。
第三,continue的使用不能省。工具调用完成后必须回到循环头部重新请求模型,否则你会陷入“模型思考了但没动作”的死循环。
1.3 为什么这个循环能成立
这个循环能跑通,核心在于模型本身具备了“上下文推理”能力:它能从工具返回的结果里判断当前信息够不够,不够就继续调用工具,够了就输出finish。也就是说,你不需要手写“如果搜索结果少于三条就第二次搜索”这种规则,模型会自己判断。
但这也带来第一个工程问题:模型会“想太多”。明明搜一次就够的任务,它可能会搜三次、五次。所以max_steps不是摆设,它是整套系统最重要的安全阀。我习惯把它设成 6-10,宁可做不完,也不能让它无限循环烧钱。
2. DeepAgents 任务分解与多智能体协作的代码骨架
最小循环只能解决“单个智能体做一件中等复杂的事”。真实业务里任务往往是复合的:比如“调研一个行业的头部玩家,输出对比表格和总结”,你让一个智能体从头做到尾,要么上下文窗口被撑爆,要么任务边界不清晰导致输出很水。这时候就需要任务分解和多智能体协作。
2.1 为什么不能把 Prompt 写得越来越长
处理复合任务最朴素的想法是给模型写一篇“史诗级 Prompt”,把所有要求塞进去。我试过这种方式,结果很糟糕:
- 上下文越长,模型对关键指令的遵循率越低,尤其是埋在中间的约束,经常被无视。
- 单次输出长度有限,一个 5000 字要求的任务,模型写到最后明显在凑字数。
- 中间环节出错无法定位,你不知道是检索环节的数据不行,还是写作环节的理解偏了。
任务分解的本质,是把大问题切成若干可独立验证的小问题,每个小问题由独立上下文执行,最后汇总。代价是增加了结构复杂度,但换来的是每一步结果都可检查、可重试。
2.2 任务规划器:让模型输出一张任务图
我一般用一个TaskPlanner来做第一步拆分。它接收用户目标,输出一组带依赖关系的任务,每个任务有独立的描述和验收标准:
class Task: def __init__(self, task_id: str, description: str, dependencies: list[str], acceptance: str): self.id = task_id self.description = description self.dependencies = dependencies self.acceptance = acceptance self.result = None self.status = "pending" class TaskPlanner: def __init__(self, llm): self.llm = llm def plan(self, goal: str) -> list[Task]: schema_hint = """请把用户目标拆解为多个子任务,任务之间如果有先后依赖请明确。 输出格式: ```json { "tasks": [ {"id": "t1", "description": "xxx", "dependencies": [], "acceptance": "yyy"}, {"id": "t2", "description": "xxx", "dependencies": ["t1"], "acceptance": "yyy"} ] }严格要求:每次只能输出一个 JSON 代码块,不要写解释。""" raw = self.llm.chat(goal + "\n" + schema_hint) parsed = extract_json_block(raw) return [Task(**t) for t in parsed["tasks"]]
这里的关键不是 JSON 解析,而是 `acceptance` 字段。任务规划时让模型顺便写清楚“什么算完成”,这一步会直接决定后续每个子智能体干活的质量。比如“调研竞品”这个任务,验收标准如果写“列出主要竞品”,那结果大概率只有三两行;如果写“每个竞品需要包含产品名称、融资情况、目标用户、近半年动态,并以表格输出”,结果就会扎实很多。 ### 2.3 执行器:任务调度的两种模式 任务拆完之后需要一个调度器来执行。依赖图谱复杂的用拓扑排序,但我实际项目中 80% 的情况是线性链或浅层并行,所以我通常只保留两个模式: - 串行模式:任务依赖前一个任务的结果,比如“先收集数据,再清洗数据,再写分析”。 - 并行模式:任务之间无依赖,比如同时调研三个不同竞品,最后汇总。 调度器核心逻辑如下: ```python class Executor: def __init__(self, worker_map: dict): self.worker_map = worker_map # {"worker_type": Worker} self.results = {} def run(self, tasks: list[Task]) -> list[Task]: remaining = tasks[:] while remaining: batch = [t for t in remaining if all(dep in self.results for dep in t.dependencies)] if not batch: raise RuntimeError("任务依赖存在循环,无法继续推进") run_in_parallel(batch, self._execute_one) for t in batch: self.results[t.id] = t.result # 用验收标准做一次自检 if not t.acceptance or len(str(t.result)) < 50: t.status = "needs_review" remaining = [t for t in remaining if t not in batch] return tasks def _execute_one(self, task: Task): worker = self.worker_map[task.worker_type] task.result = worker.run(task.description, self.results) task.status = "done"我特别加了一个粗糙的“自检”:如果模型结果太短,达不到验收标准,就把任务标记成needs_review,不会直接把结果喂给下游任务。这是从真实项目里学到的教训——LLM 输出“简洁”和“偷懒”之间没有可靠区分,必须设一条机械底线。
2.4 子智能体的个性化上下文
多智能体不意味着要部署多个模型实例,它只是在不同上下文里调用同一个模型。我在Executor里给每个 Worker 配了独立的 System Prompt 和专用工具集合,效果等同于一群不同“人设”的专家:
- 数据采集 Worker:工具只有
search_web和read_url,Prompt 强调“只负责收集原始事实,不做评价”。 - 分析 Worker:工具只有
write_file,Prompt 强调“基于给定的结构化数据输出洞察,不要补写不存在的细节”。 - 写作 Worker:工具为空,Prompt 强调“把分析结果组织成 800 字以内的报告,保留引用来源”。
这样做的直接好处是隔离:采集阶段模型不会被“你得写好看点”的指令干扰,写作阶段模型也不会因为上下文里有太多原始 URL 而分心。
3. DeepResearch 代码逐段拆解:一份研究报告是怎么生成的
DeepResearch 是 DeepAgents 思路在“信息密集型任务”上的特化。普通 Agent 的难点是控制流,DeepResearch 的难点在于信息质量:搜索结果怎么选、读到的东西怎么去重、多个来源观点冲突时以谁为准。我拆解一下我自己实现的核心管线。
3.1 从用户查询到研究子问题
第一步不是搜索,而是把用户那句模糊的话变成几个明确的研究问题。比如用户说“帮我研究一下无服务器计算的现状和趋势”,直接拿这句话去搜索,结果会又杂又浅。我做的第一层处理是查询改写加子问题生成:
def plan_research(query: str) -> list[str]: prompt = f"""你是一名研究助理。请把用户的查询拆解为 3-6 个需要分别研究的具体问题。 要求: 1. 每个问题必须可以独立搜索 2. 问题之间尽量避免重叠 3. 问题覆盖:现状、关键玩家、发展趋势、潜在风险 用户查询:{query} 输出 JSON 列表,格式:["问题1", "问题2", ...] """ raw = llm_complete([{"role": "user", "content": prompt}]) return json.loads(extract_json_block(raw))注意看,我在 prompt 里写死了四个覆盖维度。因为用户说“现状和趋势”,如果放任模型自由发挥,它大概率漏掉“关键玩家”和“潜在风险”。DeepResearch 的优势就在这里:结构化的研究框架可以让模型的搜索行动更有章法。
3.2 搜索与重写:同一问题搜三次
拿到子问题之后直接去搜索,效果一般。原因有两个:搜索接口对长句不友好;同一个问题换几个说法能得到完全不同的一批结果。
我的做法是对每个子问题做三种形态的查询词:
def rewrite_queries(question: str) -> list[str]: prompt = f"""把下面的研究问题改写成三个用于搜索引擎的关键词组合: - 版本A:精确术语版,使用原文专业词汇 - 版本B:通俗表达版,适合检索新闻和入门资料 - 版本C:趋势分析版,加入“2024”“趋势”“报告”等限定词 问题:{question} 输出 JSON 列表。""" raw = llm_complete([{"role": "user", "content": prompt}]) return json.loads(extract_json_block(raw))三个版本分别执行搜索,把结果 URL 合并、去重、按域名权重排序,然后一次性送进阅读环节。这样即使模型对某个领域术语不熟悉,也能靠不同风格的查询词兜底。
3.3 网页读取与内容清洗
搜索结果里大量 URL 是无效的:有的打不开,有的是纯导航页,有的正文和页面标题完全无关。读取环节必须做清洗,这个环节写起来最枯燥,但直接决定最终报告质量。
我维护了一个简单的清洗管线:
def clean_page(html: str, url: str) -> str: # 第一步:暴力去掉所有 script/style/iframe 标签 text = strip_tag_blocks(html, ["script", "style", "iframe", "noscript"]) # 第二步:按可见文本提取 text = extract_visible_text(text) # 第三步:合并连续空行,限制单页最长长度 text = normalize_whitespace(text) text = truncate_to_chars(text, 8000) return f"来源: {url}\n内容:\n{text}"这里的truncate_to_chars(8000)是硬性限制。一篇文章的正文动辄两三万字,如果全塞进上下文,模型注意力会被稀释。截断到 8000 字符之后,配合下一步的“信息提取”,基本能保住每篇网页最核心的信息。
3.4 单来源信息提取:一个极其有用的 trick
清洗完的页面不能直接拿去写报告,它还是太杂。我会让模型先对每个来源做一次“定向提取”,把页面内容压缩成和当前子问题相关的几条事实:
def extract_facts(question: str, page_content: str) -> list[str]: prompt = f"""研究问题:{question} 下面是一篇网页内容。请提取与问题直接相关的、可验证的事实性信息,一条一行,每行不超过 60 字。 如果内容与问题无关,请只输出:N/A 网页内容: {page_content} """ raw = llm_complete([{"role": "user", "content": prompt}]) return [line.strip("- ") for line in raw.splitlines() if line.strip() and line.strip() != "N/A"]这一步的收益非常大:一方面把每个来源压成几条事实,后续上下文占用直接降了一个数量级;另一方面模型在“提取事实”这个任务上的幻觉率比“直接写总结”低得多——你让它一条条摘信息,它不太敢编。这算是 DeepResearch 里性价比最高的一个设计。
3.5 信息汇总与冲突消解
多个来源都提取完了,就要面对信息冲突。比如一个来源说市场年增长 30%,另一个说 35%,报告不能两个数字并列,要给出判断。
我给模型设计了这样的综合指令:
def synthesize_section(question: str, facts_by_source: dict) -> str: source_text = "" for url, facts in facts_by_source.items(): source_text += f"[来源 {url}]\n" + "\n".join(facts) + "\n\n" prompt = f"""研究问题:{question} 以下是不同来源提取出的事实信息。请写一段不超过 500 字的综述,要求: 1. 对数量、占比等信息,优先采信来源更权威或信息更新鲜的数据,并做交叉说明 2. 不同来源有冲突时,同时声明两个来源的差异,用“据某来源……另有来源显示……”呈现 3. 只使用给定来源中出现过的信息,不得补充未提供的数字 {source_text} """ return llm_complete([{"role": "user", "content": prompt}])这个 prompt 里最重要的一句是“只使用给定来源中出现过的信息”。这是 DeepResearch 对抗幻觉的主要手段,你不需要模型有多强的领域知识,你需要的是它老老实实当一个二手信息的整理者。
3.6 报告组装与引用追踪
子问题综述写完之后,还剩最后一件事:把它们组装成完整报告,并且每个结论都挂上来源。我的做法比较朴素,在综述的每一段结尾,人工插入该段用到的来源链接:
def build_report(title: str, sections: list[Section], references: dict) -> str: report_lines = [f"# {title}", ""] for sec in sections: report_lines.append(f"## {sec.heading}") report_lines.append(sec.text) if sec.refs: report_lines.append("") report_lines.append("来源:") for i, ref in enumerate(sec.refs, 1): report_lines.append(f"[{i}] {ref}") report_lines.append("") return "\n".join(report_lines)关键点在于Section里的refs不是最后统一挂的,而是在synthesize_section之前就绑定好的——我让提取环节保留每个事实对应的来源 URL,综述阶段模型写句子时不带链接,我在后处理阶段按语句里提到的主题词反向映射回来源。这样即使模型在综述里没有显式标注引用,报告也能自动带上可追溯的来源。
4. 工程化实战:上下文膨胀、失控重试与成本预算
代码能跑通和能上线是两码事。DeepAgents 和 DeepResearch 这类系统,在真实环境里暴露的问题比功能逻辑多得多。我挑三个被问得最多的工程问题展开。
4.1 上下文膨胀:对话越长,模型越笨
Agent 每调用一次工具,就要把之前所有消息重新送进模型。循环十轮之后,上下文可能已经包含几万 token,其中大部分是早期搜索得来的低价值信息。模型面对一坨又臭又长的历史,要么遗漏关键指令,要么干脆开始复读。
我的处理策略是三层:
- 硬截断:只保留最近的 N 轮消息。早期步骤已经没必要再看了,模型需要的是当前状态。
- 摘要替代:自定义一个压缩函数,把超过阈值的中间轮次用一句话概括,塞回上下文顶部。比如“已完成初步搜索,获得 3 个相关来源,当前在读取第二个来源的正文”。
- 结构化状态:工具结果不直接用文本塞回,而是先经过一轮“提取关键状态”的压缩,再进入上下文。
这三层可以同时叠用,我通常在第四轮工具调用之后开启摘要替代,保证上下文长期稳定在 8k token 以内。
4.2 失控重试:模型不会因为重试而变得更聪明
新手最爱写一个重试逻辑:调用失败就重试三次,超时就重试三次。问题在于,模型接口超时重试没问题,但如果是因为工具参数错误导致失败,重试多少次都一样——模型会生成同样的错误参数。
我用的重试策略是给反馈式纠错:
def call_with_retry(fn, *args, max_retries=3): for attempt in range(max_retries): try: return fn(*args) except InvalidParamError as e: last_error = f"参数错误:{e},请检查参数名和类型后重新调用" # 把错误信息和修正指令塞回上下文 return last_error raise RuntimeError("重试次数耗尽")注意,InvalidParamError分支我返回的不是抛异常,而是把错误消息作为工具结果反馈给模型。这相当于告诉模型“你刚才那步干错了,这是为什么”,模型通常会自己修正——比你无脑重试有效得多。
4.3 Token 预算:一场深度研究的真实账单
DeepResearch 类应用模型消耗极高,很多第一次做的人会被月底账单吓到。我算过一笔账,一次 5 个子问题、每个子问题 3 个查询词、每个查询读取 2 个网页的深度研究,token 消耗大概是这样:
| 环节 | 输入 token 估算 | 输出 token 估算 | 调用次数 |
|---|---|---|---|
| 子问题生成 | 500 | 200 | 1 |
| 查询词改写 | 300 x 5 | 100 x 5 | 5 |
| 网页读取清理 | 2000 x 10 | 10 x 10 | 10 |
| 单网页事实提取 | 4000 x 10 | 200 x 10 | 10 |
| 子问题综述 | 3000 x 5 | 600 x 5 | 5 |
| 报告组装 | 2000 | 1000 | 1 |
| 合计 | 约 6 万 | 约 6000 | 32 |
按当前主流模型定价粗算,一次深入研究的模型成本大约在几块钱到十几块钱人民币的量级(不同供应商差异很大,具体以你拿到价目表为准)。这个数字对个人开发者不算小,做生产系统前一定要设置预算上限,我习惯在 pipeline 入口加一个预算对象:
class Budget: def __init__(self, limit_usd: float): self.limit_usd = limit_usd self.cost_usd = 0.0 def charge(self, tokens_in: int, tokens_out: int) -> bool: self.cost_usd += estimate_cost(tokens_in, tokens_out) return self.cost_usd <= self.limit_usd每次调用模型之前charge一次,返回False就立刻终止研究并输出当前已获得的中间结果。宁可交付一份不完整但有来源的报告,也比烧光预算后空手而归好。
5. 跑通整套系统后,最想留给你的调试清单
最后给一套调试清单。这一节的每一条,都是从实际运行的报错和劣质输出里倒推出来的,比任何架构图都实在。
5.1 日志必须记录“模型在想什么”
Agent 类应用的调试几乎全依赖日志。我建议日志至少分三层:
- 思考过程层:记录模型每一步的原始输出,不管是工具调用指令还是最终回答。
- 工具调用层:记录哪个工具、什么参数、执行耗时、返回结果摘要。
- 状态视图层:记录每一步结束时的核心状态,比如已收集多少来源、当前子问题进度。
没有这些日志,一旦最终报告质量不达标,你根本不知道是搜索没搜到,还是信息提取丢了,还是综述环节写崩了。
5.2 用最小用例做回归测试
LLM 应用也能做回归测试,但思路和传统软件不一样。与其断言模型输出某一个字,不如用一个“输入不变、输出不变”的最小用例集合,每天跑一次,肉眼对比输出质量走势。模型更新、Prompt 改动、工具接口变化,都会反映在回归用例的输出漂移上。
我这里有个更有效的方法:给回归用例固定temperature=0,同时冻结模型版本,保证对比基线稳定。等测试通过后再放开参数。
5.3 验证引用真实性:抽十个来源人工核验
DeepResearch 最怕的就是“报告写得漂亮但全是编的引用”。我的方法是把引用验证做成独立环节:每次完整研究结束后,随机抽 10 个引用,调用一次专门的验证提示词,让模型对着原始来源检查报告里的关键数字和结论是否与之一致。抽查比例超过 20% 不一致时,整个报告直接标记为“需人工复查”,不出库。
这个机制的原理很简单:模型写报告时的幻觉和它做事实核查时的幻觉是两回事。让它面对一个具体数字追查来源,它能发现我写作时生成的假数据,准确率高得惊人。
5.4 失败出口比成功路径更值得设计
我见过太多 Agent 项目把 90% 的精力花在“如何成功”,却几乎没想过“如何失败”。真实系统里,搜索全挂、接口限流、上下文超限都是常态。一套可靠的设计必须包含这些出口:
- 达到最大步数时,返回当前所有中间产物,而不是报错。
- 关键工具连续失败时,降级为“只输出已知信息不检索新模式”。
- 模型输出格式连续三次解析失败时,停止调用,把原始消息转给人工。
这些出口的存在,会让整个系统从“演示级”变成“可交付级”。
5.5 最后一件事:控制预期的边界
基于我最近重构某智能体系统的感受,最想叮嘱的一件事是:不要把 Agent 设计成一个“万能机器”,而是要设计成“可以在明确约束里完成任务的执行者”。任务边界、可验证的中间产物、明确的失败出口——这三样东西比任何花哨的 Prompt 都重要。
这套代码骨架目前能稳定支撑的工作包括:行业调研报告生成、竞品动态监控、无服务器架构方案预研,甚至技术文档自动摘要。它距离一个理想中的“全自主智能体”还有很长的路,但在实际业务中,能在用户给定主题和约束条件下稳定输出可验证、可追溯的研究结论,已经比绝大多数堆砌功能的原型有价值得多。
如果你打算在现有项目里引入这套思路,我的建议是从最小闭环开始:先拿一个不需要多智能体协作的单一任务跑通规划-执行-反思循环,再逐渐往上面加任务分解和并行调度。这一步一步的迭代,能帮你少走很多弯路。