GIN.
返回首页
Agent
LangChain
LangGraph
Agent

LangGraph

Gin10分钟

LangGraph 官方文档(点击进入)

LangGraph 基础

LangGraph 的核心是状态和图。

  • 状态 一个 TypedDict 类，规定图里的数据结构
  • 节点 本质是一个函数
  • 边 两个节点之间的连接
  • 图 边和节点组合成完整流程

状态

PYTHON
from typing_extensions import TypedDict
 
 
class State(TypedDict):
    message: list[str]

节点

节点本质是函数，接收 state，返回要更新的字段。

PYTHON
def node_a(state: State):
    return {"message": ["a"]}
 
 
def node_b(state: State):
    return {"message": ["b"]}

tool节点

PYTHON
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.graph import StateGraph, MessagesState, START, END
 
llm_with_tools = llm.bind_tools([get_weather])
 
def call_model(state: MessagesState) -> dict:
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}
 
graph = StateGraph(MessagesState)
graph.add_node("call_model", call_model)
graph.add_node("tools", ToolNode([get_weather]))
 
graph.add_edge(START, "call_model")
graph.add_conditional_edges("call_model", tools_condition)  # 有 tool_calls -> "tools"，没有 -> END
graph.add_edge("tools", "call_model")   # 工具执行完绕回模型，形成循环
 
app = graph.compile()
 
result = app.invoke({"messages": [HumanMessage(content="南京天气怎么样")]})
print(result["messages"][-1].content)

边

PYTHON
from langgraph.graph import StateGraph, START, END
 
graph = StateGraph(State)
 
graph.add_node("a", node_a)
graph.add_node("b", node_b)
 
graph.set_entry_point("a") # 定义入口边
graph.add_edge("a", "b")
graph.add_edge("b", END)
 
app = graph.compile()

条件边

需要一个 router 函数来保证节点名称绝对正确。

PYTHON
def router(state: State):
    if state["message"][-1] == "ok":
        return "success_node"
    return "error_node"
 
# 定义条件边
graph.add_conditional_edges(
    "check", # 节点名
    router,  # 路由
)

LangGraph 持久化

保存对话历史或中间状态。

PYTHON
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import InMemorySaver
 
 
class State(TypedDict):
    messages: Annotated[list, add_messages]
 
 
def chat_node(state: State):
    """
    如果需要拿到用户输入的消息
	state['message'][-1].content
 
    state(图数据): type:dict
    state['message']:消息列表 type:List
    state['message'][-1]:用户输入的数据 : type:langchain_core.messages.human.HumanMessage
    state['message'][-1].content: 用户输入的原始数据 type:str
    """
    return {"messages": [{"role": "assistant", "content": "收到"}]}
 
 
graph = StateGraph(State)
graph.add_node("chat", chat_node)
graph.add_edge(START, "chat")
graph.add_edge("chat", END)
 
app = graph.compile(checkpointer=InMemorySaver())
 
config = {"configurable": {"thread_id": "user_001"}}
 
app.invoke({"messages": [{"role": "user", "content": "你好"}]}, config)

LangGraph

LangGraph 流式输出

LangGraph 支持几种常用 stream mode。

  • updates 节点返回的状态更新
  • values 每一步后的完整状态
  • messages LLM token 流
  • custom 节点内部自定义流

官方文档：

updates

普通 Python 节点没有 LLM，也可以用 updates 看状态变化。

PYTHON
for part in graph.stream(
    inputs,
    stream_mode="updates",
    version="v2",
):
    print(part)

打字机

messages 打字机效果

messages 只会流出 LLM 调用产生的 token。

如果节点只是字符串拼接：

PYTHON
def node(state: State):
    return {"content": "hello"}

那么：

PYTHON
graph.stream(inputs, stream_mode="messages", version="v2")

不会打印任何东西。

原因：

  • messages 是 LLM token 流
  • 普通 Python 字符串不是模型 token
  • 要打字机效果，节点里必须调用 LLM

纯 LangGraph 打字机

PYTHON
import os
from typing import Any
 
from langchain_ollama import ChatOllama
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
 
 
class State(TypedDict):
    question: str
    msg: str
 
 
llm = ChatOllama(model=os.getenv("OLLAMA_MODEL", "qwen3:0.6b"))
 
 
def chat_node(state: State) -> dict[str, Any]:
    response = llm.invoke([
        ("system", "你是一个简洁、自然的中文助手。"),
        ("human", state["question"]),
    ])
    return {"msg": response.content}
 
 
workflow = StateGraph(State)
workflow.add_node("chat", chat_node)
workflow.add_edge(START, "chat")
workflow.add_edge("chat", END)
graph = workflow.compile()
 
inputs = {"question": "用一句话解释 LangGraph。", "msg": ""}
 
for part in graph.stream(inputs, stream_mode="messages", version="v2"):
    message_chunk, metadata = part["data"]
    if content := message_chunk.content:
        print(content, end="", flush=True)

