LangGraph 框架

┌─────────────────────────────────────────────────────────────┐
│                    LangGraph 设计哲学                        │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   传统 Chain 的局限：                                        │
│   ┌─────────────────────────────────────────────────────┐  │
│   │  Chain 1 → Chain 2 → Chain 3 → ... → Output        │  │
│   │                                                      │  │
│   │  问题：                                              │  │
│   │  • 无法处理循环（如迭代优化）                        │  │
│   │  • 难以实现条件分支                                  │  │
│   │  • 状态管理隐式，难以调试                            │  │
│   └─────────────────────────────────────────────────────┘  │
│                                                             │
│   LangGraph 的解决方案：                                     │
│   ┌─────────────────────────────────────────────────────┐  │
│   │                    Graph 架构                        │  │
│   │                                                      │  │
│   │              ┌─────────┐                            │  │
│   │              │  START  │                            │  │
│   │              └────┬────┘                            │  │
│   │                   │                                  │  │
│   │                   ↓                                  │  │
│   │            ┌─────────────┐                          │  │
│   │      ┌────→│   Node A    │←────┐                    │  │
│   │      │     └──────┬──────┘     │                    │  │
│   │      │            │            │                    │  │
│   │      │     ┌──────┴──────┐     │                    │  │
│   │      │     ↓             ↓     │                    │  │
│   │      │ ┌───────┐   ┌───────┐   │                    │  │
│   │      │ │Node B │   │Node C │   │                    │  │
│   │      │ └───┬───┘   └───┬───┘   │                    │  │
│   │      │     │           │       │                    │  │
│   │      │     └─────┬─────┘       │                    │  │
│   │      │           ↓             │                    │  │
│   │      │     ┌───────────┐       │                    │  │
│   │      │     │ Decision  │───────┘ (循环)             │  │
│   │      │     └─────┬─────┘                            │  │
│   │      │           │ (结束)                           │  │
│   │      │           ↓                                  │  │
│   │      │     ┌─────────┐                              │  │
│   │      └────→│   END   │                              │  │
│   │            └─────────┘                              │  │
│   │                                                      │  │
│   │  优势：                                              │  │
│   │  • 支持循环和迭代                                    │  │
│   │  • 显式状态管理                                      │  │
│   │  • 灵活的控制流                                      │  │
│   │  • 易于调试和可视化                                  │  │
│   └─────────────────────────────────────────────────────┘  │
│                                                             │
└─────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────┐
│                    Chain vs Graph 对比                       │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   LangChain Chain:                                          │
│   ┌─────────────────────────────────────────────────────┐  │
│   │                                                      │  │
│   │   Input → [Chain 1] → [Chain 2] → [Chain 3] → Output│  │
│   │                                                      │  │
│   │   • 线性流程                                         │  │
│   │   • 无循环                                           │  │
│   │   • 状态隐式传递                                     │  │
│   │   • 适合简单场景                                     │  │
│   └─────────────────────────────────────────────────────┘  │
│                                                             │
│   LangGraph:                                                │
│   ┌─────────────────────────────────────────────────────┐  │
│   │                                                      │  │
│   │           ┌────────────────────┐                    │  │
│   │           │      State         │                    │  │
│   │           └─────────┬──────────┘                    │  │
│   │                     │                                │  │
│   │          ┌──────────┼──────────┐                    │  │
│   │          ↓          ↓          ↓                    │  │
│   │     [Node A]   [Node B]   [Node C]                  │  │
│   │          │          │          │                    │  │
│   │          └──────────┼──────────┘                    │  │
│   │                     ↓                                │  │
│   │              [Conditional]                           │  │
│   │                /    \                                │  │
│   │               ↓      ↓                               │  │
│   │          [Loop]   [Output]                           │  │
│   │             │                                         │  │
│   │             └────→ (回到 Node A)                      │  │
│   │                                                      │  │
│   │   • 图状流程                                         │  │
│   │   • 支持循环和分支                                   │  │
│   │   • 显式状态管理                                     │  │
│   │   • 适合复杂场景                                     │  │
│   └─────────────────────────────────────────────────────┘  │
│                                                             │
└─────────────────────────────────────────────────────────────┘

