先交代个背景这段时间我在两个项目里分别做了两种形态的 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_urlPrompt 强调“只负责收集原始事实不做评价”。分析 Worker工具只有write_filePrompt 强调“基于给定的结构化数据输出洞察不要补写不存在的细节”。写作 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_retries3): 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 估算调用次数子问题生成5002001查询词改写300 x 5100 x 55网页读取清理2000 x 1010 x 1010单网页事实提取4000 x 10200 x 1010子问题综述3000 x 5600 x 55报告组装200010001合计约 6 万约 600032按当前主流模型定价粗算一次深入研究的模型成本大约在几块钱到十几块钱人民币的量级不同供应商差异很大具体以你拿到价目表为准。这个数字对个人开发者不算小做生产系统前一定要设置预算上限我习惯在 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 改动、工具接口变化都会反映在回归用例的输出漂移上。我这里有个更有效的方法给回归用例固定temperature0同时冻结模型版本保证对比基线稳定。等测试通过后再放开参数。5.3 验证引用真实性抽十个来源人工核验DeepResearch 最怕的就是“报告写得漂亮但全是编的引用”。我的方法是把引用验证做成独立环节每次完整研究结束后随机抽 10 个引用调用一次专门的验证提示词让模型对着原始来源检查报告里的关键数字和结论是否与之一致。抽查比例超过 20% 不一致时整个报告直接标记为“需人工复查”不出库。这个机制的原理很简单模型写报告时的幻觉和它做事实核查时的幻觉是两回事。让它面对一个具体数字追查来源它能发现我写作时生成的假数据准确率高得惊人。5.4 失败出口比成功路径更值得设计我见过太多 Agent 项目把 90% 的精力花在“如何成功”却几乎没想过“如何失败”。真实系统里搜索全挂、接口限流、上下文超限都是常态。一套可靠的设计必须包含这些出口达到最大步数时返回当前所有中间产物而不是报错。关键工具连续失败时降级为“只输出已知信息不检索新模式”。模型输出格式连续三次解析失败时停止调用把原始消息转给人工。这些出口的存在会让整个系统从“演示级”变成“可交付级”。5.5 最后一件事控制预期的边界基于我最近重构某智能体系统的感受最想叮嘱的一件事是不要把 Agent 设计成一个“万能机器”而是要设计成“可以在明确约束里完成任务的执行者”。任务边界、可验证的中间产物、明确的失败出口——这三样东西比任何花哨的 Prompt 都重要。这套代码骨架目前能稳定支撑的工作包括行业调研报告生成、竞品动态监控、无服务器架构方案预研甚至技术文档自动摘要。它距离一个理想中的“全自主智能体”还有很长的路但在实际业务中能在用户给定主题和约束条件下稳定输出可验证、可追溯的研究结论已经比绝大多数堆砌功能的原型有价值得多。如果你打算在现有项目里引入这套思路我的建议是从最小闭环开始先拿一个不需要多智能体协作的单一任务跑通规划-执行-反思循环再逐渐往上面加任务分解和并行调度。这一步一步的迭代能帮你少走很多弯路。