LangGraph 2026 深度指南：构建生产级 AI Agent 工作流

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

# 定义共享状态
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    current_step: str
    tool_results: dict
    final_answer: str

# 创建图
graph_builder = StateGraph(AgentState)

# 添加节点
graph_builder.add_node("planner", planner_node)
graph_builder.add_node("executor", executor_node)
graph_builder.add_node("reviewer", reviewer_node)

# 添加边
graph_builder.add_edge("planner", "executor")
graph_builder.add_conditional_edges(
    "executor",
    route_based_on_result,
    {"continue": "planner", "done": "reviewer"}
)
graph_builder.add_edge("reviewer", END)

# 设置入口
graph_builder.set_entry_point("planner")

# 编译
graph = graph_builder.compile()

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator

# 定义状态：messages 追加，其他字段覆盖
class ResearchState(TypedDict):
    query: str
    messages: Annotated[List[str], operator.add]  # 追加模式
    findings: Annotated[List[str], operator.add]  # 追加模式
    summary: str  # 覆盖模式
    step_count: int
    max_steps: int

def research_node(state: ResearchState) -> dict:
    """研究节点：执行搜索并追加结果"""
    query = state["query"]
    # 模拟搜索
    findings = search_web(query)
    return {
        "findings": findings,
        "messages": [f"完成搜索: {query}"],
        "step_count": state["step_count"] + 1
    }

def should_continue(state: ResearchState) -> str:
    """条件路由：判断是否继续研究"""
    if state["step_count"] >= state["max_steps"]:
        return "summarize"
    if len(state["findings"]) >= 10:
        return "summarize"
    return "research"

builder = StateGraph(ResearchState)
builder.add_node("research", research_node)
builder.add_node("summarize", summarize_node)
builder.add_edge("research", "research")  # 自循环
builder.add_conditional_edges(
    "research",
    should_continue,
    {"research": "research", "summarize": "summarize"}
)
builder.add_edge("summarize", END)
builder.set_entry_point("research")
app = builder.compile()

from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.postgres import PostgresSaver

# 测试环境：内存存储
memory = MemorySaver()
app = builder.compile(checkpointer=memory)

# 生产环境：PostgreSQL 存储
pg = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost:5432/langgraph"
)
app_prod = builder.compile(checkpointer=pg)

# 带中断的编译：在 human_review 节点前暂停
app_with_interrupt = builder.compile(
    checkpointer=pg,
    interrupt_before=["human_review"]
)

# 执行
config = {"configurable": {"thread_id": "user-123"}}
result = app_with_interrupt.invoke(
    {"query": "研究 AI Agent 安全趋势"},
    config
)
# 返回中断状态，等待人类审批

# 人类审批后继续
approved_result = app_with_interrupt.invoke(None, config)

# 时间旅行：查看历史状态
config = {"configurable": {"thread_id": "user-123"}}

# 获取所有检查点
checkpoints = list(app.get_state_history(config))
for cp in checkpoints:
    print(f"Step: {cp['metadata']['step']}")
    print(f"State: {cp['values']}")

# 回到特定步骤
target_step = 3
state_at_step = app.get_state(config, checkpoint_id=checkpoints[target_step]['id'])
print(f"State at step 3: {state_at_step['values']}")

# 从历史步骤分叉执行
fork_config = {
    "configurable": {
        "thread_id": "user-123-fork",
        "checkpoint_id": checkpoints[target_step]['id']
    }
}
forked_result = app.invoke(None, fork_config)

# 子图：研究 Agent
class ResearchSubState(TypedDict):
    query: str
    findings: Annotated[List[str], operator.add]

research_builder = StateGraph(ResearchSubState)
research_builder.add_node("search", search_node)
research_builder.add_node("analyze", analyze_node)
research_builder.add_edge("search", "analyze")
research_builder.set_entry_point("search")
research_subgraph = research_builder.compile()

# 父图：主编 Agent
class EditorState(TypedDict):
    topic: str
    research_results: List[str]
    draft: str
    final_article: str

editor_builder = StateGraph(EditorState)

# 将子图添加为节点
editor_builder.add_node("research", research_subgraph)
editor_builder.add_node("write", write_node)
editor_builder.add_node("review", review_node)

# 定义状态映射
def map_input(state: EditorState) -> dict:
    return {"query": state["topic"]}

def map_output(sub_result: dict) -> dict:
    return {"research_results": sub_result["findings"]}

editor_builder.add_edge("research", "write")
editor_builder.add_edge("write", "review")
editor_builder.add_edge("review", END)
editor_builder.set_entry_point("research")
editor_graph = editor_builder.compile()

# 节点级流式：每完成一个节点产出一个快照
for event in app.stream({"query": "最新 AI 趋势"}):
    for node_name, node_output in event.items():
        print(f"节点 {node_name} 完成")
        if "findings" in node_output:
            print(f"找到 {len(node_output['findings'])} 条结果")