"""
LangGraph State 定义示例
"""
 
from typing import TypedDict, Annotated, List
from langgraph.graph import add_messages
 
 
# ========== 1. 基础状态定义 ==========
 
class BasicState(TypedDict):
    """基础状态"""
    messages: List[str]    # 消息列表
    count: int             # 计数器
 
 
# ========== 2. 使用 Annotated 状态 ==========
 
class AnnotatedState(TypedDict):
    """带注解的状态（自动合并消息）"""
    # add_messages 会自动合并新旧消息
    messages: Annotated[List[str], add_messages]
    current_step: str
 
 
# ========== 3. 复杂状态定义 ==========
 
from langchain_core.messages import BaseMessage
 
 
class AgentState(TypedDict):
    """Agent 完整状态"""
    # 消息历史（自动合并）
    messages: Annotated[List[BaseMessage], add_messages]
    
    # 当前步骤
    current_step: str
    
    # 工具调用历史
    tool_calls: List[dict]
    
    # 中间结果
    intermediate_steps: List[tuple]
    
    # 迭代次数
    iterations: int
    
    # 是否完成
    is_finished: bool
 
 
# ========== 4. 状态 reducer 函数 ==========
 
def merge_dicts(left: dict, right: dict) -> dict:
    """合并字典的 reducer"""
    return {**left, **right}
 
 
class StateWithReducer(TypedDict):
    """带自定义 reducer 的状态"""
    data: Annotated[dict, merge_dicts]
    messages: Annotated[List[str], add_messages]

"""
LangGraph Node 定义示例
"""
 
from typing import TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
 
 
class State(TypedDict):
    messages: list
    current_step: str
 
 
# ========== 1. 基础节点 ==========
 
def simple_node(state: State) -> dict:
    """简单节点：只处理数据"""
    # 处理状态
    new_message = f"处理了 {len(state['messages'])} 条消息"
    
    # 返回状态更新（会被合并到原状态）
    return {
        "messages": [new_message],
        "current_step": "processed"
    }
 
 
# ========== 2. LLM 节点 ==========
 
def llm_node(state: State) -> dict:
    """LLM 调用节点"""
    llm = ChatOpenAI(model="gpt-4")
    
    # 调用 LLM
    response = llm.invoke(state["messages"])
    
    # 返回 AI 消息
    return {
        "messages": [response]
    }
 
 
# ========== 3. 工具节点 ==========
 
from langchain.tools import Tool
 
 
def tool_node(state: State) -> dict:
    """工具调用节点"""
    # 获取最后一条消息（应该是 AI 的工具调用请求）
    last_message = state["messages"][-1]
    
    # 执行工具调用
    if hasattr(last_message, "tool_calls"):
        results = []
        for tool_call in last_message.tool_calls:
            # 执行工具
            result = execute_tool(tool_call)
            results.append(result)
        
        return {"messages": results}
    
    return {}
 
 
def execute_tool(tool_call):
    """执行工具"""
    # 工具执行逻辑
    return f"工具 {tool_call['name']} 执行结果"
 
 
# ========== 4. 决策节点 ==========
 
def decision_node(state: State) -> str:
    """
    决策节点：返回下一个节点的名称
    用于条件边
    """
    if state.get("is_finished"):
        return "end"
    
    if len(state["messages"]) > 10:
        return "summarize"
    
    return "continue"

"""
LangGraph 图构建示例
"""
 
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
 
 
# ========== 1. 构建简单图 ==========
 
def build_simple_graph():
    """构建简单的线性图"""
    
    # 定义状态类型
    class State(TypedDict):
        messages: list
        step: str
    
    # 创建图
    graph = StateGraph(State)
    
    # 添加节点
    graph.add_node("start", lambda s: {"step": "started"})
    graph.add_node("process", lambda s: {"step": "processed"})
    graph.add_node("end", lambda s: {"step": "finished"})
    
    # 设置入口点
    graph.set_entry_point("start")
    
    # 添加边
    graph.add_edge("start", "process")
    graph.add_edge("process", "end")
    graph.add_edge("end", END)
    
    # 编译图
    return graph.compile()
 
 
