你有没有遇到过这种情况想让一个模型同时搞定写脚本改bug写使用说明三件事结果它顾此失彼代码里夹着解释文字文案里又带着代码片段最后两边都不像样。我一开始也是这么干的直到我把任务真正拆开才意识到多智能体的价值不是炫技而是把一个复杂需求拆给不同岗位去并行思考。这篇文章就带你从零做一个简单的AI任务调度助手基于LangChain生态里的 Agent 和图形编排能力把调度者、执行者、质检员跑通一条完整链路。整篇文章适合两类人一类是刚接触 LangChain想知道 Agent 怎么串起来另一类是被单一大模型一个指令干到底折磨过想看看多智能体到底怎么落地的新手。我不会把代码堆成一堵墙而是会把每个设计决策的原因讲清楚顺手把调试时踩过的坑也记下来。1. 先聊聊为什么单个大模型干不好任务调度1.1 一个模型身兼数职指令会互相干扰你可能会问我直接给一个大模型写一段很长的提示词让它先写代码再写说明最后自查不也是一种调度吗表面上可以实际跑几次就发现问题了。大模型没有真正的工作记忆它只是根据上下文里的所有内容生成下一个 token。当代码任务和文案任务挤在同一段上下文中模型很容易把两种风格混在一起写代码时突然冒出一句这段逻辑很清晰写文案时又插入代码块。还有更麻烦的如果某个子任务执行失败你要回退重跑单一大模型往往无法判断到底该重写哪部分只能整段重新生成。我自己遇到最典型的一个场景是让模型写一个Python脚本统计CSV缺失值并同时生成说明文档。模型确实给出了完整回复但脚本里的注释是给非技术人员看的营销文案风格文档里又贴了一大段源码。分开看好像每个部分都对合在一起就是没法用。这就是角色打架。1.2 多智能体不是玄学本质是把流程拆给不同大脑多智能体的思路其实很像现实中的团队分工调度者负责接收需求、拆解任务、判断下一步交给谁执行者各有专长一个写代码一个写文案互不干扰质检员最后检查产物是否满足要求不合格就打回重来。每个智能体只关注自己的窄范围提示词短了指令冲突少了上下文也更干净了。从工程角度看多智能体的价值在于可控性和可观测性。单一大模型的调用像一个黑盒你很难知道它内部到底先做了什么而多智能体系统里每个节点是什么、输入什么、输出什么都是你自己定义的。哪怕中间某个 Agent 崩了你也知道是哪一步崩的改起来也快得多。在 LangChain 生态里实现多智能体一般有两种路线一种是让一个 Agent 持有其他 Agent 的工具自己做路由另一种是用 LangGraph 这类编排框架把每个智能体做成图里的一个节点节点之间用条件边连接。我这个项目选的是后者因为条件路由更直观而且对初学者来说图结构比隐藏的 Agent 工具链更容易调试。2. 环境准备与LangChain生态里的现实坑2.1 安装与版本选择LangChain不是装一个就完事很多新手第一次装 LangChain 时会被一堆包搞晕。现在 LangChain 已经不是单个库而是拆成了很多配套包。如果你只装了langchain会发现很多 Agent 相关功能根本用不了。我这个项目里实际需要这么几个pip install langgraph langchain langchain-openai langchain-community langchain-corelanggraph是负责编排多智能体流程的核心langchain-openai是让 LangChain 能调用 OpenAI 兼容格式的接口langchain-community里装着社区维护的一堆工具集成。不装langchain-core可能会缺基础类型虽然有些包会自动带上但显式装一遍能省掉不少版本冲突的烦恼。版本方面一定要留个心眼LangGraph 从 0.2 开始 API 有过一些调整有些教程里写的老写法会直接报错或者被废弃。安装的时候可以在包名后给定版本范围例如pip install langgraph0.2 langchain0.3提示如果后面发现某个函数调用报AttributeError先别急着怀疑自己代码很可能是某个依赖被升级到了不兼容的新版本。检查一下pip list里的版本再对照 langgraph 的变更说明基本能定位。2.2 模型接入随便一个大模型API都能当大脑这个项目里所有智能体本质上都是对同一个大模型 API 的多次调用只不过每次用的系统提示词不一样。所以我只需要一个能通过OpenAI 兼容接口访问的模型服务就行不限定具体厂商。用环境变量管理密钥和地址这样代码换模型时不用改动逻辑部分import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() llm ChatOpenAI( base_urlos.getenv(LLM_BASE_URL), api_keyos.getenv(LLM_API_KEY), modelos.getenv(LLM_MODEL_NAME), temperature0.1, )把LLM_BASE_URL、LLM_API_KEY、LLM_MODEL_NAME写到.env文件里就行。我用的是一个中规模的通用大模型上下文长度中等响应速度还不错。如果模型支持结构化输出调度节点的准确率会高不少如果模型不支持后面我会说怎么用提示词约束加解析兜底来处理。这里有一个跟新手相关的关键点temperature我设置成 0.1因为调度节点的任务是对接路由决策不需要创意需要确定性。写文案的 Agent 可以把温度调高到 0.4 甚至 0.7。同一份代码里不同节点使用不同温度完全没问题只要在定义节点时给不同的 LLM 实例就行。3. 这个任务调度助手的设计调度者、执行者、质检员3.1 状态设计是第一步先想清楚每个智能体之间传什么搭多智能体之前千万别直接写代码第一个要考虑的是状态State。多智能体运行相当于一个状态机每个节点读入状态执行完毕后又把新状态传给下一个节点。状态里放什么字段直接决定了每个 Agent 到底能看到多少信息。我这套助手的核心字段如下字段类型作用querystr用户的原始需求next_actionstr调度节点决定的下一步动作assignmentstr调度器给执行者的子任务说明code_resultstr代码 Agent 的工作产物doc_resultstr文案 Agent 的工作产物inspect_feedbackstr质检员的审查意见step_countint当前已经跑了多少个节点防止死循环有些字段你可能看不出用途但它们在调试时特别重要。比如step_count如果调度器出了问题会在节点之间疯狂往返没有这个计数器整个程序就跑死了也不自知。我会在调度节点的提示词里强制要求当 step_count 大于 3 时必须输出final动作。还要考虑一个问题code_result和doc_result都是直接覆盖还是保留历史我这个 demo 里选择直接覆盖因为每次重新生成时质检员的意见已经通过assignment传给了执行者不需要保存多轮历史状态。如果你要迭代很多次最好改成列表类型保留每一步的产物方便追溯和对比。3.2 三条角色链路与回退机制我设计了三类节点调度节点dispatcher读用户 query 和已有产出物输出 JSON决定下一步是去写代码、写文案还是直接收尾。执行节点代码 Agent 负责写脚本并运行验证文案 Agent 负责写说明文档、使用指南或解释段落。质检节点inspector汇总已完成的产出判断质量是否达标不达标就带着反馈意见回到调度节点重新指派。这个流程是一个典型的规划-执行-审查环。第一次跑通的时候最容易困惑的是为什么代码 Agent 执行完还要再回到调度节点而不是直接进质检节点因为用户的需求可能是先写代码再写文案也可能是只需要文案。如果代码 Agent 执行完就直接进质检那一份没有文案的产出根本没法审查。所以我的图里执行节点回环到调度节点由调度节点判断下一个该做什么当所有子任务都完成调度节点才会给出final动作进入质检环节。质检环节也支持回退如果质检不通过它会带着inspect_feedback回到调度节点调度器把反馈写进新的assignment再指派对应的执行者改一轮。这里需要留意的是回退次数必须有限制否则一个永远改不对的任务会把整个流程拖死。我的做法是把回退次数换算进step_count超过阈值就让质检员网开一面直接放行。4. 核心代码逐段拆解状态图、路由与Agent实现4.1 基础设施状态类型、模型初始化与Python执行工具代码从定义状态类型开始。我使用TypedDict来描述状态结构这样写节点函数时能获得 IDE 提示不容易因拼写错误翻车from typing import TypedDict class AssistantState(TypedDict): query: str next_action: str assignment: str code_result: str doc_result: str inspect_feedback: str step_count: int final_answer: str然后是 Python 执行工具。代码 Agent 生成脚本后不能光停在我写好了这一步最好真跑一遍看看有没有语法错误。我给 Agent 配一个自定义工具用exec执行传入的代码片段并捕获输出import io import contextlib from langchain_core.tools import tool tool def run_python_snippet(code: str) - str: 在本地进程内执行一段Python代码返回标准输出与错误信息。 仅用于demo验证短脚本生产环境务必放到隔离容器里。 stdout_buf io.StringIO() stderr_buf io.StringIO() try: with contextlib.redirect_stdout(stdout_buf), contextlib.redirect_stderr(stderr_buf): exec(code, {__name__: __main__}) except Exception as exc: return f运行异常: {type(exc).__name__}: {exc}\n标准输出:\n{stdout_buf.getvalue()} return f标准输出:\n{stdout_buf.getvalue()}\n标准错误:\n{stderr_buf.getvalue()}警告直接在本地exec运行代码是很危险的做法尤其当代码 Agent 的输出不可信时。这个 demo 只是教学用我建议你在沙箱容器里跑。即使如此代码 Agent 的执行结果也要当作外部输出看待不要盲目信任。4.2 调度节点的实现负责拆解与路由调度节点本身不需要工具它就是一次带结构化输出要求的 LLM 调用。我会让模型返回一个 JSON包含next_action和assignment两个字段。调度器的系统提示词需要把可选项讲清楚code表示去写代码doc表示去写文案final表示所有产出已完成准备交给质检。同时还要提醒模型参考已有的code_result和doc_result避免重复劳动。from langchain_core.output_parsers import JsonOutputParser from langchain_core.prompts import ChatPromptTemplate DISPATCH_PROMPT 你是任务调度器负责拆解用户需求并安排执行者。 可选动作 - code: 让代码Agent去编写并验证Python代码 - doc: 让文案Agent去撰写说明文档或文案 - final: 所有必要产物都已生成进入质检环节 用户需求{query} 已有代码产物{code_result} 已有文案产物{doc_result} 当前执行计数{step_count} 规则 1. 如果用户需要代码而代码产物为空优先选择code。 2. 如果用户需要说明文档而文案产物为空选择doc。 3. 如果步骤超过3轮不管怎样都选择final。 4. 输出JSON格式{next_action: code, assignment: 给执行者的子任务描述, reason: 简单说明理由} 只输出JSON不要输出其他内容。 parser JsonOutputParser() def dispatch_node(state: AssistantState) - dict: if state.get(step_count, 0) 3: return { next_action: final, assignment: 所有任务已经执行过多轮请质检后直接汇总, step_count: state.get(step_count, 0) 1, } prompt ChatPromptTemplate.from_template(DISPATCH_PROMPT) chain prompt | llm | parser result chain.invoke({ query: state[query], code_result: state.get(code_result, 还没有代码产物), doc_result: state.get(doc_result, 还没有文案产物), step_count: state.get(step_count, 0), }) return { next_action: result.get(next_action, final), assignment: result.get(assignment, ), step_count: state.get(step_count, 0) 1, }这个实现里有一个很容易忽略但很重要的点我利用step_count作为硬性逃生阀即使大模型输出的next_action一直不老实超过 3 轮也会被强行改成final。4.3 执行者节点代码Agent和文案Agent代码 Agent 用 LangChain 的create_react_agent创建让它能调用run_python_snippet工具来实际执行生成的脚本。系统提示词限定它的职责范围只写代码、只验证代码不要写说明文档代码需要直接能跑并给出必要注释。from langchain.agents import create_react_agent from langchain_core.prompts import PromptTemplate CODE_AGENT_TMPL 你是一名Python工程师。你只负责根据任务描述编写可运行的Python代码 并用run_python_snippet工具运行验证。不要把任务描述之外的文案混进来。 任务描述 {assignment} 要求 - 代码要简洁包含必要的注释。 - 先调用run_python_snippet运行一次即使报错也要把错误信息原样反馈。 - 最终回复时先给完整代码再附一行执行结果摘要。 code_agent_prompt PromptTemplate.from_template(CODE_AGENT_TMPL) code_agent_runnable create_react_agent(llm, [run_python_snippet], promptcode_agent_prompt)执行节点里的工作就是把这个 Agent 跑起来并把最终输出塞回状态里def code_agent_node(state: AssistantState) - dict: result code_agent_runnable.invoke({assignment: state.get(assignment, )}) return {code_result: result[output]}文案 Agent 其实不需要工具我用一个普通的链来生成文案DOC_PROMPT 你是文档撰写员。你不修改代码只根据已有材料撰写用户能看懂的说明文字。 要求语气清晰、步骤明确、不要空话。 已有代码产物 {code_result} 任务描述 {assignment} 请输出说明文档正文。 def doc_agent_node(state: AssistantState) - dict: prompt ChatPromptTemplate.from_template(DOC_PROMPT) chain prompt | llm result chain.invoke({ code_result: state.get(code_result, 暂无代码产物), assignment: state.get(assignment, ), }) return {doc_result: result.content if hasattr(result, content) else str(result)}可以看到文案 Agent 的输入里包含了code_result。这是多智能体协作的关键上一个角色的输出成为下一个角色的输入上下文。如果把这一步省掉文案 Agent 就成了盲写写出来的说明和实际代码可能对不上。4.4 质检节点与最终汇总质检节点的职责是站在用户视角看整体结果。它对代码和文案做一次通读输出两个判断字段pass或fail以及反馈意见。实现上依然是给 LLM 一个结构化的输出要求。INSPECT_PROMPT 你是质量审查员。请检查以下代码和文案是否匹配、是否完整、是否有明显错误。 代码产物 {code_result} 文案产物 {doc_result} 用户需求 {query} 输出JSON格式 {verdict: pass 或 fail, feedback: 你的具体审查意见} 只输出JSON。 def inspect_node(state: AssistantState) - dict: prompt ChatPromptTemplate.from_template(INSPECT_PROMPT) chain prompt | llm | JsonOutputParser() result chain.invoke({ code_result: state.get(code_result, ), doc_result: state.get(doc_result, ), query: state[query], }) verdict result.get(verdict, fail) feedback result.get(feedback, ) return {next_action: verdict, inspect_feedback: feedback}汇总节点的任务就简单多了把代码和文案拼成最终答案def aggregate_node(state: AssistantState) - dict: final_answer ( f### 代码部分\n\n{state.get(code_result, )}\n\n f### 文案部分\n\n{state.get(doc_result, )}\n\n f### 质检意见\n\n{state.get(inspect_feedback, )} ) return {final_answer: final_answer}4.5 把图组装起来并跑一个完整例子现在到了最关键的一步用 LangGraph 把这些节点连成一张图。有四个节点需要分别处理dispatch、code_agent、doc_agent、inspect、aggregate。连接规则是dispatch根据next_action无条件选择去code_agent、doc_agent还是inspect。code_agent结束后回到dispatch让它继续判断是否需要doc_agent。doc_agent结束后同样回到dispatch。inspect根据next_action判断是回dispatch继续修改还是去aggregate收尾。from langgraph.graph import StateGraph, END graph StateGraph(AssistantState) graph.add_node(dispatch, dispatch_node) graph.add_node(code_agent, code_agent_node) graph.add_node(doc_agent, doc_agent_node) graph.add_node(inspect, inspect_node) graph.add_node(aggregate, aggregate_node) graph.set_entry_point(dispatch) graph.add_conditional_edges( dispatch, lambda state: state.get(next_action, final), { code: code_agent, doc: doc_agent, final: inspect, }, ) graph.add_edge(code_agent, dispatch) graph.add_edge(doc_agent, dispatch) graph.add_conditional_edges( inspect, lambda state: aggregate if state.get(next_action) pass else dispatch, { aggregate: aggregate, dispatch: dispatch, }, ) graph.add_edge(aggregate, END) app graph.compile()然后就可以用一句话来测试整个调度助手result app.invoke({ query: 帮我写一个Python脚本读取data.csv并统计每列缺失值数量然后用中文写一段脚本使用说明。, next_action: , assignment: , code_result: , doc_result: , inspect_feedback: , step_count: 0, final_answer: , }) print(result[final_answer])第一次跑通的时候你会看到 LangGraph 打印出一串节点执行日志比如dispatch - code_agent - dispatch - doc_agent - dispatch - inspect - aggregate这个顺序正是调度器自己规划出来的。这种感觉挺奇妙的——你并没有在代码里写死流程而是让模型在几个选项之间做选择最终走出一条合理的执行路径。5. 第一次跑往往不会顺利我踩过的三个实际问题5.1 路由结果不够结构化带来的解析危机第一次搭建时我让调度器直接输出文本然后在代码里用if code in output这种办法判断下一步。听起来很省事实际上一团糟。模型可能输出我们应该先 code然后 doc也可能输出CODE大小写和格式都不稳定判断逻辑越写越脏。后来我改成用JsonOutputParser明确要求只输出 JSON。结果在小模型上又出现一个问题模型不遵守指令输出的 JSON 前面多了一句解释性文字或者直接把引号写弯了解析器照样报错。解决思路有两个一是换用支持response_format{type: json_object}的模型在 API 层面强制结构输出二是做好幂等的解析兜底用正则把第一个{到最后一个}之间的内容截出来再解析。我现在的做法是写一个小的解析函数如果直接解析失败就尝试提取 JSON 片段import json import re def safe_parse_json(text: str) - dict: text text.strip() try: return json.loads(text) except json.JSONDecodeError: match re.search(r\{.*\}, text, re.S) if match: return json.loads(match.group(0)) raise把safe_parse_json接到dispatch_node里之后路由解析基本稳定了。这里想提醒所有新手不要让提示词依赖模型自觉要在代码层面做防御。5.2 会无限循环的调度器递归限制来得太迟LangGraph 默认对递归深度有限制超过限制会抛RecursionLimit。我一开始没加step_count硬保护结果有一次模型反复给code_agent派任务明明代码已经生成了它还是说还需要进一步验证于是code_agent执行完回到dispatch又来一轮直到递归上限被触发整个调用链崩掉。这个问题在日志里看起来非常奇怪code_agent和dispatch两个节点互相跳像死机了一样。我当时第一反应是检查图结构怀疑边连错了排查了半天才发现纯粹是调度器的决策没被约束。解决方案就是在dispatch_node里加硬编码的轮次判断。我把它放在所有 LLM 调用之前如果轮次超限直接返回final。这个改动很小效果却立竿见影。后来我又把阈值从 3 改成 4因为有些合规流程确实需要代码→文案→质检→修改三个来回给定更大余量后减少误杀。5.3 ReAct Agent的中间步骤被截断结果缺胳膊少腿代码 Agent 用的是create_react_agent它内部会执行多轮思考-行动-观察循环。有一次它生成了很长的代码我拿到code_result后发现结尾是断的明显是被截断了。排查下来不是上下文窗口不够而是max_iterations或者输出长度限制卡住了。create_react_agent支持通过max_iterations控制思考轮数默认值在有些版本里偏小。当模型需要先思考、再跑代码、再看结果、再修正的时候轮数根本不够。我给这个 Agent 显式设置了max_iterations5并且把handle_parsing_errorsTrue打开让解析异常不至于中断整个 Agentcode_agent_runnable create_react_agent( llm, [run_python_snippet], promptcode_agent_prompt, max_iterations5, handle_parsing_errorsTrue, )还有一个容易被忽略的点代码生成 Agent 返回内容可能包含 Markdown 代码块标记比如python ...。传给文案 Agent 没问题但如果你直接拿它去喂给exec就会因为多余的标记字符而报错。我的建议是在run_python_snippet工具内部先做一层处理把代码块标记剥掉清理后再执行。6. 这个Demo还能怎么长进从玩具到能用的调度助手6.1 参数和提示词层面的性价比优化如果想让这套调度助手的产出更稳定第一步不是堆更多节点而是调模型参数。我把调度器和质检员的温度都保持 0.1文案 Agent 的温度设在 0.4 左右。原因前面说过路由和审查需要确定性文案生成需要一点变化性。提示词方面也有优化空间。DISPATCH_PROMPT 里我是把code_result和doc_result的完整内容塞进去的任务复杂后这会变得很长。你可以先用一个 LLM 调用把已有产出做压缩摘要再交给调度器判断能显著降低 token 消耗。这个摘要智能体本质上是多智能体架构里的信息预处理节点加进去不需要改动整体图结构。另外质检节点现在只输出pass/fail和一句反馈略显粗糙。可以升级为按代码可运行性、文档匹配度、完整性三个维度打分并给出具体修复建议。模型在结构化打分任务上的表现通常比笼统的判断是否通过好不少因为每个维度都有明确检查目标。6.2 扩展方向记忆、人工确认、多渠道输入我目前这个 demo 是无状态的每次调用都是独立任务不会记住上一次运行的历史。如果想让助手具备对话记忆StateGraph 编译时启用持久化机制把状态存到后端用户再次访问时能接着上一轮继续。这一步对任务调度的实际体验提升很明显比如用户说上次那个脚本再改成支持Excel调度器能结合历史上下文理解上次那个是谁。更实际的扩展是引入人工确认节点。比如代码 Agent 生成了删除数据库数据的脚本这种高风险任务不应该让审查 Agent 独自拍板而是应该中断流程等人类确认后再继续。LangGraph 提供了中断和恢复的机制可以在关键节点挂起。这样整个调度系统就从全自动变成了人在回路更适合真正的工作流。还可以继续加新角色资料搜索 Agent、表格数据分析 Agent、翻译 Agent等等。每加一个执行者只需要在调度器的动作列表里加一个枚举值再加一个对应的执行节点和一条条件边其余部分完全不用动。这种可扩展性正是把任务调度拆成多智能体架构带来的最大红利。我个人最后想多提醒一句多智能体系统确实能解决单模型角色混乱的问题但它不是银弹。节点越多、回退次数越多调用成本就越高延迟也越长。做的时候不要为了多智而多智先把用户需求拆出必要角色能用两个 Agent 就别加第三个保持图结构简单才能一眼看出问题出在哪。
你有没有遇到过这种情况想让一个模型同时搞定写脚本改bug写使用说明三件事结果它顾此失彼代码里夹着解释文字文案里又带着代码片段最后两边都不像样。我一开始也是这么干的直到我把任务真正拆开才意识到多智能体的价值不是炫技而是把一个复杂需求拆给不同岗位去并行思考。这篇文章就带你从零做一个简单的AI任务调度助手基于LangChain生态里的 Agent 和图形编排能力把调度者、执行者、质检员跑通一条完整链路。整篇文章适合两类人一类是刚接触 LangChain想知道 Agent 怎么串起来另一类是被单一大模型一个指令干到底折磨过想看看多智能体到底怎么落地的新手。我不会把代码堆成一堵墙而是会把每个设计决策的原因讲清楚顺手把调试时踩过的坑也记下来。1. 先聊聊为什么单个大模型干不好任务调度1.1 一个模型身兼数职指令会互相干扰你可能会问我直接给一个大模型写一段很长的提示词让它先写代码再写说明最后自查不也是一种调度吗表面上可以实际跑几次就发现问题了。大模型没有真正的工作记忆它只是根据上下文里的所有内容生成下一个 token。当代码任务和文案任务挤在同一段上下文中模型很容易把两种风格混在一起写代码时突然冒出一句这段逻辑很清晰写文案时又插入代码块。还有更麻烦的如果某个子任务执行失败你要回退重跑单一大模型往往无法判断到底该重写哪部分只能整段重新生成。我自己遇到最典型的一个场景是让模型写一个Python脚本统计CSV缺失值并同时生成说明文档。模型确实给出了完整回复但脚本里的注释是给非技术人员看的营销文案风格文档里又贴了一大段源码。分开看好像每个部分都对合在一起就是没法用。这就是角色打架。1.2 多智能体不是玄学本质是把流程拆给不同大脑多智能体的思路其实很像现实中的团队分工调度者负责接收需求、拆解任务、判断下一步交给谁执行者各有专长一个写代码一个写文案互不干扰质检员最后检查产物是否满足要求不合格就打回重来。每个智能体只关注自己的窄范围提示词短了指令冲突少了上下文也更干净了。从工程角度看多智能体的价值在于可控性和可观测性。单一大模型的调用像一个黑盒你很难知道它内部到底先做了什么而多智能体系统里每个节点是什么、输入什么、输出什么都是你自己定义的。哪怕中间某个 Agent 崩了你也知道是哪一步崩的改起来也快得多。在 LangChain 生态里实现多智能体一般有两种路线一种是让一个 Agent 持有其他 Agent 的工具自己做路由另一种是用 LangGraph 这类编排框架把每个智能体做成图里的一个节点节点之间用条件边连接。我这个项目选的是后者因为条件路由更直观而且对初学者来说图结构比隐藏的 Agent 工具链更容易调试。2. 环境准备与LangChain生态里的现实坑2.1 安装与版本选择LangChain不是装一个就完事很多新手第一次装 LangChain 时会被一堆包搞晕。现在 LangChain 已经不是单个库而是拆成了很多配套包。如果你只装了langchain会发现很多 Agent 相关功能根本用不了。我这个项目里实际需要这么几个pip install langgraph langchain langchain-openai langchain-community langchain-corelanggraph是负责编排多智能体流程的核心langchain-openai是让 LangChain 能调用 OpenAI 兼容格式的接口langchain-community里装着社区维护的一堆工具集成。不装langchain-core可能会缺基础类型虽然有些包会自动带上但显式装一遍能省掉不少版本冲突的烦恼。版本方面一定要留个心眼LangGraph 从 0.2 开始 API 有过一些调整有些教程里写的老写法会直接报错或者被废弃。安装的时候可以在包名后给定版本范围例如pip install langgraph0.2 langchain0.3提示如果后面发现某个函数调用报AttributeError先别急着怀疑自己代码很可能是某个依赖被升级到了不兼容的新版本。检查一下pip list里的版本再对照 langgraph 的变更说明基本能定位。2.2 模型接入随便一个大模型API都能当大脑这个项目里所有智能体本质上都是对同一个大模型 API 的多次调用只不过每次用的系统提示词不一样。所以我只需要一个能通过OpenAI 兼容接口访问的模型服务就行不限定具体厂商。用环境变量管理密钥和地址这样代码换模型时不用改动逻辑部分import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() llm ChatOpenAI( base_urlos.getenv(LLM_BASE_URL), api_keyos.getenv(LLM_API_KEY), modelos.getenv(LLM_MODEL_NAME), temperature0.1, )把LLM_BASE_URL、LLM_API_KEY、LLM_MODEL_NAME写到.env文件里就行。我用的是一个中规模的通用大模型上下文长度中等响应速度还不错。如果模型支持结构化输出调度节点的准确率会高不少如果模型不支持后面我会说怎么用提示词约束加解析兜底来处理。这里有一个跟新手相关的关键点temperature我设置成 0.1因为调度节点的任务是对接路由决策不需要创意需要确定性。写文案的 Agent 可以把温度调高到 0.4 甚至 0.7。同一份代码里不同节点使用不同温度完全没问题只要在定义节点时给不同的 LLM 实例就行。3. 这个任务调度助手的设计调度者、执行者、质检员3.1 状态设计是第一步先想清楚每个智能体之间传什么搭多智能体之前千万别直接写代码第一个要考虑的是状态State。多智能体运行相当于一个状态机每个节点读入状态执行完毕后又把新状态传给下一个节点。状态里放什么字段直接决定了每个 Agent 到底能看到多少信息。我这套助手的核心字段如下字段类型作用querystr用户的原始需求next_actionstr调度节点决定的下一步动作assignmentstr调度器给执行者的子任务说明code_resultstr代码 Agent 的工作产物doc_resultstr文案 Agent 的工作产物inspect_feedbackstr质检员的审查意见step_countint当前已经跑了多少个节点防止死循环有些字段你可能看不出用途但它们在调试时特别重要。比如step_count如果调度器出了问题会在节点之间疯狂往返没有这个计数器整个程序就跑死了也不自知。我会在调度节点的提示词里强制要求当 step_count 大于 3 时必须输出final动作。还要考虑一个问题code_result和doc_result都是直接覆盖还是保留历史我这个 demo 里选择直接覆盖因为每次重新生成时质检员的意见已经通过assignment传给了执行者不需要保存多轮历史状态。如果你要迭代很多次最好改成列表类型保留每一步的产物方便追溯和对比。3.2 三条角色链路与回退机制我设计了三类节点调度节点dispatcher读用户 query 和已有产出物输出 JSON决定下一步是去写代码、写文案还是直接收尾。执行节点代码 Agent 负责写脚本并运行验证文案 Agent 负责写说明文档、使用指南或解释段落。质检节点inspector汇总已完成的产出判断质量是否达标不达标就带着反馈意见回到调度节点重新指派。这个流程是一个典型的规划-执行-审查环。第一次跑通的时候最容易困惑的是为什么代码 Agent 执行完还要再回到调度节点而不是直接进质检节点因为用户的需求可能是先写代码再写文案也可能是只需要文案。如果代码 Agent 执行完就直接进质检那一份没有文案的产出根本没法审查。所以我的图里执行节点回环到调度节点由调度节点判断下一个该做什么当所有子任务都完成调度节点才会给出final动作进入质检环节。质检环节也支持回退如果质检不通过它会带着inspect_feedback回到调度节点调度器把反馈写进新的assignment再指派对应的执行者改一轮。这里需要留意的是回退次数必须有限制否则一个永远改不对的任务会把整个流程拖死。我的做法是把回退次数换算进step_count超过阈值就让质检员网开一面直接放行。4. 核心代码逐段拆解状态图、路由与Agent实现4.1 基础设施状态类型、模型初始化与Python执行工具代码从定义状态类型开始。我使用TypedDict来描述状态结构这样写节点函数时能获得 IDE 提示不容易因拼写错误翻车from typing import TypedDict class AssistantState(TypedDict): query: str next_action: str assignment: str code_result: str doc_result: str inspect_feedback: str step_count: int final_answer: str然后是 Python 执行工具。代码 Agent 生成脚本后不能光停在我写好了这一步最好真跑一遍看看有没有语法错误。我给 Agent 配一个自定义工具用exec执行传入的代码片段并捕获输出import io import contextlib from langchain_core.tools import tool tool def run_python_snippet(code: str) - str: 在本地进程内执行一段Python代码返回标准输出与错误信息。 仅用于demo验证短脚本生产环境务必放到隔离容器里。 stdout_buf io.StringIO() stderr_buf io.StringIO() try: with contextlib.redirect_stdout(stdout_buf), contextlib.redirect_stderr(stderr_buf): exec(code, {__name__: __main__}) except Exception as exc: return f运行异常: {type(exc).__name__}: {exc}\n标准输出:\n{stdout_buf.getvalue()} return f标准输出:\n{stdout_buf.getvalue()}\n标准错误:\n{stderr_buf.getvalue()}警告直接在本地exec运行代码是很危险的做法尤其当代码 Agent 的输出不可信时。这个 demo 只是教学用我建议你在沙箱容器里跑。即使如此代码 Agent 的执行结果也要当作外部输出看待不要盲目信任。4.2 调度节点的实现负责拆解与路由调度节点本身不需要工具它就是一次带结构化输出要求的 LLM 调用。我会让模型返回一个 JSON包含next_action和assignment两个字段。调度器的系统提示词需要把可选项讲清楚code表示去写代码doc表示去写文案final表示所有产出已完成准备交给质检。同时还要提醒模型参考已有的code_result和doc_result避免重复劳动。from langchain_core.output_parsers import JsonOutputParser from langchain_core.prompts import ChatPromptTemplate DISPATCH_PROMPT 你是任务调度器负责拆解用户需求并安排执行者。 可选动作 - code: 让代码Agent去编写并验证Python代码 - doc: 让文案Agent去撰写说明文档或文案 - final: 所有必要产物都已生成进入质检环节 用户需求{query} 已有代码产物{code_result} 已有文案产物{doc_result} 当前执行计数{step_count} 规则 1. 如果用户需要代码而代码产物为空优先选择code。 2. 如果用户需要说明文档而文案产物为空选择doc。 3. 如果步骤超过3轮不管怎样都选择final。 4. 输出JSON格式{next_action: code, assignment: 给执行者的子任务描述, reason: 简单说明理由} 只输出JSON不要输出其他内容。 parser JsonOutputParser() def dispatch_node(state: AssistantState) - dict: if state.get(step_count, 0) 3: return { next_action: final, assignment: 所有任务已经执行过多轮请质检后直接汇总, step_count: state.get(step_count, 0) 1, } prompt ChatPromptTemplate.from_template(DISPATCH_PROMPT) chain prompt | llm | parser result chain.invoke({ query: state[query], code_result: state.get(code_result, 还没有代码产物), doc_result: state.get(doc_result, 还没有文案产物), step_count: state.get(step_count, 0), }) return { next_action: result.get(next_action, final), assignment: result.get(assignment, ), step_count: state.get(step_count, 0) 1, }这个实现里有一个很容易忽略但很重要的点我利用step_count作为硬性逃生阀即使大模型输出的next_action一直不老实超过 3 轮也会被强行改成final。4.3 执行者节点代码Agent和文案Agent代码 Agent 用 LangChain 的create_react_agent创建让它能调用run_python_snippet工具来实际执行生成的脚本。系统提示词限定它的职责范围只写代码、只验证代码不要写说明文档代码需要直接能跑并给出必要注释。from langchain.agents import create_react_agent from langchain_core.prompts import PromptTemplate CODE_AGENT_TMPL 你是一名Python工程师。你只负责根据任务描述编写可运行的Python代码 并用run_python_snippet工具运行验证。不要把任务描述之外的文案混进来。 任务描述 {assignment} 要求 - 代码要简洁包含必要的注释。 - 先调用run_python_snippet运行一次即使报错也要把错误信息原样反馈。 - 最终回复时先给完整代码再附一行执行结果摘要。 code_agent_prompt PromptTemplate.from_template(CODE_AGENT_TMPL) code_agent_runnable create_react_agent(llm, [run_python_snippet], promptcode_agent_prompt)执行节点里的工作就是把这个 Agent 跑起来并把最终输出塞回状态里def code_agent_node(state: AssistantState) - dict: result code_agent_runnable.invoke({assignment: state.get(assignment, )}) return {code_result: result[output]}文案 Agent 其实不需要工具我用一个普通的链来生成文案DOC_PROMPT 你是文档撰写员。你不修改代码只根据已有材料撰写用户能看懂的说明文字。 要求语气清晰、步骤明确、不要空话。 已有代码产物 {code_result} 任务描述 {assignment} 请输出说明文档正文。 def doc_agent_node(state: AssistantState) - dict: prompt ChatPromptTemplate.from_template(DOC_PROMPT) chain prompt | llm result chain.invoke({ code_result: state.get(code_result, 暂无代码产物), assignment: state.get(assignment, ), }) return {doc_result: result.content if hasattr(result, content) else str(result)}可以看到文案 Agent 的输入里包含了code_result。这是多智能体协作的关键上一个角色的输出成为下一个角色的输入上下文。如果把这一步省掉文案 Agent 就成了盲写写出来的说明和实际代码可能对不上。4.4 质检节点与最终汇总质检节点的职责是站在用户视角看整体结果。它对代码和文案做一次通读输出两个判断字段pass或fail以及反馈意见。实现上依然是给 LLM 一个结构化的输出要求。INSPECT_PROMPT 你是质量审查员。请检查以下代码和文案是否匹配、是否完整、是否有明显错误。 代码产物 {code_result} 文案产物 {doc_result} 用户需求 {query} 输出JSON格式 {verdict: pass 或 fail, feedback: 你的具体审查意见} 只输出JSON。 def inspect_node(state: AssistantState) - dict: prompt ChatPromptTemplate.from_template(INSPECT_PROMPT) chain prompt | llm | JsonOutputParser() result chain.invoke({ code_result: state.get(code_result, ), doc_result: state.get(doc_result, ), query: state[query], }) verdict result.get(verdict, fail) feedback result.get(feedback, ) return {next_action: verdict, inspect_feedback: feedback}汇总节点的任务就简单多了把代码和文案拼成最终答案def aggregate_node(state: AssistantState) - dict: final_answer ( f### 代码部分\n\n{state.get(code_result, )}\n\n f### 文案部分\n\n{state.get(doc_result, )}\n\n f### 质检意见\n\n{state.get(inspect_feedback, )} ) return {final_answer: final_answer}4.5 把图组装起来并跑一个完整例子现在到了最关键的一步用 LangGraph 把这些节点连成一张图。有四个节点需要分别处理dispatch、code_agent、doc_agent、inspect、aggregate。连接规则是dispatch根据next_action无条件选择去code_agent、doc_agent还是inspect。code_agent结束后回到dispatch让它继续判断是否需要doc_agent。doc_agent结束后同样回到dispatch。inspect根据next_action判断是回dispatch继续修改还是去aggregate收尾。from langgraph.graph import StateGraph, END graph StateGraph(AssistantState) graph.add_node(dispatch, dispatch_node) graph.add_node(code_agent, code_agent_node) graph.add_node(doc_agent, doc_agent_node) graph.add_node(inspect, inspect_node) graph.add_node(aggregate, aggregate_node) graph.set_entry_point(dispatch) graph.add_conditional_edges( dispatch, lambda state: state.get(next_action, final), { code: code_agent, doc: doc_agent, final: inspect, }, ) graph.add_edge(code_agent, dispatch) graph.add_edge(doc_agent, dispatch) graph.add_conditional_edges( inspect, lambda state: aggregate if state.get(next_action) pass else dispatch, { aggregate: aggregate, dispatch: dispatch, }, ) graph.add_edge(aggregate, END) app graph.compile()然后就可以用一句话来测试整个调度助手result app.invoke({ query: 帮我写一个Python脚本读取data.csv并统计每列缺失值数量然后用中文写一段脚本使用说明。, next_action: , assignment: , code_result: , doc_result: , inspect_feedback: , step_count: 0, final_answer: , }) print(result[final_answer])第一次跑通的时候你会看到 LangGraph 打印出一串节点执行日志比如dispatch - code_agent - dispatch - doc_agent - dispatch - inspect - aggregate这个顺序正是调度器自己规划出来的。这种感觉挺奇妙的——你并没有在代码里写死流程而是让模型在几个选项之间做选择最终走出一条合理的执行路径。5. 第一次跑往往不会顺利我踩过的三个实际问题5.1 路由结果不够结构化带来的解析危机第一次搭建时我让调度器直接输出文本然后在代码里用if code in output这种办法判断下一步。听起来很省事实际上一团糟。模型可能输出我们应该先 code然后 doc也可能输出CODE大小写和格式都不稳定判断逻辑越写越脏。后来我改成用JsonOutputParser明确要求只输出 JSON。结果在小模型上又出现一个问题模型不遵守指令输出的 JSON 前面多了一句解释性文字或者直接把引号写弯了解析器照样报错。解决思路有两个一是换用支持response_format{type: json_object}的模型在 API 层面强制结构输出二是做好幂等的解析兜底用正则把第一个{到最后一个}之间的内容截出来再解析。我现在的做法是写一个小的解析函数如果直接解析失败就尝试提取 JSON 片段import json import re def safe_parse_json(text: str) - dict: text text.strip() try: return json.loads(text) except json.JSONDecodeError: match re.search(r\{.*\}, text, re.S) if match: return json.loads(match.group(0)) raise把safe_parse_json接到dispatch_node里之后路由解析基本稳定了。这里想提醒所有新手不要让提示词依赖模型自觉要在代码层面做防御。5.2 会无限循环的调度器递归限制来得太迟LangGraph 默认对递归深度有限制超过限制会抛RecursionLimit。我一开始没加step_count硬保护结果有一次模型反复给code_agent派任务明明代码已经生成了它还是说还需要进一步验证于是code_agent执行完回到dispatch又来一轮直到递归上限被触发整个调用链崩掉。这个问题在日志里看起来非常奇怪code_agent和dispatch两个节点互相跳像死机了一样。我当时第一反应是检查图结构怀疑边连错了排查了半天才发现纯粹是调度器的决策没被约束。解决方案就是在dispatch_node里加硬编码的轮次判断。我把它放在所有 LLM 调用之前如果轮次超限直接返回final。这个改动很小效果却立竿见影。后来我又把阈值从 3 改成 4因为有些合规流程确实需要代码→文案→质检→修改三个来回给定更大余量后减少误杀。5.3 ReAct Agent的中间步骤被截断结果缺胳膊少腿代码 Agent 用的是create_react_agent它内部会执行多轮思考-行动-观察循环。有一次它生成了很长的代码我拿到code_result后发现结尾是断的明显是被截断了。排查下来不是上下文窗口不够而是max_iterations或者输出长度限制卡住了。create_react_agent支持通过max_iterations控制思考轮数默认值在有些版本里偏小。当模型需要先思考、再跑代码、再看结果、再修正的时候轮数根本不够。我给这个 Agent 显式设置了max_iterations5并且把handle_parsing_errorsTrue打开让解析异常不至于中断整个 Agentcode_agent_runnable create_react_agent( llm, [run_python_snippet], promptcode_agent_prompt, max_iterations5, handle_parsing_errorsTrue, )还有一个容易被忽略的点代码生成 Agent 返回内容可能包含 Markdown 代码块标记比如python ...。传给文案 Agent 没问题但如果你直接拿它去喂给exec就会因为多余的标记字符而报错。我的建议是在run_python_snippet工具内部先做一层处理把代码块标记剥掉清理后再执行。6. 这个Demo还能怎么长进从玩具到能用的调度助手6.1 参数和提示词层面的性价比优化如果想让这套调度助手的产出更稳定第一步不是堆更多节点而是调模型参数。我把调度器和质检员的温度都保持 0.1文案 Agent 的温度设在 0.4 左右。原因前面说过路由和审查需要确定性文案生成需要一点变化性。提示词方面也有优化空间。DISPATCH_PROMPT 里我是把code_result和doc_result的完整内容塞进去的任务复杂后这会变得很长。你可以先用一个 LLM 调用把已有产出做压缩摘要再交给调度器判断能显著降低 token 消耗。这个摘要智能体本质上是多智能体架构里的信息预处理节点加进去不需要改动整体图结构。另外质检节点现在只输出pass/fail和一句反馈略显粗糙。可以升级为按代码可运行性、文档匹配度、完整性三个维度打分并给出具体修复建议。模型在结构化打分任务上的表现通常比笼统的判断是否通过好不少因为每个维度都有明确检查目标。6.2 扩展方向记忆、人工确认、多渠道输入我目前这个 demo 是无状态的每次调用都是独立任务不会记住上一次运行的历史。如果想让助手具备对话记忆StateGraph 编译时启用持久化机制把状态存到后端用户再次访问时能接着上一轮继续。这一步对任务调度的实际体验提升很明显比如用户说上次那个脚本再改成支持Excel调度器能结合历史上下文理解上次那个是谁。更实际的扩展是引入人工确认节点。比如代码 Agent 生成了删除数据库数据的脚本这种高风险任务不应该让审查 Agent 独自拍板而是应该中断流程等人类确认后再继续。LangGraph 提供了中断和恢复的机制可以在关键节点挂起。这样整个调度系统就从全自动变成了人在回路更适合真正的工作流。还可以继续加新角色资料搜索 Agent、表格数据分析 Agent、翻译 Agent等等。每加一个执行者只需要在调度器的动作列表里加一个枚举值再加一个对应的执行节点和一条条件边其余部分完全不用动。这种可扩展性正是把任务调度拆成多智能体架构带来的最大红利。我个人最后想多提醒一句多智能体系统确实能解决单模型角色混乱的问题但它不是银弹。节点越多、回退次数越多调用成本就越高延迟也越长。做的时候不要为了多智而多智先把用户需求拆出必要角色能用两个 Agent 就别加第三个保持图结构简单才能一眼看出问题出在哪。
平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测)