# Token 级流式：LLM 逐字输出
async for event in app.astream_events(
    {"query": "最新 AI 趋势"},
    version="v2"
):
    kind = event["event"]
    if kind == "on_chat_model_stream":
        # LLM 产生的单个 token
        token = event["data"]["chunk"].content
        print(token, end="", flush=True)
    elif kind == "on_tool_start":
        print(f"\n[调用工具: {event['name']}]")
    elif kind == "on_tool_end":
        print(f"\n[工具完成]")

// 前端 WebSockets 流式输出（Node.js 示例）
import { streamToWebSocket } from './stream-utils';

// 将 LangGraph 流式输出推送到前端
app.on('connection', (ws) => {
  ws.on('message', async (query) => {
    const config = { configurable: { thread_id: generateId() } };
    
    for await (const event of graph.stream(
      { query },
      { streamMode: 'values' }
    )) {
      ws.send(JSON.stringify({
        type: 'node_complete',
        node: Object.keys(event)[0],
        state: event
      }));
    }
    
    ws.send(JSON.stringify({ type: 'complete' }));
  });
});

# 2026 新特性：Human 节点
from langgraph.types import Human

class ApprovalState(TypedDict):
    proposal: str
    risk_assessment: str
    human_decision: str
    approved: bool

builder = StateGraph(ApprovalState)
builder.add_node("analyze", analyze_risk)

# Human 节点：等待人类审批
builder.add_node(
    "approve",
    Human(
        schema={"decision": {"type": "string", "enum": ["approve", "reject", "modify"]},
                "reason": {"type": "string"}},
        timeout=3600,  # 1 小时超时
        auto_fallback="reject"  # 超时自动拒绝
    )
)

builder.add_node("execute", execute_if_approved)
builder.add_node("notify", notify_result)

builder.add_edge("analyze", "approve")
builder.add_conditional_edges(
    "approve",
    lambda s: "execute" if s["human_decision"] == "approve" else "notify",
    {"execute": "execute", "notify": "notify"}
)
builder.add_edge("execute", "notify")
builder.add_edge("notify", END)
builder.set_entry_point("analyze")

from langgraph.config import RetryPolicy
from langgraph.types import Command

# 节点级错误处理：重试 + 降级
builder.add_node(
    "llm_call",
    llm_node,
    retry=RetryPolicy(
        max_attempts=3,
        backoff_factor=2,
        retry_on=[TimeoutError, ConnectionError]
    ),
    fallback=degraded_response  # 降级函数
)

# 自定义回调：推送到监控系统
class MonitoringCallback(BaseCallbackHandler):
    def on_chain_end(self, outputs, **kwargs):
        metrics.increment("agent.node.completed")
        metrics.histogram(
            "agent.node.duration",
            kwargs.get("run_time", 0)
        )
    
    def on_tool_end(self, output, **kwargs):
        metrics.increment(f"agent.tool.{kwargs['name']}.called")

app = builder.compile(
    checkpointer=pg_saver,
    interrupt_before=["human_review"]
)

# 带回调执行
result = app.invoke(
    {"query": "分析市场趋势"},
    config={
        "configurable": {"thread_id": "session-1"},
        "callbacks": [MonitoringCallback()]
    }
)

# 完整的三阶段 Agent 系统
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator

# === 共享状态 ===
class PipelineState(TypedDict):
    topic: str
    research_findings: Annotated[List[dict], operator.add]
    draft: str
    review_feedback: str
    revisions: int
    max_revisions: int
    final_article: str
    approved: bool

# === 研究子图 ===
class ResearchState(TypedDict):
    topic: str
    findings: Annotated[List[dict], operator.add]
    search_count: int

def search_and_analyze(state: ResearchState) -> dict:
    results = search_topic(state["topic"])
    analyzed = [analyze_source(r) for r in results]
    return {"findings": analyzed, "search_count": state["search_count"] + 1}

def needs_more_research(state: ResearchState) -> str:
    if state["search_count"] >= 3:
        return "done"
    if len(state["findings"]) >= 5:
        return "done"
    return "continue"

research_builder = StateGraph(ResearchState)
research_builder.add_node("search", search_and_analyze)
research_builder.add_conditional_edges(
    "search", needs_more_research,
    {"continue": "search", "done": END}
)
research_builder.set_entry_point("search")
research_graph = research_builder.compile()

# === 父图 ===
builder = StateGraph(PipelineState)

# 添加子图作为节点
builder.add_node("research", research_graph)
builder.add_node("write", write_draft)
builder.add_node("review", review_draft)
builder.add_node("revise", revise_draft)
builder.add_node("publish", publish_article)

# 流程控制
builder.add_edge("research", "write")
builder.add_edge("write", "review")
builder.add_conditional_edges(
    "review",
    lambda s: "revise" if s["revisions"] < s["max_revisions"] else "publish",
    {"revise": "revise", "publish": "publish"}
)
builder.add_edge("revise", "review")
builder.add_edge("publish", END)
builder.set_entry_point("research")

# 编译（带检查点和审核中断）
pipeline = builder.compile(
    checkpointer=PostgresSaver.from_conn_string(DB_URL),
    interrupt_before=["publish"]
)