name: langgraph-state-graph description: 使用LangGraph的StateGraph模式构建有状态智能体工作流。设计状态模式、创建节点、定义带条件路由的边，并启用持久化。 allowed-tools: Read, Grep, Write, Edit, Bash, Glob, WebFetch

langgraph-state-graph

使用LangGraph的StateGraph模式构建有状态的智能体工作流。设计状态模式、创建节点、定义带条件路由的边，并启用持久化。

概述

LangGraph是一个用于构建有状态、多参与者应用程序的库。StateGraph是其核心抽象，支持：

循环计算图（不同于DAG）
状态持久化和检查点
人机交互模式
条件分支和路由
多智能体协调

功能

状态模式设计

使用TypedDict或Pydantic定义类型化状态模式
配置消息传递的状态通道
设置状态更新的归约函数
设计对话历史的累加器模式

图构建

将节点创建为函数或可运行对象
定义边（普通边、条件边、入口点）
配置开始和结束节点
实现条件边的路由逻辑

持久化与检查点

配置检查点后端（SQLite、PostgreSQL、Redis）
在每一步启用状态快照
支持恢复中断的工作流
基于线程的对话持久化

人机交互

在工作流中插入中断点
在继续之前收集人工反馈
支持审批门和输入收集
使用更新后的状态从中断处恢复

使用

基本StateGraph模式

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages

# 定义状态模式
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    current_step: str
    iteration: int

# 创建节点
def agent_node(state: AgentState) -> AgentState:
    # 处理状态并返回更新
    return {"current_step": "processed", "iteration": state["iteration"] + 1}

def tool_node(state: AgentState) -> AgentState:
    # 基于智能体决策执行工具
    return {"current_step": "tools_executed"}

# 构建图
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)

# 定义边
graph.set_entry_point("agent")
graph.add_edge("agent", "tools")
graph.add_conditional_edges(
    "tools",
    lambda state: "end" if state["iteration"] >= 3 else "continue",
    {"end": END, "continue": "agent"}
)

# 编译
app = graph.compile()

条件路由

def router(state: AgentState) -> str:
    """基于状态条件进行路由。"""
    last_message = state["messages"][-1]

    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    elif state["iteration"] >= state.get("max_iterations", 10):
        return "end"
    else:
        return "agent"

graph.add_conditional_edges(
    "agent",
    router,
    {
        "tools": "tool_executor",
        "agent": "agent",
        "end": END
    }
)

带检查点的持久化

from langgraph.checkpoint.sqlite import SqliteSaver

# 配置检查点器
memory = SqliteSaver.from_conn_string(":memory:")

# 编译并启用持久化
app = graph.compile(checkpointer=memory)

# 使用thread_id运行以实现持久化
config = {"configurable": {"thread_id": "conversation-1"}}
result = app.invoke(initial_state, config=config)

# 从检查点恢复
result = app.invoke(None, config=config)  # 从最后状态继续

人机交互

from langgraph.graph import StateGraph

graph = StateGraph(AgentState)
# ... 添加节点 ...

# 编译并设置中断点
app = graph.compile(
    checkpointer=memory,
    interrupt_before=["tool_executor"]  # 在工具执行前暂停
)

# 首次调用 - 在中断处暂停
result = app.invoke(initial_state, config)

# 人工批准后，恢复执行
result = app.invoke(None, config)  # 继续通过中断点

任务定义

const langgraphStateGraphTask = defineTask({
  name: 'langgraph-state-graph-design',
  description: '设计和实现LangGraph StateGraph工作流',

  inputs: {
    workflowName: { type: 'string', required: true },
    stateSchema: { type: 'object', required: true },
    nodes: { type: 'array', required: true },
    edges: { type: 'array', required: true },
    enablePersistence: { type: 'boolean', default: true },
    interruptPoints: { type: 'array', default: [] }
  },

  outputs: {
    graphCode: { type: 'string' },
    stateSchemaCode: { type: 'string' },
    compiledGraph: { type: 'boolean' },
    artifacts: { type: 'array' }
  },

  async run(inputs, taskCtx) {
    return {
      kind: 'skill',
      title: `设计StateGraph: ${inputs.workflowName}`,
      skill: {
        name: 'langgraph-state-graph',
        context: {
          workflowName: inputs.workflowName,
          stateSchema: inputs.stateSchema,
          nodes: inputs.nodes,
          edges: inputs.edges,
          enablePersistence: inputs.enablePersistence,
          interruptPoints: inputs.interruptPoints,
          instructions: [
            '分析工作流需求和状态需求',
            '设计具有适当类型的状态模式',
            '创建具有状态转换的节点函数',
            '定义边和条件路由逻辑',
            '如果启用则配置持久化',
            '为人机交互添加中断点',
            '编译并验证图'
          ]
        }
      },
      io: {
        inputJsonPath: `tasks/${taskCtx.effectId}/input.json`,
        outputJsonPath: `tasks/${taskCtx.effectId}/result.json`
      }
    };
  }
});

langgraph状态图构建器 langgraph-state-graph

name: langgraph-state-graph description: 使用LangGraph的StateGraph模式构建有状态智能体工作流。设计状态模式、创建节点、定义带条件路由的边，并启用持久化。 allowed-tools: Read, Grep, Write, Edit, Bash, Glob, WebFetch

langgraph-state-graph

概述

功能

状态模式设计

图构建

持久化与检查点

人机交互

使用

基本StateGraph模式

条件路由

带检查点的持久化

人机交互

任务定义

适用流程

外部依赖

参考资料

相关技能

相关智能体