AI 不再“乱跑”：LangChain × LangGraph 打造可控多阶段智能流程

大型语言模型虽然强大，但真要在实际系统里用起来，光靠单次调用可不行。你需要的是多步骤推理、有状态的数据传递，还得有可靠的编排——这正是LangGraph和LangChain这对组合的用武之地。一个提供基础模块，一个提供结构化的工作流引擎。接下来，我们就用一个三步骤的文本摘要管道，一步步看看怎么实现，顺便聊聊关键参数调优和状态管理。

一个使用最新 LangChain + LangGraph APIs 创建多步骤、有状态 LLM 管道的实践指南

图片来源：Markus Spiske on Unsplash

引言

功能强大是没错，但在实际系统中，LLM很少能单打独斗。你需要的是多步骤推理、有状态的数据传递、可靠的编排——说白了，就是需要一个结构化的流程。LangGraph正是干这个的，它给LangChain加了一个工作流引擎，让每一步都能在前一步的基础上推进。接下来，咱们就用一个具体的例子——三步骤文本摘要管道，从处理文本到优化再到验证，看看整个怎么实现。

什么是 LangChain 和 LangGraph？

在动手写代码前，先理清这两个工具的角色。LangChain提供的是积木块——模型、提示、检索器、工具；LangGraph则提供把这些积木拼装起来的图纸和引擎，让它们变成可靠、可组合的管道。你可以把它想象成专为LLM设计的Airflow，不过更贴近AI推理。

设置你的环境

第一步，安装最新版的LangChain和LangGraph：

pip install -U langchain langgraph openai

还需要一个OpenAI API密钥（或其它提供者的密钥）：

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包装器初始化一个LLM，这里用GPT-5，temperature设成0.2：

from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-5", temperature=0.2)

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:nn{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:nn{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.nn"
            "TEXT:n{text}nnSUMMARY: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搞定。