create_agent 打字机

create_agent 返回的也是 LangGraph 图，所以写法类似。

PYTHON
import os
from langchain.agents import create_agent
from langchain_ollama import ChatOllama
 
model = ChatOllama(model=os.getenv("OLLAMA_MODEL", "qwen3:0.6b"))
 
agent = create_agent(
    model=model,
    tools=[],
    system_prompt="你是一个简洁、自然的中文助手。",
)
 
inputs = {
    "messages": [
        {"role": "user", "content": "解释 create_agent 和 LangGraph 的关系。"}
    ]
}
 
for part in agent.stream(inputs, stream_mode="messages", version="v2"):
    message_chunk, metadata = part["data"]
    if content := message_chunk.content:
        print(content, end="", flush=True)

callbacks 打字机

也可以不用 for stream，用 LangChain callback。

PYTHON
from langchain_core.callbacks import StreamingStdOutCallbackHandler
from langchain_ollama import ChatOllama
 
llm = ChatOllama(
    model="qwen3:0.6b",
    callbacks=[StreamingStdOutCallbackHandler()],
)
 
llm.invoke("你好")

区别：

  • callback 写法简单
  • for stream 更适合自己控制过滤、token 统计、只显示某个字段

token 消耗

每个 provider 的 usage 返回不一样。

Ollama 通常最后一个 chunk 才有总 usage。

PYTHON
for part in graph.stream(inputs, stream_mode="messages", version="v2"):
    message_chunk, metadata = part["data"]
 
    if content := message_chunk.content:
        print(content, end="", flush=True)
 
    if usage := message_chunk.usage_metadata:
        print("\n最终 token 消耗:", usage)

如果想在每个文字后面显示实时数量，可以自行累计。

PYTHON
count = 0
 
for part in graph.stream(inputs, stream_mode="messages", version="v2"):
    message_chunk, metadata = part["data"]
 
    if content := message_chunk.content:
        count += 1
        print(f"{content}[out≈{count}]", end="", flush=True)

注意：

  • 自己累计的是展示计数
  • provider 返回的 usage_metadata 才是最终真实用量

结构化输出 + 打字机效果

结构化输出需要完整内容解析。

打字机效果需要 token streaming。

所以最稳的做法：

  1. 流式阶段打印 token
  2. 流结束后解析完整 JSON / Pydantic 对象
PYTHON
from pydantic import BaseModel, Field
from langchain_core.output_parsers import PydanticOutputParser
 
 
class Answer(BaseModel):
    msg: str = Field(description="展示给用户看的回复")
    age: int = Field(description="年龄")
 
 
parser = PydanticOutputParser(pydantic_object=Answer)
 
full_text = ""
 
for part in graph.stream(inputs, stream_mode="messages", version="v2"):
    message_chunk, metadata = part["data"]
 
    if content := message_chunk.content:
        full_text += content
        print(content, end="", flush=True)
 
answer = parser.parse(full_text)

只让 msg 字段打字机

需求：

  • 结构化对象有 msg、age
  • 只展示 msg
  • age 不展示，只保存到 state 或最终结果

推荐图结构：

TEXT
START -> stream_msg -> structure -> END
  • stream_msg 生成并流式展示自然语言 msg
  • structure 根据 msg 和问题生成结构化对象 {msg, age}
  • 外层只打印 stream_msg 节点的 token
PYTHON
for part in graph.stream(inputs, stream_mode="messages", version="v2"):
    message_chunk, metadata = part["data"]
 
    if metadata.get("langgraph_node") != "stream_msg":
        continue
 
    if content := message_chunk.content:
        print(content, end="", flush=True)

单节点实现只流 msg

也可以在一个节点中完成。

思路：

  1. 节点内部用 msg_llm.stream() 生成 msg
  2. 用 get_stream_writer() 只把 msg 的 delta 发出去
  3. 同一个节点里再生成结构化对象 {msg, age}
  4. 外层只监听 custom
PYTHON
from typing import Any
from pydantic import BaseModel
from langgraph.config import get_stream_writer
 
 
class Answer(BaseModel):
    msg: str
    age: int
 
 
def chat_node(state: State) -> dict[str, Any]:
    writer = get_stream_writer()
 
    msg = ""
    for chunk in msg_llm.stream(state["question"]):
        if chunk.content:
            msg += chunk.content
            writer({"field": "msg", "delta": chunk.content})
 
    structured = json_llm.with_structured_output(Answer).invoke(
        f"根据问题和回复生成结构化结果。问题：{state['question']} 回复：{msg}"
    )
 
    return {
        "msg": structured.msg,
        "age": structured.age,
        "answer": structured.model_dump(),
    }

外层：

PYTHON
for part in graph.stream(inputs, stream_mode="custom", version="v2"):
    data = part["data"]
 
    if data.get("field") == "msg":
        print(data["delta"], end="", flush=True)

这个方案最接近“指定某个字段可以流式传输”。

