LangGraph--CompiledStateGraph

一、类的定位与特点

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

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

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

方法分类总览

① 执行类（核心运行）

② 回放 / 检查点 / 状态类

③ Streaming 事件类

④ 调度、边控制类

⑤ 工具相关方法

三、核心方法详解

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

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

适用于：

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

示例

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

特点

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

2. ainvoke(input) → 异步执行

适合：

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

示例

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

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

适用于：

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

示例

forstepinapp.stream({"query":"写一首诗"}):print(step)

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

适用于：

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

示例

asyncforeventinapp.astream({"query":"写一个优美的句子"}):print(event)

输出结构

每个 event 是：

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

高级方法：astream_events

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

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

示例

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

四、完整 demo

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

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

① 构建图结构

fromlanggraph.graphimportStateGraph,MessagesStatefromlanggraph.prebuiltimportToolNode,tools_condition# ======== 定义 State ========classMyState(MessagesState):x:int=0result:str=""# ======== 定义节点逻辑 ========defstep1(state:MyState):new_x=state["x"]+3return{"x":new_x}defok(state:MyState):msg={"role":"assistant","content":"检测通过，执行工具"}return{"result":"通过检查","messages":state["messages"]+[msg]}deffail(state:MyState):msg={"role":"assistant","content":"检测失败，执行工具"}return{"result":"未通过检查","messages":state["messages"]+[msg]}# 工具（模拟）defadd_tool(value:int):""" 加 10 的工具函数。 参数： value (int): 输入数值。 返回： int: 输入值加 10。 """returnvalue+10tools=[add_tool]tool_node=ToolNode(tools)# judge 函数defjudge(state):return"OK"ifstate["x"]>5else"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 流式运行

asyncforchunkinapp.astream({"messages":[],"x":2}):print(chunk)

输出示例（结构化）

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

五、场景级示例

① Web API 示例：Flask + astream

fromflaskimportFlask,request,Responseimportjsonimportasyncio app_server=Flask(__name__)@app_server.route("/run",methods=["POST"])defrun_graph():data=request.jsonasyncdefgenerate():asyncforeventinapp.astream(data):yieldjson.dumps(event,ensure_ascii=False)+"\n"returnResponse(generate(),mimetype='text/event-stream')

② Token 级事件监听（LLM streaming）

asyncforeinapp.astream_events({"query":"给我讲一个故事"}):ife["event"]=="token":print(e["data"],end="")

③ 获取执行状态

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

六、总结