与LangGraph集成

LangChain代理与LangGraph的关系

LangChain的create_agent函数实际上是基于LangGraph构建的。理解这个关系对于构建复杂应用很重要。

LangChain层（高级抽象）
        ↑
    create_agent()
        ↓
LangGraph层（低级编排）
        ↓
执行引擎

何时迁移到LangGraph

迁移条件

场景1：需要复杂的条件逻辑

# ❌ LangChain 难以表达
# 如果用户说"转账"，需要进行复杂的条件检查：
# - 检查余额
# - 验证收款人
# - 审查过去的转账历史
# - 根据金额进行不同级别的审批

# ✅ 使用LangGraph更清晰
from langgraph.graph import StateGraph

def should_require_approval(state):
    amount = state["transfer_amount"]
    return amount > 5000

graph = StateGraph(CommandState)
graph.add_conditional_edges(
    "validate_transfer",
    should_require_approval,
    {
        True: "approve",
        False: "execute"
    }
)

场景2：需要并行处理

# ❌ LangChain 顺序执行
# 代理一次处理一个任务

# ✅ LangGraph 支持并行
from langgraph.graph import StateGraph

# 并行执行多个工具
def parallel_research(state):
    # 同时调用多个搜索工具
    pass

场景3：需要循环或自定义流程

# ❌ LangChain 固定的代理循环

# ✅ LangGraph 自定义循环
graph.add_edge("step1", "step2")
graph.add_edge("step2", "decision_point")
graph.add_conditional_edges(
    "decision_point",
    routing_function,
    {
        "continue": "step3",
        "retry": "step1",
        "end": END
    }
)

从LangChain迁移到LangGraph

1. 简单的代理转换

LangChain版本：

agent = create_agent(
    model="anthropic:claude-sonnet-4",
    tools=[get_weather, get_forecast],
    system_prompt="你是天气助手"
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "明天天气?"}]
})

LangGraph等价版本：

from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-sonnet-4")
llm_with_tools = llm.bind_tools([get_weather, get_forecast])

def model_node(state: MessagesState):
    messages = state["messages"]
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}

def should_continue(state: MessagesState):
    messages = state["messages"]
    last_message = messages[-1]
    
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    return END

def tool_node(state: MessagesState):
    messages = state["messages"]
    last_message = messages[-1]
    
    tool_results = []
    for tool_call in last_message.tool_calls:
        tool_name = tool_call["name"]
        tool_input = tool_call["args"]
        
        # 执行工具
        result = tools_dict[tool_name].invoke(tool_input)
        tool_results.append({
            "type": "tool",
            "content": result,
            "tool_call_id": tool_call["id"],
            "name": tool_name
        })
    
    return {"messages": tool_results}

# 构建图
graph = StateGraph(MessagesState)

graph.add_node("model", model_node)
graph.add_node("tools", tool_node)

graph.add_edge(START, "model")
graph.add_conditional_edges("model", should_continue, {
    "tools": "tools",
    END: END
})
graph.add_edge("tools", "model")

app = graph.compile()

result = app.invoke({
    "messages": [{"role": "user", "content": "明天天气?"}]
})

2. 包装LangChain代理为LangGraph节点

from langgraph.graph import StateGraph, START, END
from typing import TypedDict

class AgentState(TypedDict):
    question: str
    result: str

# 使用LangChain代理
langchain_agent = create_agent(
    model="anthropic:claude-sonnet-4",
    tools=[...]
)

def agent_node(state: AgentState):
    """将LangChain代理包装为LangGraph节点"""
    result = langchain_agent.invoke({
        "messages": [{"role": "user", "content": state["question"]}]
    })
    
    return {
        "result": result["messages"][-1]["content"]
    }

# 在LangGraph中使用
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_edge(START, "agent")
graph.add_edge("agent", END)

app = graph.compile()

LangGraph的优势

1. 完全的控制

在LangGraph中，你可以控制执行的每一个方面：

# 自定义节点执行
def custom_node(state):
    # 在执行前做处理
    print(f"处理状态: {state}")
    
    # 执行任何Python代码
    result = custom_logic(state)
    
    # 修改状态
    state["processed"] = True
    
    return state

# 自定义条件边
def complex_routing(state):
    if state["user_type"] == "premium":
        if state["request_complexity"] > 5:
            return "advanced_path"
    return "basic_path"

2. 状态管理

from typing import Annotated, TypedDict
from operator import add

class AgentState(TypedDict):
    messages: list
    context: dict
    tool_calls: Annotated[list, add]  # 使用reducer自动聚合
    iteration: int

# 状态自动管理
def node1(state: AgentState):
    state["iteration"] += 1
    return state

3. 可视化

from PIL import Image

# LangGraph可以生成流程图
graph = compiled_graph
image = graph.get_graph().draw_mermaid_png()

# 或者获取Mermaid格式
mermaid_code = graph.get_graph().to_mermaid()

混合使用

在大型应用中，可以混合使用LangChain和LangGraph：

# 简单的对话用LangChain
simple_agent = create_agent(
    model="anthropic:claude-sonnet-4",
    tools=[simple_tools]
)

# 复杂的自定义流程用LangGraph
complex_graph = StateGraph(ComplexState)
# ...构建图...

# 在LangGraph中调用LangChain代理
def langchain_node(state):
    result = simple_agent.invoke({...})
    return process_result(result)

complex_graph.add_node("langchain_step", langchain_node)

最佳实践

• 使用LangChain处理标准的代理工作流
• 使用LangGraph处理复杂的多步骤编排
• 对于简单应用，优先使用LangChain
• 需要自定义时才迁移到LangGraph
• 使用中间层包装LangChain代理，便于未来迁移

常见陷阱

陷阱1：过度工程化

# ❌ 简单任务过度使用LangGraph
# LangChain完全足够

# ✅ 保持简单
agent = create_agent(...)

陷阱2：忘记处理无限循环

# ❌ 可能导致无限循环
def should_continue(state):
    return "continue"  # 总是真

graph.add_conditional_edges("node", should_continue)

# ✅ 总是包含停止条件
def should_continue(state):
    if state["iterations"] >= MAX_ITERATIONS:
        return END
    return "continue"

陷阱3：状态爆炸

# ❌ 状态包含过多数据
class HugeState(TypedDict):
    all_documents = [...]  # 10GB的数据

# ✅ 只保存必要的数据
class MinimalState(TypedDict):
    document_ids: list
    results: dict

迁移检查清单

在从LangChain迁移到LangGraph前检查：

• ✅ 需要超过5个步骤？
• ✅ 需要条件分支？
• ✅ 需要并行执行？
• ✅ 需要自定义循环逻辑？
• ✅ 性能是否是关键因素？

如果大多数是"是"，那么LangGraph是更好的选择。