LangChain与LangGraph：AI不再“乱跑”，打造可控多阶段智能LLM流程的实践指南

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

图片来源：Markus Spiske on Unsplash

引言

大型语言模型（LLMs）功能强大，然而，在现实世界的系统中应用它们往往需要明确的结构。单一的模型调用通常不足以满足需求；它需要多步骤推理、有状态的数据传递以及可靠的流程编排。LangGraph正是为此而生，它为LangChain提供了定义结构化、有状态工作流的能力，确保每个步骤都能在前一步的基础上有效推进。

本指南将详细介绍如何使用 LangChain 和 LangGraph APIs 构建一个顺序工作流。文章将通过一个实际例子——一个三步骤的文本摘要管道，涵盖文本处理、优化和最终输出验证——来展示其实现过程。

什么是 LangChain 和 LangGraph？

在深入代码实现之前，首先了解LangChain和LangGraph的基础概念。

LangChain 提供构建模块，包括模型、提示、检索器和工具。

LangGraph 提供工作流引擎，一种将这些构建模块连接成可靠、可组合管道的方法。

LangGraph 可以被视为专为 LangChain 设计的 LLM 工作流管理系统，专注于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：定义节点

节点是工作流的构建模块，每个节点完成一个单一任务并更新工作流状态。

本文将定义以下三个节点：

1. 摘要输入文本。
2. 优化摘要以提高清晰度。
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:

{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:

{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.

"
            "TEXT:
{text}

SUMMARY:
{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.
"
        "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. 
"
        "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("
 Final Validated Summary:
")
    print(result["validated_summary"])

为什么使用 LangGraph？

LangGraph的核心设计理念是可组合性：用户可以从简单的功能开始，逐步将工作流扩展为完整的智能体（Agent）系统。其主要功能包括：

• 有状态执行：跟踪所有中间结果。
• 可组合节点：每个步骤都是一个可重用的函数。
• 类型安全状态：通过静态类型检查尽早发现错误。
• 可扩展设计：支持后续添加分支、循环或检查点。

以下是一些建议的扩展教程思路：

• 添加一个检索步骤（例如，在摘要之前获取背景信息）。
• 添加一个评分或评估节点以检查摘要质量。
• 使用检查点以从失败的节点恢复。
• 使用 LangServe 部署工作流。

结论

LangGraph代表了AI开发者工作流设计的下一代解决方案，它将Python的简洁性与结构化的流程编排能力完美结合。

通过结合LangChain提供的构建模块和LangGraph的状态管理能力，开发者能够将临时性的LLM调用升级为可靠的、生产级别的管道，并且所有实现均基于纯Python代码。