2.5 LangGraph 与 LangChain 的集成模式

2.5 LangGraph 与 LangChain 的集成模式

复用而非重写

本节在全册位置：讲清生态边界。LangGraph 站在 LangChain 之上，模型、提示、检索器、工具都是 LangChain 原语，直接当节点用。集成模式有两种：把 Runnable 当节点，或把预置 ReAct 智能体当起点再改装。我们强烈建议从现有 Runnable 起步，而不是把已验证的链拆了重写。

模式：Runnable 即节点

任何 LangChain Runnable 都能用 add_node 直接挂进图，因为节点契约就是「输入状态、输出增量」。ToolNode 与 tools_condition 来自 langgraph.prebuilt，免写样板。create_react_agent 能一键生成带工具的循环智能体，再用 add_conditional_edges 加业务分支。类比金融：LangChain 是清算组件库，LangGraph 是编排这些组件的流程引擎。复用成本低的本质是契约统一。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
from langchain_core.messages import add_messages

class S(TypedDict):
    msgs: Annotated[list, add_messages]

model = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_messages([("system", "你简洁回答"), ("user", "{q}")])
chain = prompt | model

def call(s: S):
    return {"msgs": [chain.invoke({"q": s["msgs"][-1].content})]}

b = StateGraph(S)
b.add_node("call", call)
b.add_edge(START, "call")
b.add_edge("call", END)
g = b.compile()
​

# 更省事：用预置 ReAct 智能体做底，再接业务边
from langgraph.prebuilt import create_react_agent
from langchain_core.tools import tool

@tool
def calc(x: int) -> int:
    """平方"""
    return x * x

agent = create_react_agent(model, [calc])
# agent 本身就是一个编译好的图，可直接 invoke，也可再包一层边
​

案例：复用检索器

背景：已有 LangChain 向量检索器，想接入有环研究流程，不想为图重写一遍检索逻辑。

操作：把检索器包成节点，条件边决定「够不够」再检索还是生成。

结果：检索器零改造，仅作为图中的一个节点，迁移成本接近零。

解读：集成成本低的本质是契约统一：Runnable 即节点，旧资产直接进图。

变式：检索节点也可换成子图（见 2.2），便于加重写与重排，旧检索器作为子图内部一环。

工程清单

• 任意 LangChain Runnable 都能直接当节点挂进图，契约统一是低迁移成本的根源。

• ToolNode 与 tools_condition 免去样板，是集成最快的路径，先跑通再改装。

• create_react_agent 一键生成带工具的循环智能体，可作为改装起点而非黑盒。

常见误区

重复造轮子把 LangChain 组件又包一层链，既丢生态又引入多余状态转换；复用现有 Runnable 比自己重写更稳更快。

常用 Runnable 组件与节点化对照

LangChain 生态里最常见的可复用件基本都能直接进图，下表给出组件、节点化方式和注意点：

「契约是状态入、增量出」这句话是集成的总开关：只要函数签名是输入一个状态、输出一个增量 dict，任何 LangChain 组件都能包进去。集成前先看这个组件能不能满足这条签名，不满足就在外面垫一层适配函数，而不是改写组件本身。

把既有链迁进图的迁移模板

手头已有一条验证过的链，迁进图有三步，缺一不可：

# 第一步：定义状态，链需要的输入输出都成为字段
class S(TypedDict):
    question: str
    answer: str

# 第二步：把链包成节点，输入从状态读，输出写回状态
def run_chain(s: S):
    return {"answer": old_chain.invoke({"q": s["question"]})}

# 第三步：注册并连线，链原来的错误处理可以升级为降级边
b = StateGraph(S)
b.add_node("qa", run_chain)
b.add_edge(START, "qa")
b.add_edge("qa", END)
​

迁移时的两个提醒：链里的中间变量不要塞进状态，保持旧链内部结构不变；链的输入输出如果和状态字段名不一致，在包节点时做映射，别为了对齐字段去改链的 prompt。这样迁移风险最小，回滚也只需要切回旧调用入口。

create_react_agent 的可扩展点

预置智能体不是黑盒，它的内部就是一张图，有三处可以改装：

agent = create_react_agent(model, tools, prompt=system_prompt)

# 可扩展点一：加状态字段，再追加业务节点
# 可扩展点二：加条件边，在工具循环外接业务分支
# 可扩展点三：替换 tools_condition，用自定义路由控制循环
​

先看 agent 的结构再决定改哪：agent.get_graph() 能把内部图打出来，确认 tool 节点和 agent 节点的名字，再做 add_node、add_conditional_edges 扩展。最常犯的错是不知道内部节点名就硬接边，结果边指向不存在的节点，编译期就报错——所以第一步永远是先看结构。