☰
LangGraph提示链实战:构建稳定可控的AI Agent工作流
2026/10/20 21:49:05 网站建设 项目流程

做 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(text=text)) result2 = llm.invoke(prompt2.format(summary=result1)) result3 = llm.invoke(prompt3.format(insights=result2))

没毛病,但一旦链条变长、分支变多、出现循环或人工审批点,这种线性脚本就失控了。你没法可视化每一步的状态,没法容易地回滚重试,没法插入条件判断,更没法在中间暂停。

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( model="gpt-4o-mini", temperature=0.3, api_key=os.getenv("OPENAI_API_KEY"), base_url=os.getenv("OPENAI_BASE_URL") # 可选,兼容代理/中转 )

这里解释一下temperature=0.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( model="gpt-4o-mini", temperature=0.3, api_key=os.getenv("OPENAI_API_KEY"), base_url=os.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(article=state["article"])) return {"summary": resp.content.strip()} def extract_points_node(state: ChainState) -> dict: resp = llm.invoke(POINTS_PROMPT.format( article=state["article"], summary=state["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( summary=state["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", encoding="utf-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这样的 reducer:

from 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( ..., timeout=30, max_retries=2, request_timeout=30 )

如果某步骤经常超时,也可以考虑换更快的模型。提示链的好处就在这里:你完全可以为了速度在摘要节点用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 的各种高级用法,都是在这个基础上做加法。遇到状态不对、节点死循环、模型输出解析失败,回看这篇里的排查思路,能省下你半天查文档的时间。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询