一、类的定位与特点

CompiledStateGraph 是一个 可运行的智能体工作流，支持：

功能	描述
有状态(State)	所有节点共享一个 state schema
自动调度	框架决定下一个要执行哪个节点
工具调用	支持自动判断何时调用 ToolNode
条件分支	根据 state 或 judge 函数决定路线
Streaming 输出	支持部分生成（如大模型 token 流）
异步执行	适合 Web API、服务端执行
Checkpoint	错误恢复、回放、持久化
人类参与	支持 Human-in-the-loop

二、完整方法（系统结构化总结）

我按照使用频率和逻辑把所有方法分成五类。

方法分类总览

① 执行类（核心运行）

方法	说明
invoke	同步执行，输入 → 输出
ainvoke	异步执行（Web API 场景推荐）
stream	同步流式执行
astream	异步流式执行（最常用）

② 回放 / 检查点 / 状态类

方法	说明
get_state	获取执行中的状态（含 memory / checkpoint）
get_subgraphs	获取子图（如工具子图）

③ Streaming 事件类

方法	说明
astream_events	异步事件流（比 astream 更底层）
astream_log	记录执行日志（包含 token / 工具调用）
astream_final	仅监听最终输出

④ 调度、边控制类

方法	说明
_execute_graph	内部方法，执行图（你一般不用手动调用）
_get_iterator	内部迭代器，为流式输出服务

⑤ 工具相关方法

方法	说明
get_tools	获取工具节点信息
list_tools	返回可调用工具

三、核心方法详解

下面我会依次解释最关键的几个方法，并附带对应 demo。

1. invoke(input) → 同步一次性执行

适用于：

• 简单流程
• 不需要流式输出
• 不需要异步

示例

result = app.invoke({"query": "你好"})
print(result)

特点

• 等所有节点执行完才返回
• 返回最新的完整 state

2. ainvoke(input) → 异步执行

适合：

• Web API（Flask / FastAPI）
• LangServe
• 高并发场景

示例

result = await app.ainvoke({"query": "你好"})

3. stream(input) → 同步流式执行（逐步输出）

适用于：

• 终端应用
• 模型回答长文本时想边生成边显示

示例

for step in app.stream({"query": "写一首诗"}):
    print(step)

4. astream(input) → 异步流式执行

适用于：

• WebSocket / SSE
• 前端实时输出
• 前后端分离部署

示例

async for event in app.astream({"query": "写一个优美的句子"}):
    print(event)

输出结构

每个 event 是：

{
  "node": "某个节点名",
  "state": {... 当前 state ...},
  "event": "node_end / tool_start / tool_end / token" 等等
}

高级方法：astream_events

比 astream 更底层，会输出更详细的事件，包括：

• token 级别输出
• 工具调用开始/结束
• 节点切换
• 状态更新

示例

async for event in app.astream_events({"query": "天气如何"}):
    print(event['event'], event.get('data'))

四、完整 demo

这是一个 综合案例，包含：

• state schema
• step1 → 条件判断 → OK/FAIL
• 工具调用节点 ToolNode
• Streaming 输出

① 构建图结构

from langgraph.graph import StateGraph, MessagesState
from langgraph.prebuilt import ToolNode, tools_condition

# ======== 定义 State ========
class MyState(MessagesState):
    x: int = 0
    result: str = ""


# ======== 定义节点逻辑 ========
def step1(state: MyState):
    new_x = state["x"] + 3
    return {"x": new_x}

def ok(state: MyState):
    msg = {"role": "assistant", "content": "检测通过，执行工具"}
    return {
        "result": "通过检查",
        "messages": state["messages"] + [msg]
    }

def fail(state: MyState):
    msg = {"role": "assistant", "content": "检测失败，执行工具"}
    return {
        "result": "未通过检查",
        "messages": state["messages"] + [msg]
    }


# 工具（模拟）
def add_tool(value: int):
    """
    加 10 的工具函数。

    参数：
        value (int): 输入数值。

    返回：
        int: 输入值加 10。
    """
    return value + 10

tools = [add_tool]
tool_node = ToolNode(tools)

# judge 函数
def judge(state):
    return "OK" if state["x"] > 5 else "FAIL"

---

② 构建流图并编译

graph = StateGraph(MyState)

graph.add_node("step1", step1)
graph.add_node("OK", ok)
graph.add_node("FAIL", fail)
graph.add_node("tool", tool_node)

graph.set_entry_point("step1")

# 条件分支
graph.add_conditional_edges("step1", judge, {
    "OK": "OK",
    "FAIL": "FAIL"
})

# 最后统一走工具节点
graph.add_edge("OK", "tool")
graph.add_edge("FAIL", "tool")

# 编译图
app = graph.compile()

③ astream 流式运行

async for chunk in app.astream({"messages": [], "x": 2}):
    print(chunk)

输出示例（结构化）

{'step1': {'x': 5}}
{'FAIL': {'result': '未通过检查', 'messages': [{'role': 'assistant', 'content': '检测失败，执行工具'}]}}
{'tool': {'messages': []}}

五、场景级示例

① Web API 示例：Flask + astream

from flask import Flask, request, Response
import json
import asyncio

app_server = Flask(__name__)

@app_server.route("/run", methods=["POST"])
def run_graph():
    data = request.json

    async def generate():
        async for event in app.astream(data):
            yield json.dumps(event, ensure_ascii=False) + "\n"

    return Response(generate(), mimetype='text/event-stream')

② Token 级事件监听（LLM streaming）

async for e in app.astream_events({"query": "给我讲一个故事"}):
    if e["event"] == "token":
        print(e["data"], end="")

③ 获取执行状态

state = app.get_state()
print(state.values)

六、总结

类别	方法	用途
执行	invoke / ainvoke	完整执行一次
流式	stream / astream	边执行边输出
事件	astream_events	更细粒度事件
工具	list_tools	查看工具
状态	get_state	获取当前状态