LangGraph 本身没有 stream_field="msg" 这种参数，但 custom stream 可以自己约定字段。

流式结构化输出的限制

为什么不推荐直接流 JSON 后解析 msg

如果让模型直接流 JSON，流式过程中可能是半截内容。

TEXT
{"msg": "你
好", "age

中间片段不是合法 JSON。

所以官方也提示：

  • structured output 的 tool call args 在 streaming 过程中可能不完整
  • 渲染前要检查字段是否完整
  • 最稳的是流结束后再解析完整对象

文档：

中断（Human-in-the-Loop）

interrupt() 让 graph 在某个节点内部暂停，把数据交还给外层调用者，等待人工输入后再恢复。

依赖三个东西配合：

  • interrupt()：暂停当前节点，返回一个 payload 给外部
  • checkpointer：暂停期间保存 state，没有它无法恢复
  • Command(resume=...)：带着人类的决定恢复执行

核心认知

  • interrupt() 不是断点，不会自动弹窗口
  • 调用 invoke() 后会正常返回，只是返回结果里多了一个 "__interrupt__" 字段
  • 检测、展示、收集人工输入这些交互逻辑，全部要自己写
  • 必须配置 checkpointer，否则暂停后 state 直接丢失，无法恢复
  • 恢复执行时 thread_id 必须和暂停时一致，否则等于开了一次新对话

最小示例：手写审批节点

PYTHON
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode
from langchain_core.messages import HumanMessage, ToolMessage
 
llm_with_tools = create_llm(DEEPSEEK).bind_tools([get_weather])
 
 
def call_model(state: MessagesState) -> dict:
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}
 
 
def human_approval(state: MessagesState) -> dict:
    last_message = state["messages"][-1]
    call = last_message.tool_calls[0]
 
    decision = interrupt({
        "question": f"是否允许调用 {call['name']}，参数: {call['args']}？",
        "tool_call": call,
    })
 
    if decision == "reject":
        return {"messages": [ToolMessage(
            content="用户拒绝了此次工具调用。",
            tool_call_id=call["id"],
        )]}
    return {}  # 批准：不改变 state，往下走去真正执行工具
 
 
def route_after_model(state: MessagesState) -> str:
    last = state["messages"][-1]
    return "human_approval" if last.tool_calls else END
 
 
def route_after_approval(state: MessagesState) -> str:
    last = state["messages"][-1]
    return "call_model" if isinstance(last, ToolMessage) else "tools"
 
 
graph = StateGraph(MessagesState)
graph.add_node("call_model", call_model)
graph.add_node("human_approval", human_approval)
graph.add_node("tools", ToolNode([get_weather]))
 
graph.add_edge(START, "call_model")
graph.add_conditional_edges("call_model", route_after_model)
graph.add_conditional_edges("human_approval", route_after_approval)
graph.add_edge("tools", "call_model")
 
checkpointer = InMemorySaver()
app = graph.compile(checkpointer=checkpointer)

调用与恢复

PYTHON
config = {"configurable": {"thread_id": "session-1"}}
 
result = app.invoke(
    {"messages": [HumanMessage(content="南京天气怎么样")]},
    config=config,
)
 
while "__interrupt__" in result:
    interrupt_info = result["__interrupt__"][0]
    print("[需要审批]:", interrupt_info.value["question"])
 
    user_decision = input("approve / reject: ").strip().lower()
 
    result = app.invoke(
        Command(resume=user_decision),
        config=config,   # 必须用同一个 thread_id，否则接不上暂停的执行
    )
 
print("[最终回答]:", result["messages"][-1].content)

create_agent 内置写法：HumanInTheLoopMiddleware

不用手写审批节点和路由，官方中间件封装了这套逻辑。

PYTHON
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
 
agent = create_agent(
    model=model,
    tools=[write_file, execute_sql, get_weather],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "write_file": True,                                    # 全部决策类型都允许
                "execute_sql": {"allowed_decisions": ["approve", "reject"]},  # 不允许编辑
                "get_weather": False,                                    # 安全操作，不审批
            },
            description_prefix="Tool execution pending approval",
        ),
    ],
    checkpointer=InMemorySaver(),
)

四种决策类型：

  • approve 原样批准
  • edit 修改参数后再执行
  • reject 带反馈拒绝
  • respond 直接回应（用于"询问用户"式的工具）

常见坑

  • 忘记加 checkpointer：中断后无法恢复，报错或行为异常
  • 恢复时 thread_id 传错/传漏：等于重新开始一次新对话，之前暂停的状态找不到
  • InMemorySaver 只在当前进程内存里，进程一重启数据就没了，仅适合本地测试；生产用 PostgresSaver 等持久化实现
  • interrupt() 所在的节点，从暂停点恢复后会重新从头执行这个节点（不是从 interrupt() 那一行继续），所以节点里 interrupt() 之前的代码要保证幂等，不要放有副作用的逻辑