# ========== 2. 构建条件分支图 ==========
 
def build_conditional_graph():
    """构建带条件分支的图"""
    
    class State(TypedDict):
        messages: list
        route: str
    
    # 定义路由函数
    def route_decision(state: State) -> str:
        if state.get("route") == "A":
            return "node_a"
        elif state.get("route") == "B":
            return "node_b"
        else:
            return "node_c"
    
    graph = StateGraph(State)
    
    # 添加节点
    graph.add_node("start", lambda s: {"route": "A"})
    graph.add_node("node_a", lambda s: {"route": "processed_A"})
    graph.add_node("node_b", lambda s: {"route": "processed_B"})
    graph.add_node("node_c", lambda s: {"route": "processed_C"})
    
    graph.set_entry_point("start")
    
    # 添加条件边
    graph.add_conditional_edges(
        "start",
        route_decision,
        {
            "node_a": "node_a",
            "node_b": "node_b",
            "node_c": "node_c"
        }
    )
    
    # 所有路径都结束
    graph.add_edge("node_a", END)
    graph.add_edge("node_b", END)
    graph.add_edge("node_c", END)
    
    return graph.compile()
 
 
# ========== 3. 构建循环图 ==========
 
def build_loop_graph():
    """构建带循环的图"""
    
    class State(TypedDict):
        messages: list
        iterations: int
        is_finished: bool
    
    def check_continue(state: State) -> str:
        """检查是否继续循环"""
        if state["is_finished"] or state["iterations"] >= 5:
            return "end"
        return "continue"
    
    def process_node(state: State) -> dict:
        """处理节点"""
        return {
            "iterations": state["iterations"] + 1,
            "messages": [f"迭代 {state['iterations'] + 1}"]
        }
    
    def finish_node(state: State) -> dict:
        """结束节点"""
        return {"is_finished": True}
    
    graph = StateGraph(State)
    
    graph.add_node("process", process_node)
    graph.add_node("finish", finish_node)
    
    graph.set_entry_point("process")
    
    # 条件边：可以形成循环
    graph.add_conditional_edges(
        "process",
        check_continue,
        {
            "continue": "process",  # 回到 process 形成循环
            "end": "finish"
        }
    )
    
    graph.add_edge("finish", END)
    
    return graph.compile()

"""
使用 LangGraph 实现 ReAct Agent
"""
 
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain.tools import tool
 
 
# ========== 1. 定义状态 ==========
 
class ReActState(TypedDict):
    """ReAct Agent 状态"""
    messages: Annotated[List[dict], "add_messages"]
    thoughts: List[str]
    actions: List[dict]
    iterations: int
 
 
# ========== 2. 定义工具 ==========
 
@tool
def search(query: str) -> str:
    """搜索互联网获取信息"""
    # 模拟搜索
    mock_results = {
        "天气": "北京今天晴天，15-25°C",
        "新闻": "今日科技新闻：AI 持续突破",
        "股票": "苹果股价 $178.50"
    }
    for key, value in mock_results.items():
        if key in query:
            return value
    return f"未找到关于 '{query}' 的信息"
 
 
@tool
def calculate(expression: str) -> str:
    """执行数学计算"""
    try:
        return f"计算结果: {eval(expression)}"
    except Exception as e:
        return f"计算错误: {e}"
 
 
# ========== 3. 定义节点 ==========
 
def agent_node(state: ReActState) -> dict:
    """Agent 思考节点"""
    llm = ChatOpenAI(model="gpt-4")
    tools = [search, calculate]
    llm_with_tools = llm.bind_tools(tools)
    
    response = llm_with_tools.invoke(state["messages"])
    
    return {
        "messages": [response],
        "iterations": state["iterations"] + 1
    }
 
 
def should_continue(state: ReActState) -> str:
    """判断是否继续"""
    last_message = state["messages"][-1]
    
    # 如果有工具调用，继续执行
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    
    # 否则结束
    return "end"
 
 
# ========== 4. 构建图 ==========
 
