LangGraph 深度解析：基于图的工作流引擎如何重塑 Agent 开发

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

class AgentState(TypedDict):
    messages: Annotated[list, add]  # 消息列表自动追加
    tool_results: list
    iteration: int

def call_model(state: AgentState):
    response = llm.invoke(state["messages"])
    return {"messages": [response], "iteration": state["iteration"] + 1}

def should_continue(state: AgentState):
    if state["iteration"] > 5:
        return END
    return "tools"

# 构建图
graph = StateGraph(AgentState)
graph.add_node("agent", call_model)
graph.add_node("tools", tool_node)
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue)
graph.add_edge("tools", "agent")

app = graph.compile()

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

class ResearchState(TypedDict):
    topic: str
    search_results: list
    findings: str
    needs_more: bool
    depth: int

def search_web(state):
    results = search_tool(state["topic"])
    return {"search_results": results, "depth": state["depth"] + 1}

def analyze(state):
    prompt = f"分析以下搜索结果，提炼关键发现：{state['search_results']}"
    findings = llm.invoke(prompt)
    return {"findings": findings}

def decide(state):
    if "需要更多信息" in state["findings"] and state["depth"] < 3:
        return {"needs_more": True}
    return {"needs_more": False}

def route(state):
    return "search" if state["needs_more"] else END

# 构建带检查点的图
graph = StateGraph(ResearchState)
graph.add_node("search", search_web)
graph.add_node("analyze", analyze)
graph.add_node("decide", decide)

graph.set_entry_point("search")
graph.add_edge("search", "analyze")
graph.add_edge("analyze", "decide")
graph.add_conditional_edges("decide", route)

# 编译时加 checkpoint，支持暂停/恢复
app = graph.compile(checkpointer=MemorySaver())

def review_node(state):
    # 暂停等待人工输入
    human_feedback = interrupt("请审核分析结果，是否继续？")
    if human_feedback == "approve":
        return {"approved": True}
    return {"approved": False, "feedback": human_feedback}