
做 AI 应用开发的人今年大概都绕不开一个词Agent。但很多人一开始就冲着“全自动多智能体协作平台”去结果项目烂尾、调试到怀疑人生。我自己的体会是真正该先掌握的不是花哨的编排框架而是最朴素也最稳固的Prompt Chaining 提示链。只要你手头有一个具体任务、需要多个LLM调用前后衔接LangGraph 就能让这个链条变得可控、可复用、可调试。这篇文章我会用一个真实的案例把提示链从设计到落地的完整过程拆给你看代码全部免费放出来照着抄就能跑。适合谁读刚接触 LangGraph 但不知道从哪下手的开发者已经用 LangChain 写过几条链但想升级成有状态、有图结构的更规范方案的人以及所有被“复杂Agent”伤害过、想回归简单可靠设计的朋友。读完之后你不仅能写出一条可用的提示链还会明白为什么“先跑通线性链”比“直接上循环图”重要得多。1. Prompt Chaining 是什么以及它凭什么值得学1.1 一次调用 vs 一个链条先说个直觉如果给大模型一个问题它能直接给你最终答案那当然最好。但现实是稍微复杂一点的任务一步到位的结果往往不稳定。比如你想让模型“从一篇技术文章里提取三个核心观点再生成一个适合社交媒体的摘要”如果只用一次调用模型可能在摘要里掺入原文的噪音或者第二个任务直接污染了提取质量。Prompt Chaining 的思路就是把一个大任务拆成多个小步骤每一步用一个独立的LLM调用去完成上一步的输出作为下一步的输入。这就像流水线装配不是一个工人从头装到尾而是每个工位只干一件事干完交给下一个工位。每个工位都能单独质检、单独优化出了问题也能精确锁定是哪个环节的锅。1.2 为什么用 LangGraph 而不是普通 Python 脚本其实不用任何框架纯 Python 也能写提示链result1 llm.invoke(prompt1.format(texttext)) result2 llm.invoke(prompt2.format(summaryresult1)) result3 llm.invoke(prompt3.format(insightsresult2))没毛病但一旦链条变长、分支变多、出现循环或人工审批点这种线性脚本就失控了。你没法可视化每一步的状态没法容易地回滚重试没法插入条件判断更没法在中间暂停。LangGraph 解决的问题就是把整个执行流程定义成一张图每个节点是一个函数边表示数据流动。它天然支持状态管理、条件分支、循环、中断和恢复。而且它的底层是 LangChain节点里随便调ChatOpenAI、ChatAnthropic或其他模型属于原生生态不用绕路。换句话说Prompt Chaining 是设计模式LangGraph 是实现这个模式的工程框架。先把模式搞懂再用框架落地后面扩展成更好的形态就不用推翻重来。2. 核心架构拆解节点、状态、边2.1 状态是链的灵魂LangGraph 里的状态State是一个数据类可以是 TypedDict 或者 Pydantic BaseModel。每个节点执行完都会产生一个更新后的状态传递给下一个节点。最重要的点是“状态在每条边上流动而不是直接用函数返回值接函数返回值。”你可能会问这和我直接用a f(x); b g(a)有什么区别区别在于LangGraph 的状态允许你保留整个上下文的累积信息而不只是上一步的单一输出。比如你第一步提取了关键词第二步要根据关键词写摘要第三步还要把关键词和摘要一起拿去生成标签。如果你用普通 Python可能需要手动维护一个字典传来传去。而在 LangGraph 里状态里始终有keywords和summary两个字段每个节点只关心它需要的那个字段别的字段自动保留。这就是“状态管理”在长链条里的核心价值。2.2 节点每个节点做好一件事节点就是普通 Python 函数输入是当前状态输出是一个字典表示要更新哪些字段。比如def extract_keywords_node(state): keywords llm.invoke(...) return {keywords: keywords}这里的关键是“单一职责”。如果某个节点既要抽取关键词又要写摘要那说明你的节点拆分粒度太粗后面调试会很被动。2.3 边白盒的数据流向边的定义就一句话根据状态决定下一步走向哪个节点。在 LangGraph 里最简单的是一条直线START - node1 - node2 - END。但你也可以在中间加条件边比如“如果摘要长度超过 500 字就走压缩节点否则直接到最终报告”。这个用普通脚本写起来也能实现但条件一多代码里全是if读起来很乱。LangGraph 图本身就是可视化的一眼能看到分支全貌。2.4 构建图与编译LangGraph 需要你用StateGraph先把节点和边搭起来然后调用compile()得到一个可执行的对象。之后你就像调用普通函数一样调用它graph StateGraph(MyState) graph.add_node(node1, node1) graph.add_edge(START, node1) ... app graph.compile() result app.invoke({text: ...})这一步很像后端开发里的“路由配置”不直接写执行逻辑而是把逻辑挂载到图结构上。责任分离清楚测试也好做。提示如果你用过 React Flow 这类前端节点编辑器会觉得这个模型非常亲切。LangGraph 就是在后端实现了类似 Node-RED 的编排能力只不过节点是 Python 函数而已。3. 完整实操从长文生成结构化知识报告3.1 场景设计这一步我选一个大家都容易代入的任务输入一篇技术文章正文输出一份结构化知识报告包含核心摘要、关键技术点、适用场景、一句话总结。为什么选这个因为它足够小方便聚焦提示链本身的逻辑又足够典型覆盖了“摘要提取-知识结构化-包装输出”三种常见环节几乎可以平移到任何业务里。整个链路我规划成 4 个节点summarize_node生成长文摘要。extract_points_node提取关键技术点限定额度。build_report_node基于摘要和关键点生成最终报告。save_report_node把结果写入本地 Markdown 文件。前三个是 LLM 调用第四个是纯 IO 操作用来演示“链中不仅能接 LLM也能接普通函数”。3.2 环境准备先装依赖pip install langgraph langchain-core langchain-openai注意LangGraph 的 API 在 0.1 后基本稳定了我使用的是 0.2.x 版本。早期版本的 API 变化较大如果你看别的教程遇到不兼容优先以官方文档为准。还需要一个 LLM 的 ENV 变量。示例代码里我用的是 OpenAI 兼容接口方便你替换成任意兼容的服务或本地模型from langchain_openai import ChatOpenAI import os llm ChatOpenAI( modelgpt-4o-mini, temperature0.3, api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL) # 可选兼容代理/中转 )这里解释一下temperature0.3在提示链场景里你更希望每一步“稳定产出、不要发挥”所以温度偏低。尤其像提取关键点这种事实型任务温度太高容易胡说。当然如果某个节点需要创意比如写广告文案可以单独给那个节点换一个温度更高的模型实例。3.3 定义状态结构状态用 TypedDict 定义即可from typing_extensions import TypedDict class ChainState(TypedDict): article: str # 输入原文 summary: str # 节点1的输出 points: list[str] # 节点2的输出 report: str # 节点3的输出 saved_path: str # 节点4的输出每个字段的含义和生命周期很明确。article一开始由用户传入之后不被节点覆盖summary和points分别由各自的节点填充report最终由报告节点生成。这种字段级别的状态管理让我在排查问题时能精确确认“状态更新到哪一步了”。3.4 完整免费源代码下面是完整代码。我把提示词用system和human分离这样既能约束模型的角色也能传入动态数据。import os from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI # ---------- 1. 模型初始化 ---------- llm ChatOpenAI( modelgpt-4o-mini, temperature0.3, api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL), ) # ---------- 2. 状态定义 ---------- class ChainState(TypedDict): article: str summary: str points: list[str] report: str saved_path: str # ---------- 3. 提示词模板 ---------- SUMMARY_PROMPT 你是一个资深技术编辑。请阅读下面的文章写一段150字以内的摘要要求覆盖核心主题和最终结论。 文章 {article} POINTS_PROMPT 根据下面的文章摘要提取最重要的3个技术要点。每个要点用一句话表达不要展开论述不要添加原文中不存在的信息。 文章 {article} 摘要 {summary} REPORT_PROMPT 根据摘要和技术要点生成一份结构化知识报告包含以下四个部分 1. 核心摘要 2. 关键技术点 3. 适用场景 4. 一句话总结 要求 - 报告面向有技术背景的读者 - 语言简洁专业不要废话 - 技术点部分使用列表输出 摘要 {summary} 技术要点 {points} # ---------- 4. 节点函数 ---------- def summarize_node(state: ChainState) - dict: resp llm.invoke(SUMMARY_PROMPT.format(articlestate[article])) return {summary: resp.content.strip()} def extract_points_node(state: ChainState) - dict: resp llm.invoke(POINTS_PROMPT.format( articlestate[article], summarystate[summary] )) # 简单按换行拆分成列表实际工程可要求模型输出JSON后再解析 raw_lines [line.strip() for line in resp.content.strip().split(\n) if line.strip()] points [line.lstrip(-* ).strip() for line in raw_lines] return {points: points[:3]} def build_report_node(state: ChainState) - dict: resp llm.invoke(REPORT_PROMPT.format( summarystate[summary], points\n.join(state[points]) )) return {report: resp.content.strip()} def save_report_node(state: ChainState) - dict: path knowledge_report.md with open(path, w, encodingutf-8) as f: f.write(state[report]) return {saved_path: path} # ---------- 5. 构建图 ---------- builder StateGraph(ChainState) builder.add_node(summarize, summarize_node) builder.add_node(extract_points, extract_points_node) builder.add_node(build_report, build_report_node) builder.add_node(save_report, save_report_node) builder.add_edge(START, summarize) builder.add_edge(summarize, extract_points) builder.add_edge(extract_points, build_report) builder.add_edge(build_report, save_report) builder.add_edge(save_report, END) app builder.compile() # ---------- 6. 调用 ---------- if __name__ __main__: sample_article LangGraph是一个用于构建有状态AI Agent的框架。它基于图结构管理节点和边允许开发者定义复杂的控制流。 与LangChain的传统链式调用相比LangGraph能更自然地表达循环、分支和人类介入。 它支持细粒度的状态管理允许在每一步保存上下文并且通过检查点机制支持交互式Agent。 在实际项目中LangGraph可以用于自动化代码审查、客户支持、数据分析等场景。 result app.invoke({article: sample_article}) print( 最终报告 ) print(result[report]) print( 保存文件 ) print(result[saved_path])3.5 运行效果与解释这段代码跑起来后会在当前目录生成knowledge_report.md同时在终端打印最终报告。整个执行顺序由图中边保证save_report一定在build_report之后不会出现“文件写了但内容还是空”的竞态。这里有个很实用的细节如果某个节点中途报错LangGraph 会抛出异常并终止整个图执行而不会像普通脚本那样继续往下走。这意味着你可以在开发阶段发现“第2步模型返回了空内容”这类问题而不是等到最后拿到一份残缺报告才察觉。3.6 为什么我不直接在提示词里一次搞定你可能会想直接把“摘要提取要点报告”写进一个提示词让大模型一步生成不就行了一次调用当然能输出一个看起来差不多的报告但仔细对比会发现几个致命问题输出格式不稳定一步生成时模型经常把摘要和要点写混或者少写某个章节。无法单独优化如果提取的技术点质量差你没法单独替换那个环节的模型或提示词。成本浪费一次超长提示词生成 800 字报告和三次短任务生成 800 字报告虽然总 Token 可能差不多但长 Prompt 更容易触发模型注意力衰减后面的内容质量下降。提示链的另一个隐藏优势是中间结果可以复用。比如你以后只想输“核心摘要”而不想生成完整报告直接调用到第一个节点结束即可。这在普通的一步调用里根本无法实现。4. 实测中的常见问题与排查实录4.1 状态字段被覆盖LangGraph 节点返回的字典会默认做“合并更新”也就是说你返回{summary: ...}只会更新summary字段不会清空其他字段。但如果你定义状态时用了TypedDict且某个字段类型是list情况就有点特殊默认的 reducer 是“整体替换”不是“追加”。我第一次写extract_points_node时直接返回{points: [A, B, C]}第二次运行时想追加新的点结果发现旧点全没了。这是因为我没有定义points字段的 reducer。如果你需要累积多个节点生成的列表就得用operator.add这样的 reducerfrom typing import Annotated from operator import add class ChainState(TypedDict): article: str summary: str points: Annotated[list[str], add] report: str saved_path: str这个坑非常隐蔽尤其是从普通 Python 转过来的人会觉得“赋值就是覆盖”但 LangGraph 的状态管理本质上更像一个有取舍的数据库字段行为你可以通过Annotated自定义。遇到状态值不对先检查 reducer 定义。4.2 模型输出格式不听话我在extract_points_node里简单按换行拆列表这个方法在gpt-4o-mini上表现不错但一旦换成其他模型它可能输出1.前缀、-前缀甚至直接输出一行带“要点”的前缀。工程上最稳妥的做法是让模型输出 JSON然后用解析器抽取from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import JsonOutputParser POINTS_PROMPT 请提取三个技术要点输出JSON格式{points: [要点1, 要点2, 要点3]} 文章{article} 摘要{summary}然后parser JsonOutputParser() parsed parser.invoke(resp.content) points parsed[points]这样至少能保证解析不炸。提示词里明确格式 用解析器兜底是提示链中所有带结构化输出的节点的通用方案。4.3 某个节点长时间不返回原因通常是模型调用超时或者提示词太长导致响应慢。排查思路先单独调用那个节点函数喂固定输入看耗时。检查base_url指向的服务是否限流、是否需要加代理配置。给ChatOpenAI加上timeout和max_retries参数llm ChatOpenAI( ..., timeout30, max_retries2, request_timeout30 )如果某步骤经常超时也可以考虑换更快的模型。提示链的好处就在这里你完全可以为了速度在摘要节点用gpt-4o-mini在报告节点用一个更强模型因为每一步都是独立的 LLM 调用。4.4 调试神器给节点加日志LangGraph 本身提供了回调但最朴素的调试方式是在每个节点里打印关键中间值。反正你是开发阶段直接print不丢人。def summarize_node(state: ChainState) - dict: print(f[summarize] 输入的article长度: {len(state[article])}) resp llm.invoke(...) print(f[summarize] 输出的summary长度: {len(resp.content)}) return {summary: resp.content.strip()}这类日志在排查“哪一步开始输出为空”时非常有价值。我在实际项目中遇到过一个经典问题第二步生成的points只有一条导致最终报告缺两个技术点。一开始以为是模型问题加了日志才发现是第一步摘要里漏了相关信息导致第二步无中生有的能力变弱。这是链条的“误差级联”效应也是为什么要一步步检查而非只看终态。4.5 用图可视化确认流程LangGraph 编译后的app对象可以用get_graph()输出结构虽然不同版本 API 不同但基本思路是from langgraph.graph import StateGraph ... print(app.get_graph().draw_ascii())如果你用的版本支持draw_mermaid_png()还可以导出成图片。可视化对排查“边连错了”“少了一个节点”特别有效。我见过有人把build_report和extract_points顺序连反最后报告里出现“根据报告提取要点”这种尴尬内容画图一眼就能看出来。5. 从线性链扩展成更复杂结构的可能性5.1 条件分支变长链条顺着提示链往下走最常见的需求就是条件分支。比如摘要节点的输出字数超过 200就把摘要再压缩一次否则直接进入提取要点环节。在 LangGraph 里这个分支用条件边实现def should_compress(state: ChainState) - str: if len(state[summary]) 200: return compress return extract_points然后在构建图时builder.add_conditional_edges( summarize, should_compress, {compress: compress_summary, extract_points: extract_points} )这样图就不再是纯线性而是有了第一个“决策点”。提示链的稳定性加上条件分支后能覆盖更多业务逻辑比如“如果输入为空就跳过”“如果用户提供了关键词就直接用数据库数据”。5.2 循环自省式优化有时一个节点生成的结果不满足要求需要循环重试。LangGraph 可以加一个验证节点如果验证失败就回到上一个节点重新生成。def quality_check(state: ChainState) - str: if 关键技术点 not in state[report]: return rebuild return end配合条件边build_report最多重试两次。这种有限循环比无限重试稳定得多而且能防止模型在坏结果里绕圈白白烧钱。5.3 人工审批控制权和信任在企业落地时人工介入点是必须的。LangGraph 的interrupt机制能暂停图执行等待人输入。提示链中常见的用法是save_report执行前先暂停让用户确认报告是否合格不合格走修改节点。这个机制极大扩展了“链”的适用范围从全自动变成了半自动。不过要提醒你人工审批会增加产品复杂度务必在核心链条跑通后再加否则会因为状态恢复问题把开发打回原型。5.4 检查点机制每条链都留痕LangGraph 的持久化功能允许你在每次节点执行后保存状态之后可以恢复到任意一步。这对“追踪一次运行”特别有用。对于提示链来说更实际的价值在于“失败重跑”时不必重新调用之前已经成功且耗时的节点。你可以给compile传一个检查点存储例如内存存储或数据库存储。如果在生产环境跑务必认真考虑持久化方案而不是把 LangGraph 当成只能在内存里跑的一次性脚本。毕竟链越长重跑成本越高。6. 一点实践经验什么时候该停手我踩过不少坑也见过同事把 3 个节点的链扩展成 30 个节点的“万灵图”最后调试时根本找不到问题在哪。我的建议是如果任务能用一次llm.invoke稳定解决就别上链。如果比较急先把最简单的两个节点链跑通再逐步加节点。每个节点在加入大链之前单独用两三组测试数据验证过。在提示词里每个节点的输出格式越明确越好最好用 JSON不要用自然语言。给每个节点起名时用动词加对象比如summarize_article比node1清晰一百倍。我个人现在的习惯是所有新任务先画清楚“链路图”哪怕只是画在草稿纸上。确认哪些信息需要被传递哪些信息是阶段性产物哪些最终要落到硬盘或数据库里。然后在 LangGraph 里按图施工先写一个能跑的线性链再考虑分支和循环。这样不会因为框架的花哨功能而忘记核心目标——让每一步可控、让输出稳定。提示链不是最炫的技术但却是所有复杂 Agent 方案的稳定地基。先把这篇文章里的链路跑通你会发现后面对 LangGraph 的各种高级用法都是在这个基础上做加法。遇到状态不对、节点死循环、模型输出解析失败回看这篇里的排查思路能省下你半天查文档的时间。