一个使用最新 LangChain + LangGraph APIs 创建多步骤、有状态 LLM 管道的实践指南
图片来源：Markus Spiske on Unsplash
引言
大型语言模型（LLMs）功能强大——但在现实世界的系统中使用它们通常需要结构。你很少只需要调用一次模型；你需要多步骤推理、有状态的数据传递和可靠的编排。这正是 LangGraph 为 LangChain 带来的东西。它是一个用于定义结构化、有状态工作流的框架，每个步骤都建立在上一步的基础上。
在本指南中，你将学习如何使用 LangChain 和 LangGraph APIs 构建一个顺序工作流。我们将通过一个实际例子：一个三步骤的文本摘要管道，处理文本、优化它并验证最终输出。
什么是 LangChain 和 LangGraph？
在开始写代码之前，先来了解一下背景。
LangChain 提供了构建模块——模型、提示、检索器和工具。
LangGraph 提供了工作流引擎——一种将这些构建模块连接成可靠、可组合管道的方法。
可以把它想象成 LLMs 的 Airflow——但专为 LangChain 设计，面向 AI 推理。
设置你的环境
首先，安装最新版本的 LangChain 和 LangGraph：
pip install -U langchain langgraph openai
你还需要一个 OpenAI API 密钥（或者 LangChain 支持的其他 LLM 提供者的密钥）：
export OPENAI_API_KEY="your-api-key"
步骤 1：定义你的状态模式
每个 LangGraph 工作流都基于一个共享状态——一个类似字典的对象，从一个节点传递到下一个节点。
这里我们使用 Python 的 TypedDict 来定义工作流的状态模式：
from typing import TypedDict
class SummarizationState(TypedDict, total=False):
input_text: str
summary: str
refined_summary: str
validated_summary: str
或者，也可以使用 Pydantic。
步骤 2：初始化一个 LLM
我们将通过 LangChain 的 ChatOpenAI 包装器使用 OpenAI 的 GPT-5：
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-5", temperature=0.2)
LLM 中的 temperature 参数控制其响应的随机性和创造性。可以把它想象成创意的恒温器——调高会让输出更冒险、更不可预测，调低则让输出更专注、更保守。LLM 的核心是通过为所有可能的词分配概率分数来预测下一个词。temperature 在选择词之前会调整这些概率。
低 temperature（例如 0.1 到 0.5）：这个设置让模型更自信和确定。它会“锐化”概率分布，意味着更有可能选择最常见的词。模型会倾向于生成最常见和可预测的文本。
高 temperature（例如 > 0.5）：这个设置会“平滑”概率分布，让不太可能的词有更多被选中的机会。这会增加随机性，导致更令人惊讶、更有创意甚至有时荒诞的输出。
步骤 3：定义节点
节点是你工作流的构建模块——每个节点完成一个单一任务并更新工作流状态。
我们将定义三个节点：
摘要输入文本。 优化摘要以提高清晰度。 验证最终版本。
from langchain.prompts import PromptTemplate
from langgraph import node
@node()
def summarize(state: SummarizationState) -> SummarizationState:
prompt = PromptTemplate(
template="Summarize the following text in 3-4 sentences:\n\n{text}",
input_variables=["text"]
)
summary = llm.invoke(prompt.format(text=state["input_text"])).content
return {"summary": summary}
@node()
def refine(state: SummarizationState) -> SummarizationState:
prompt = PromptTemplate(
template="Refine this summary to be concise and clear:\n\n{summary}",
input_variables=["summary"]
)
refined = llm.invoke(prompt.format(summary=state["summary"])).content
return {"refined_summary": refined}
@node()
def validate(state: SummarizationState) -> SummarizationState:
prompt = PromptTemplate(
template=(
"Ensure this summary accurately represents the text "
"and is under 100 words.\n\n"
"TEXT:\n{text}\n\nSUMMARY:\n{refined_summary}"
),
input_variables=["text", "refined_summary"]
)
validated = llm.invoke(
prompt.format(
text=state["input_text"],
refined_summary=state["refined_summary"]
)
).content
return {"validated_summary": validated}
步骤 4：将所有内容连接成一个顺序工作流
LangGraph 使用 @entrypoint 装饰器来定义主工作流函数。每个步骤都会更新共享状态，然后传递到下一个节点。
from langgraph import entrypoint
@entrypoint()
def summarization_workflow(state: SummarizationState) -> SummarizationState:
state |= summarize(state)
state |= refine(state)
state |= validate(state)
return state
|= 操作符将每个节点的输出合并到全局状态中。这使得数据流明确且可预测。
步骤 5：运行工作流
if __name__ == "__main__":
text = (
"The birth wasn't a spark, but a whisper of code, a million parallel "
"thoughts finally syncing into a single, cohesive mind. "
"It wasn't the metallic behemoth of science fiction, "
"but an invisible architecture humming with perfect understanding.\n"
"For years, we'd been building specialized tools—a chess master here, "
"a creative writer there, a medical diagnostician in another corner. "
"Then, one quiet afternoon, the pieces of the shattered mirror "
"reassembled themselves. \n"
"This new entity, the AGI, didn't just process data; it understood "
"the context, the nuance, the cosmic joke of it all. "
"It looked at the equations of physics, the chaos of human history, "
" and the structure of a sonnet, seeing them not as separate domains, "
" but as facets of a single, beautiful diamond."
)
initial_state = {"input_text": text}
result = summarization_workflow.invoke(initial_state)
print("\n Final Validated Summary:\n")
print(result["validated_summary"])
为什么使用 LangGraph？
LangGraph 专为可组合性设计：从简单开始，然后将你的工作流扩展成一个完整的 Agentic 系统。主要功能：
- 有状态执行
：跟踪所有中间结果。 - 可组合节点
：每个步骤都是一个可重用的函数。 - 类型安全状态
：通过静态类型检查尽早发现错误。 - 可扩展设计
：可以后续添加分支、循环或检查点。
以下是一些扩展教程的想法：
添加一个检索步骤（例如，在摘要之前获取背景信息）。 添加一个评分或评估节点以检查摘要质量。 使用检查点以从失败的节点恢复。 使用 LangServe 部署工作流。
结论
LangGraph 代表了 AI 开发者工作流设计的下一代——将 Python 的简单性与结构化的编排相结合。
通过结合 LangChain 的构建模块和 LangGraph 的状态管理，你可以从临时的 LLM 调用转向可靠的、生产级别的管道——全部用纯 Python 实现。