def build_react_agent():
    """构建 ReAct Agent 图"""
    
    # 创建图
    workflow = StateGraph(ReActState)
    
    # 添加节点
    workflow.add_node("agent", agent_node)
    workflow.add_node("tools", ToolNode([search, calculate]))
    
    # 设置入口
    workflow.set_entry_point("agent")
    
    # 添加条件边
    workflow.add_conditional_edges(
        "agent",
        should_continue,
        {
            "tools": "tools",
            "end": END
        }
    )
    
    # 工具执行后回到 agent
    workflow.add_edge("tools", "agent")
    
    # 编译
    return workflow.compile()
 
 
# ========== 5. 使用 Agent ==========
 
if __name__ == "__main__":
    agent = build_react_agent()
    
    # 执行查询
    result = agent.invoke({
        "messages": [{"role": "user", "content": "北京今天天气怎么样？"}],
        "thoughts": [],
        "actions": [],
        "iterations": 0
    })
    
    print(result["messages"][-1].content)

"""
使用 LangGraph 实现多 Agent 协作
"""
 
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
 
 
# ========== 1. 定义状态 ==========
 
class MultiAgentState(TypedDict):
    """多 Agent 状态"""
    messages: Annotated[List, "add_messages"]
    research_result: str
    draft_result: str
    review_result: str
    iterations: int
    is_approved: bool
 
 
# ========== 2. 定义 Agent 节点 ==========
 
def researcher_node(state: MultiAgentState) -> dict:
    """研究 Agent"""
    llm = ChatOpenAI(model="gpt-4")
    
    prompt = f"""
    你是一个研究专家。请对以下主题进行深入研究：
    
    主题：{state['messages'][0]['content']}
    
    请提供：
    1. 核心概念
    2. 相关背景
    3. 关键信息点
    """
    
    response = llm.invoke([HumanMessage(content=prompt)])
    
    return {
        "research_result": response.content,
        "messages": [AIMessage(content=f"[研究完成] {response.content[:100]}...")]
    }
 
 
def writer_node(state: MultiAgentState) -> dict:
    """写作 Agent"""
    llm = ChatOpenAI(model="gpt-4")
    
    prompt = f"""
    你是一个专业写作专家。请基于以下研究结果撰写文章：
    
    研究结果：
    {state['research_result']}
    
    要求：
    1. 结构清晰
    2. 语言流畅
    3. 内容准确
    """
    
    response = llm.invoke([HumanMessage(content=prompt)])
    
    return {
        "draft_result": response.content,
        "messages": [AIMessage(content="[初稿完成]")]
    }
 
 
def reviewer_node(state: MultiAgentState) -> dict:
    """审核 Agent"""
    llm = ChatOpenAI(model="gpt-4")
    
    prompt = f"""
    你是一个专业审核专家。请审核以下文章：
    
    文章：
    {state['draft_result']}
    
    请评估：
    1. 内容准确性
    2. 结构合理性
    3. 语言流畅度
    
    如果满意，请回复 APPROVED
    如果需要修改，请回复 NEEDS_REVISION 并说明原因
    """
    
    response = llm.invoke([HumanMessage(content=prompt)])
    
    is_approved = "APPROVED" in response.content
    
    return {
        "review_result": response.content,
        "is_approved": is_approved,
        "iterations": state["iterations"] + 1,
        "messages": [AIMessage(content=f"[审核{'通过' if is_approved else '需修改'}]")]
    }
 
 
def route_review(state: MultiAgentState) -> str:
    """根据审核结果路由"""
    if state["is_approved"]:
        return "end"
    if state["iterations"] >= 3:
        return "end"  # 最多迭代 3 次
    return "revise"
 
 
# ========== 3. 构建协作图 ==========
 
