LangGraph / LangChain 详解

最简单的 LangGraph Agent——LLM 自主决定何时调用工具、何时结束。

# pip install langgraph langchain-openai

from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

# 定义工具
def get_weather(city: str) -> str:
    """获取城市天气"""
    data = {"北京": "晴天 28°C", "上海": "多云 25°C", "深圳": "雷阵雨 30°C"}
    return data.get(city, f"未找到 {city} 的天气数据")

def search_flights(origin: str, destination: str) -> str:
    """搜索航班"""
    return f"{origin} → {destination}: MU5101 08:00 ¥980 / CA1234 14:30 ¥1200"

# 创建 Agent（一行搞定）
agent = create_react_agent(
    model=ChatOpenAI(model="gpt-4o"),
    tools=[get_weather, search_flights],
)

# 运行
result = agent.invoke({
    "messages": [{"role": "user", "content": "我想从北京飞上海，帮我看看天气和航班"}]
})

for msg in result["messages"]:
    print(f"[{msg.type}] {msg.content[:200]}")

手动定义图结构，精确控制流转逻辑。以客服意图分类为例。

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

# 1. 定义状态
class OrderState(TypedDict):
    user_input: str
    intent: str        # "refund" | "exchange" | "inquiry"
    response: str

# 2. 定义节点
def classify_intent(state: OrderState) -> dict:
    """意图分类"""
    resp = llm.invoke(
        f"将以下客服请求分类为 refund/exchange/inquiry，只返回分类词：\n{state['user_input']}"
    )
    return {"intent": resp.content.strip().lower()}

def handle_refund(state: OrderState) -> dict:
    return {"response": f"已为您发起退款流程。原始请求：{state['user_input']}"}

def handle_exchange(state: OrderState) -> dict:
    return {"response": f"已为您创建换货工单。原始请求：{state['user_input']}"}

def handle_inquiry(state: OrderState) -> dict:
    resp = llm.invoke(f"作为客服回答：{state['user_input']}")
    return {"response": resp.content}

# 3. 路由函数
def route_by_intent(state: OrderState) -> Literal["refund", "exchange", "inquiry"]:
    return state["intent"]

# 4. 构建图
graph = StateGraph(OrderState)
graph.add_node("classify", classify_intent)
graph.add_node("refund", handle_refund)
graph.add_node("exchange", handle_exchange)
graph.add_node("inquiry", handle_inquiry)

graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route_by_intent, {
    "refund": "refund",
    "exchange": "exchange",
    "inquiry": "inquiry",
})
graph.add_edge("refund", END)
graph.add_edge("exchange", END)
graph.add_edge("inquiry", END)

# 5. 编译 & 运行
app = graph.compile()
result = app.invoke({"user_input": "我买的耳机坏了，想退款"})
print(result["response"])

在关键节点暂停执行，等待人工确认后继续。适合转账、发布等高风险操作。

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class TransferState(TypedDict):
    amount: float
    to_account: str
    approved: bool
    result: str

def prepare_transfer(state: TransferState) -> dict:
    return {"result": f"准备转账 ¥{state['amount']} 到 {state['to_account']}"}

def execute_transfer(state: TransferState) -> dict:
    if state.get("approved"):
        return {"result": f"✅ 已成功转账 ¥{state['amount']} 到 {state['to_account']}"}
    return {"result": "❌ 转账已被拒绝"}

graph = StateGraph(TransferState)
graph.add_node("prepare", prepare_transfer)
graph.add_node("execute", execute_transfer)
graph.add_edge(START, "prepare")
graph.add_edge("prepare", "execute")
graph.add_edge("execute", END)

# interrupt_before=["execute"] 会在进入 execute 节点前暂停
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer, interrupt_before=["execute"])

config = {"configurable": {"thread_id": "transfer-001"}}

# 第一次运行 → 在 execute 前暂停
result = app.invoke(
    {"amount": 50000, "to_account": "622848***1234", "approved": False},
    config=config,
)
print("⏸️ 等待审批...", result["result"])

# 人工审批通过 → 更新状态
app.update_state(config, {"approved": True})

# 恢复执行
result = app.invoke(None, config=config)
print(result["result"])  # ✅ 已成功转账 ¥50000 到 622848***1234

主 Agent（Supervisor）将任务分配给专门的子 Agent，各 Agent 独立执行后汇总。

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

# 子 Agent 1：研究员
def web_search(query: str) -> str:
    """搜索网络信息"""
    return f"搜索结果：关于'{query}'的最新数据和报告..."

researcher = create_react_agent(
    model=llm,
    tools=[web_search],
    prompt="你是研究员，负责搜索和收集信息。只返回事实，不做主观分析。",
)

# 子 Agent 2：分析师
def create_chart(data: str) -> str:
    """生成数据图表"""
    return f"已根据数据生成可视化图表"

analyst = create_react_agent(
    model=llm,
    tools=[create_chart],
    prompt="你是数据分析师，负责分析研究员提供的数据并生成报告。",
)

# 主 Agent：协调者——把子 Agent 封装为工具
def call_researcher(task: str) -> str:
    """委托研究员收集信息"""
    result = researcher.invoke({"messages": [{"role": "user", "content": task}]})
    return result["messages"][-1].content

def call_analyst(task: str) -> str:
    """委托分析师分析数据"""
    result = analyst.invoke({"messages": [{"role": "user", "content": task}]})
    return result["messages"][-1].content

supervisor = create_react_agent(
    model=llm,
    tools=[call_researcher, call_analyst],
    prompt="你是项目主管。分解用户需求，协调研究员和分析师完成任务。",
)

result = supervisor.invoke({
    "messages": [{"role": "user", "content": "调研 2026 年新能源汽车市场并出分析报告"}]
})
print(result["messages"][-1].content)

使用 Checkpointer 让对话状态跨请求持久化，Agent 记住对话上下文。

from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI

def lookup_order(order_id: str) -> str:
    """查询订单状态"""
    orders = {
        "ORD001": "已发货，预计明天到达",
        "ORD002": "待支付",
        "ORD003": "已签收",
    }
    return orders.get(order_id, "订单不存在")

checkpointer = MemorySaver()

agent = create_react_agent(
    model=ChatOpenAI(model="gpt-4o"),
    tools=[lookup_order],
    prompt="你是电商客服助手。记住用户的上下文，提供连贯的对话服务。",
    checkpointer=checkpointer,
)

config = {"configurable": {"thread_id": "user-session-42"}}

# 第一轮
r1 = agent.invoke(
    {"messages": [{"role": "user", "content": "帮我查一下 ORD001"}]},
    config=config,
)
print("客服:", r1["messages"][-1].content)

# 第二轮 — Agent 记住了上下文
r2 = agent.invoke(
    {"messages": [{"role": "user", "content": "那 ORD002 呢？"}]},
    config=config,
)
print("客服:", r2["messages"][-1].content)

# 第三轮 — 跨轮次记忆
r3 = agent.invoke(
    {"messages": [{"role": "user", "content": "刚才那两个订单哪个先到？"}]},
    config=config,
)
print("客服:", r3["messages"][-1].content)