LangChain vs LangGraph：Agent 框架的演进与选择

LangChain（工具库）
    ├── LangGraph（图式 Agent 编排）
    ├── LangSmith（可观测性）
    └── LangServe（API 部署）

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

# Chain = Prompt + LLM + Parser
prompt = ChatPromptTemplate.from_template("解释 {topic} 给 5 岁小孩听")
llm = ChatOpenAI(model="gpt-4o")
parser = StrOutputParser()

chain = prompt | llm | parser  # 管道语法

result = chain.invoke({"topic": "量子计算"})

# 问题 1：过度抽象
# 简单的 API 调用被包装了 5 层
chain = (
    RunnablePassthrough.assign(
        context=lambda x: retriever.invoke(x["question"])
    )
    | prompt
    | llm.bind_tools(tools)
    | RunnableLambda(parse_tool_calls)
    | RunnableLambda(execute_tools)
    | llm
    | parser
)
# 调试困难：报错信息深埋在调用栈中

# 问题 2：版本碎片化
# langchain / langchain-core / langchain-community 版本不同步
# 升级一个包可能破坏其他包

# 问题 3：性能开销
# 每层抽象都有 Python 对象创建开销
# 简单任务直接用 OpenAI SDK 更快

# LangChain：链式（线性）
A → B → C → D

# LangGraph：图式（有环、有分支）
    A
   / \
  B   C
   \ / \
    D   E
    |
    F → B  (循环)

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

# 1. 定义状态
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    needs_tool: bool
    tool_result: str

# 2. 定义节点函数
def call_llm(state: AgentState):
    response = llm.invoke(state["messages"])
    return {"messages": [response], "needs_tool": bool(response.tool_calls)}

def call_tool(state: AgentState):
    results = [execute(tc) for tc in state["messages"][-1].tool_calls]
    return {"tool_result": results, "needs_tool": False}

def should_use_tool(state: AgentState):
    return "tools" if state["needs_tool"] else END

# 3. 构建图
graph = StateGraph(AgentState)

graph.add_node("llm", call_llm)
graph.add_node("tools", call_tool)

graph.set_entry_point("llm")
graph.add_conditional_edges("llm", should_use_tool)
graph.add_edge("tools", "llm")  # 工具执行后回到 LLM

# 4. 编译运行
app = graph.compile()
result = app.invoke({"messages": [{"role": "user", "content": "北京天气"}]})

from langgraph.graph import StateGraph, END

class TeamState(TypedDict):
    task: str
    research_result: str
    code_result: str
    review_result: str
    iteration: int

def researcher(state):
    """研究 Agent：搜索资料"""
    research = search_web(state["task"])
    return {"research_result": research}

def coder(state):
    """编码 Agent：写代码"""
    code = generate_code(state["task"], state["research_result"])
    return {"code_result": code}

def reviewer(state):
    """审查 Agent：代码审查"""
    review = review_code(state["code_result"])
    return {
        "review_result": review["feedback"],
        "iteration": state.get("iteration", 0) + 1,
    }

def should_continue(state):
    if "approved" in state.get("review_result", ""):
        return END
    if state.get("iteration", 0) >= 3:
        return END
    return "coder"  # 需要修改

# 构建多 Agent 图
graph = StateGraph(TeamState)
graph.add_node("researcher", researcher)
graph.add_node("coder", coder)
graph.add_node("reviewer", reviewer)

graph.set_entry_point("researcher")
graph.add_edge("researcher", "coder")
graph.add_edge("coder", "reviewer")
graph.add_conditional_edges("reviewer", should_continue)

app = graph.compile()

from langgraph.checkpoint import MemorySaver

# 编译时启用中断点
app = graph.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["reviewer"],  # 审查前暂停
)

# 运行到中断点
config = {"configurable": {"thread_id": "1"}}
result = app.invoke({"task": "实现登录功能"}, config)

# 人工审查后继续
app.update_state(config, {"review_result": "approved"})
result = app.invoke(None, config)  # 继续

# LangGraph 独有功能：回到历史状态
history = list(app.get_state_history(config))

# 回到第 3 步，修改后重新执行
old_state = history[3]
app.update_state(config, old_state.values)
result = app.invoke(None, config)

import langsmith

# LangSmith 自动追踪 LangGraph 执行
# 每个节点的输入/输出/耗时都记录

# 查看执行轨迹
trace = {
    "nodes": ["llm", "tools", "llm"],
    "total_time": "3.2s",
    "llm_calls": 2,
    "tool_calls": 1,
    "tokens": {"input": 1200, "output": 350},
}

# 评估
evaluator = langsmith.Evaluator()
results = evaluator.evaluate(
    dataset="my_test_set",
    agent=app,
    evaluators=["correctness", "helpfulness"],
)

def should_use_langgraph(scenario):
    if scenario in ["简单问答", "单次翻译", "文本摘要"]:
        return False  # 直接用 API SDK
    
    if scenario in ["RAG问答", "简单Chain"]:
        return False  # 用 LangChain 或直接 API
    
    if scenario in [
        "多步推理Agent",
        "多Agent协作",
        "需要循环的工作流",
        "人机协作审批",
        "复杂状态管理",
        "需要回滚/重试",
    ]:
        return True
    
    return False