def build_multi_agent_graph():
    """构建多 Agent 协作图"""
    
    workflow = StateGraph(MultiAgentState)
    
    # 添加节点
    workflow.add_node("researcher", researcher_node)
    workflow.add_node("writer", writer_node)
    workflow.add_node("reviewer", reviewer_node)
    
    # 设置入口
    workflow.set_entry_point("researcher")
    
    # 添加边
    workflow.add_edge("researcher", "writer")
    workflow.add_edge("writer", "reviewer")
    
    # 添加条件边
    workflow.add_conditional_edges(
        "reviewer",
        route_review,
        {
            "revise": "writer",  # 修改：回到 writer
            "end": END           # 通过：结束
        }
    )
    
    return workflow.compile()
 
 
# ========== 4. 使用 ==========
 
if __name__ == "__main__":
    graph = build_multi_agent_graph()
    
    result = graph.invoke({
        "messages": [HumanMessage(content="人工智能的发展历史")],
        "research_result": "",
        "draft_result": "",
        "review_result": "",
        "iterations": 0,
        "is_approved": False
    })
    
    print("最终文章：")
    print(result["draft_result"])

"""
LangGraph 状态持久化
"""
 
from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3
 
 
# ========== 1. 内存检查点 ==========
 
def with_memory_checkpoint():
    """使用内存检查点"""
    from langgraph.graph import StateGraph, END
    
    class State(TypedDict):
        count: int
    
    graph = StateGraph(State)
    graph.add_node("inc", lambda s: {"count": s["count"] + 1})
    graph.set_entry_point("inc")
    graph.add_edge("inc", END)
    
    # 编译时添加检查点
    memory = MemorySaver()
    app = graph.compile(checkpointer=memory)
    
    return app, memory
 
 
# ========== 2. SQLite 检查点 ==========
 
def with_sqlite_checkpoint(db_path="checkpoints.db"):
    """使用 SQLite 检查点"""
    conn = sqlite3.connect(db_path, check_same_thread=False)
    saver = SqliteSaver(conn)
    
    # ... 构建图
    # app = graph.compile(checkpointer=saver)
    
    return saver
 
 
# ========== 3. 使用检查点恢复执行 ==========
 
def resume_execution():
    """恢复执行示例"""
    app, memory = with_memory_checkpoint()
    
    # 第一次执行
    config = {"configurable": {"thread_id": "conversation-1"}}
    result1 = app.invoke({"count": 0}, config)
    print(f"第一次: count = {result1['count']}")  # 1
    
    # 恢复执行（使用相同的 thread_id）
    result2 = app.invoke({"count": result1['count']}, config)
    print(f"第二次: count = {result2['count']}")  # 2
 
 
# ========== 4. 时间旅行（回滚状态）==========
 
def time_travel():
    """时间旅行示例"""
    app, memory = with_memory_checkpoint()
    config = {"configurable": {"thread_id": "conversation-1"}}
    
    # 执行多次
    app.invoke({"count": 0}, config)
    app.invoke({"count": 1}, config)
    app.invoke({"count": 2}, config)
    
    # 获取历史状态
    history = list(memory.list(config))
    
    # 回滚到之前的状态
    if len(history) > 1:
        checkpoint = history[-2]  # 上上个状态
        # 可以基于此状态继续执行

"""
LangGraph 人机协作
"""
 
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
 
 
# ========== 1. 审核节点 ==========
 
def human_review_node(state):
    """人工审核节点（需要人工确认）"""
    # 这个节点不执行任何操作
    # 只是标记需要人工审核
    return {"needs_review": True}
 
 
# ========== 2. 构建需要人工确认的图 ==========
 
def build_human_in_loop_graph():
    """构建人机协作图"""
    
    class State(TypedDict):
        content: str
        needs_review: bool
        approved: bool
    
    def generate_node(state):
        return {"content": "生成的内容...", "needs_review": False}
    
    def finalize_node(state):
        return {"approved": True}
    
    workflow = StateGraph(State)
    
    workflow.add_node("generate", generate_node)
    workflow.add_node("human_review", human_review_node)
    workflow.add_node("finalize", finalize_node)
    
    workflow.set_entry_point("generate")
    workflow.add_edge("generate", "human_review")
    workflow.add_edge("human_review", "finalize")
    workflow.add_edge("finalize", END)
    
    # 编译时设置 interrupt_before
    memory = MemorySaver()
    app = workflow.compile(
        checkpointer=memory,
        interrupt_before=["human_review"]  # 在 human_review 前中断
    )
    
    return app
 
 
