LangChain 与 LangGraph

2022.10  LangChain 发布，聚焦 LLM 链式调用
2023.03  加入 Agent 和工具能力
2023.08  引入 LCEL（LangChain Expression Language）
2023.11  拆分为 langchain-core / langchain-community / 合作伙伴包
2024.01  LangGraph 成为官方 Agent 编排方案
2024.06  LangChain v0.2 稳定版发布
2025     LangGraph Platform 上线

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

# 每个组件都是一个 Runnable
prompt = ChatPromptTemplate.from_template("用一句话解释什么是 {topic}")
model = ChatOpenAI(model="gpt-4o")
parser = StrOutputParser()

# 用 | 组合成链
chain = prompt | model | parser

# 同步调用
result = chain.invoke({"topic": "量子计算"})

# 异步调用
result = await chain.ainvoke({"topic": "量子计算"})

# 流式输出
for chunk in chain.stream({"topic": "量子计算"}):
    print(chunk, end="", flush=True)

# 批量调用
results = chain.batch([
    {"topic": "量子计算"},
    {"topic": "区块链"},
    {"topic": "神经网络"}
])

from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver

# 内存检查点（开发用）
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# SQLite 检查点（生产用）
with SqliteSaver.from_conn_string("checkpoints.db") as checkpointer:
    app = graph.compile(checkpointer=checkpointer)

# 运行时指定 thread_id，支持断点续传
config = {"configurable": {"thread_id": "user-123"}}

# 第一次运行
result = app.invoke({"messages": ["你好"]}, config)

# 后续运行自动恢复上下文
result = app.invoke({"messages": ["继续刚才的话题"]}, config)

# 时间旅行：回溯到历史状态
history = app.get_state_history(config)
past_state = app.get_state(config, history[-2].config)  # 上一步的状态

# 从历史状态分叉
app.update_state(past_state.config, {"messages": ["换一种方法试试"]})

from langgraph.types import interrupt, Command

def sensitive_action_node(state: AgentState):
    """需要人类确认的敏感操作"""
    action = state["pending_action"]
    
    # 中断执行，等待人类确认
    human_response = interrupt({
        "question": f"确认执行以下操作？\n{action}",
        "options": ["确认", "修改", "取消"]
    })
    
    if human_response["choice"] == "确认":
        result = execute_action(action)
        return {"results": result, "status": "completed"}
    elif human_response["choice"] == "修改":
        return {"status": "needs_revision", "feedback": human_response["feedback"]}
    else:
        return {"status": "cancelled"}

# 编译时指定中断点
app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["sensitive_action"]  # 在执行此节点前暂停
)

# 运行到中断点
result = app.invoke(initial_state, config)

# 人类确认后继续
app.invoke(Command(resume={"choice": "确认"}), config)

# 将复杂逻辑封装为子图
def create_research_subgraph():
    """创建一个研究子图"""
    sub = StateGraph(ResearchState)
    sub.add_node("search", search_node)
    sub.add_node("analyze", analyze_node)
    sub.add_node("summarize", summarize_node)
    
    sub.add_edge(START, "search")
    sub.add_edge("search", "analyze")
    sub.add_edge("analyze", "summarize")
    sub.add_edge("summarize", END)
    
    return sub.compile()

# 在父图中使用子图
research = create_research_subgraph()

parent = StateGraph(MainState)
parent.add_node("research", research)  # 子图作为节点
parent.add_node("write", writing_node)
parent.add_edge(START, "research")
parent.add_edge("research", "write")
parent.add_edge("write", END)

# 多 Agent 协作图
graph = StateGraph(TeamState)

graph.add_node("researcher", researcher_agent)
graph.add_node("writer", writer_agent)
graph.add_node("reviewer", reviewer_agent)
graph.add_node("supervisor", supervisor_agent)

# Supervisor 决定下一个执行的 Agent
graph.add_conditional_edges("supervisor", route_to_agent, {
    "researcher": "researcher",
    "writer": "writer",
    "reviewer": "reviewer",
    "done": END
})

graph.add_edge(START, "supervisor")
graph.add_edge("researcher", "supervisor")
graph.add_edge("writer", "supervisor")
graph.add_edge("reviewer", "supervisor")

# 流式获取每个节点的输出
for event in app.stream({"messages": ["分析数据"]}, config, stream_mode="updates"):
    node_name = list(event.keys())[0]
    node_output = event[node_name]
    print(f"[{node_name}] {node_output}")

# 流式获取 LLM 输出（token by token）
for event in app.stream({"messages": ["分析数据"]}, config, stream_mode="messages"):
    if hasattr(event, "content"):
        print(event.content, end="", flush=True)

# langgraph.json - 项目配置
{
    "dependencies": ["."],
    "graphs": {
        "agent": "./graph.py:make_graph"
    },
    "env": ".env"
}

# 本地开发
langgraph dev          # 启动开发服务器
langgraph up           # Docker 部署

# 云端部署
langgraph deploy       # 部署到 LangGraph Cloud

LangSmith 提供：
- Trace 可视化：完整查看每一步的输入输出
- 评估框架：自动化测试 Agent 质量
- 数据集管理：管理测试用例
- 在线监控：实时查看 Agent 性能
- A/B 测试：对比不同版本的 Agent