# ========== 3. 使用人机协作 ==========
 
if __name__ == "__main__":
    app = build_human_in_loop_graph()
    config = {"configurable": {"thread_id": "human-review-1"}}
    
    # 第一次执行（会在 human_review 前暂停）
    result = app.invoke({"content": "", "needs_review": False, "approved": False}, config)
    
    # 检查是否需要人工审核
    state = app.get_state(config)
    if state.values.get("needs_review"):
        print("需要人工审核...")
        print(f"内容: {state.values['content']}")
        
        # 人工确认后继续
        user_input = input("是否批准？(y/n): ")
        
        if user_input.lower() == 'y':
            # 继续执行
            result = app.invoke(None, config)
            print(f"最终结果: {result}")

"""
LangGraph 并行执行
"""
 
from langgraph.graph import StateGraph, END
import asyncio
 
 
# ========== 1. 并行节点定义 ==========
 
async def parallel_task_a(state):
    """并行任务 A"""
    await asyncio.sleep(1)
    return {"result_a": "任务 A 完成"}
 
 
async def parallel_task_b(state):
    """并行任务 B"""
    await asyncio.sleep(1)
    return {"result_b": "任务 B 完成"}
 
 
async def parallel_task_c(state):
    """并行任务 C"""
    await asyncio.sleep(1)
    return {"result_c": "任务 C 完成"}
 
 
# ========== 2. 构建并行图 ==========
 
def build_parallel_graph():
    """构建并行执行图"""
    
    class State(TypedDict):
        result_a: str
        result_b: str
        result_c: str
    
    workflow = StateGraph(State)
    
    # 添加并行节点
    workflow.add_node("task_a", parallel_task_a)
    workflow.add_node("task_b", parallel_task_b)
    workflow.add_node("task_c", parallel_task_c)
    
    # 汇总节点
    async def aggregate(state):
        return {"summary": f"{state['result_a']}, {state['result_b']}, {state['result_c']}"}
    
    workflow.add_node("aggregate", aggregate)
    
    # 设置入口（会并行执行多个节点）
    workflow.set_entry_point("task_a")
    workflow.set_entry_point("task_b")  # 注意：需要使用 fan-out 模式
    
    # 使用 fan-out 模式
    # 实际实现需要使用条件边或 fan-out 节点
    
    return workflow.compile()

"""
LangGraph 可视化
"""
 
from IPython.display import Image, display
from langgraph.graph import StateGraph, END
 
 
def visualize_graph(app):
    """可视化图结构"""
    try:
        # 获取图的 Mermaid 图
        mermaid = app.get_graph().draw_mermaid()
        print(mermaid)
        
        # 如果在 Jupyter 中
        # display(Image(app.get_graph().draw_mermaid_png()))
    except Exception as e:
        print(f"可视化失败: {e}")
 
 
# 生成 ASCII 图
def print_graph_ascii(app):
    """打印 ASCII 格式的图"""
    graph_str = app.get_graph().draw_ascii()
    print(graph_str)

"""
LangGraph 调试技巧
"""
 
# 1. 使用 verbose 模式
def debug_invoke(app, state, config):
    """调试执行"""
    # 打印初始状态
    print(f"初始状态: {state}")
    
    # 执行
    result = app.invoke(state, config)
    
    # 打印最终状态
    print(f"最终状态: {result}")
    
    return result
 
 
# 2. 逐步执行
def step_through(app, state, config):
    """逐步执行"""
    for event in app.stream(state, config):
        for node_name, node_output in event.items():
            print(f"节点 {node_name} 输出: {node_output}")
 
 
# 3. 检查中间状态
def inspect_state(app, config):
    """检查状态"""
    state = app.get_state(config)
    print(f"当前状态: {state.values}")
    print(f"下一步: {state.next}")

# 使用条件边实现循环
graph.add_conditional_edges(
    "process",
    check_continue,  # 返回 "continue" 或 "end"
    {
        "continue": "process",  # 回到自身
        "end": END
    }
)