LangGraph 基础入门
模块总结

# ── 依赖导入 ───────────────────────────────────────────
from typing import Annotated
from typing_extensions import TypedDict
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import MemorySaver

# ════════════════════════════════════════════════════
# 【L3】工具定义：@tool 装饰器自动生成 JSON Schema
#   LLM 通过 Schema 了解每个工具的参数和用途
#   ToolNode 根据 LLM 输出的 tool_call 自动匹配并执行
# ════════════════════════════════════════════════════
@tool
def get_weather(city: str) -> str:
    """获取指定城市的当前天气信息。"""
    # 实际项目中调用真实天气 API，此处模拟返回
    weather_data = {
        "北京": "晴，26°C，微风",
        "上海": "多云，22°C，东南风3级",
        "成都": "阴，18°C，无风",
    }
    return weather_data.get(city, f"{city}：天气数据暂不可用")

@tool
def search_attractions(city: str, category: str = "all") -> str:
    """搜索城市的热门景点。category 可为 'nature'（自然）、'culture'（文化）或 'all'（全部）。"""
    attractions = {
        "北京": ["故宫", "长城", "颐和园", "天坛"],
        "上海": ["外滩", "豫园", "东方明珠", "田子坊"],
        "成都": ["大熊猫基地", "宽窄巷子", "武侯祠", "锦里"],
    }
    spots = attractions.get(city, [])
    return f"{city}热门景点：{', '.join(spots)}" if spots else f"暂无{city}景点数据"

@tool
def estimate_budget(city: str, days: int, style: str = "standard") -> str:
    """估算旅行预算。style 可为 'budget'（经济）、'standard'（标准）、'luxury'（豪华）。"""
    per_day = {"budget": 300, "standard": 600, "luxury": 1500}
    daily_cost = per_day.get(style, 600)
    total = daily_cost * days
    return f"{city} {days}天{style}旅行预计费用：¥{total}（每天约¥{daily_cost}）"

# 工具列表：注册所有可用工具
tools = [get_weather, search_attractions, estimate_budget]

# ════════════════════════════════════════════════════
# 【L1 + L2】State 定义
#   · 继承 MessagesState（L2）：内置 messages 字段 +
#     add_messages Reducer，开箱即用
#   · 扩展自定义字段（L1）：TypedDict 风格添加额外状态
# ════════════════════════════════════════════════════
class TravelState(MessagesState):
    # MessagesState 已内置：messages: Annotated[list[AnyMessage], add_messages]
    # 下面是扩展字段
    destination: str         # 用户确定的目的地（默认覆盖策略）
    travel_days: int          # 旅行天数

# ════════════════════════════════════════════════════
# 【L2】LLM 初始化 + bind_tools
#   · bind_tools 将工具的 JSON Schema 注入 LLM 的系统提示
#   · LLM 输出 tool_call 时，内容是 JSON 格式的参数
# ════════════════════════════════════════════════════
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_with_tools = llm.bind_tools(tools)  # 【L2】核心：让 LLM 知道有哪些工具

# ════════════════════════════════════════════════════
# 【L2 + L4】Agent 节点：ReAct 循环的推理端
#   · 读取完整消息历史，带入工具上下文，调用 LLM
#   · 返回 {"messages": [新的 AIMessage]}
#   · add_messages Reducer 自动追加，不覆盖历史
# ════════════════════════════════════════════════════
def agent(state: TravelState):
    system_prompt = SystemMessage(content=(
        "你是一个专业的旅行规划助手。你可以查询天气、搜索景点、估算预算。"
        "请根据用户需求，综合使用工具制定完整的旅行建议。"
    ))
    # state["messages"] 包含完整的对话历史（由 add_messages Reducer 维护）
    response = llm_with_tools.invoke([system_prompt] + state["messages"])
    return {"messages": [response]}  # 返回字典，Reducer 自动追加

# ════════════════════════════════════════════════════
# 【L1】StateGraph：图的构建器
# 【L3】ToolNode：自动执行工具调用的内置节点
# ════════════════════════════════════════════════════
tool_node = ToolNode(tools)  # 【L3】传入工具列表，自动匹配 tool_call 并执行

builder = StateGraph(TravelState)   # 【L1】创建图，传入 State Schema

# ── 注册节点 ────────────────────────────────────────
builder.add_node("agent", agent)       # 【L1】LLM 推理节点
builder.add_node("tools", tool_node)  # 【L3】工具执行节点

# ── 连接边 ──────────────────────────────────────────
builder.add_edge(START, "agent")       # 【L1】入口 → agent

# 【L3】条件边：LLM 有工具调用 → tools，否则 → END
builder.add_conditional_edges(
    "agent",
    tools_condition,   # 内置路由函数，检测 tool_calls 是否存在
)

# 【L4】关键！这条回边让图形成环，实现 ReAct 循环
# tools 执行完工具后，把结果（ToolMessage）追加进 messages，
# 然后重新路由到 agent，让 LLM 看到工具结果继续推理
builder.add_edge("tools", "agent")

# ── 【L5】挂载 MemorySaver，实现跨轮对话记忆 ──────
memory = MemorySaver()
graph = builder.compile(checkpointer=memory)  # 【L5】编译时注入 Checkpointer

# ════════════════════════════════════════════════════
# 【L5】thread_id：会话标识
#   · 相同 thread_id → 共享同一对话历史
#   · 不同 thread_id → 完全独立，互不干扰
# ════════════════════════════════════════════════════
config = {"configurable": {"thread_id": "user_alice_trip_001"}}

# ── 第 1 轮：询问北京旅行建议 ─────────────────────
result_1 = graph.invoke(
    {"messages": [HumanMessage(content="我想去北京旅行3天，帮我规划一下")]},
    config=config,
)
# Agent 会自动调用：get_weather("北京") → search_attractions("北京") →
# estimate_budget("北京", 3, "standard")，然后综合给出回答

# ── 第 2 轮：追问（自动记得第 1 轮内容）────────────
result_2 = graph.invoke(
    {"messages": [HumanMessage(content="如果改成豪华出行，费用大概是多少？")]},
    config=config,  # 同一 thread_id → 记得"北京3天"的上下文
)
# LLM 知道上下文是北京3天，直接调用 estimate_budget("北京", 3, "luxury")

# ── 多用户隔离示例 ──────────────────────────────────
alice_config = {"configurable": {"thread_id": "alice_001"}}
bob_config   = {"configurable": {"thread_id": "bob_001"}}
# alice 和 bob 的对话历史完全隔离，互不影响

# ── 查看完整消息历史 ────────────────────────────────
for msg in result_2["messages"]:
    role = msg.__class__.__name__.replace("Message", "")
    print(f"[{role}] {msg.content[:80]}")

# ── langgraph.json（项目根目录）──────────────────────
# 【L6】LangGraph Studio / Cloud 的部署配置文件
# · "graphs"：声明图的名称 → 对应的 Python 模块路径
# · "dependencies"：项目依赖，打包时自动安装
# · "env"：环境变量文件路径

# langgraph.json
{
  "graphs": {
    "travel_agent": "./travel_agent.py:graph"
    //                    ↑ 文件路径        ↑ 变量名
  },
  "dependencies": ["."],
  "env": ".env"
}

# ── 本地启动 LangGraph Studio ────────────────────────
# 【L6】在项目根目录运行，会自动打开 Studio 界面
# $ langgraph dev
#   → 启动本地 API 服务（默认 http://127.0.0.1:2024）
#   → 自动打开 Studio 可视化界面
#   → 可在 Studio 里直接输入消息测试图的运行

# ── Studio 的调试能力 ────────────────────────────────
# 【L6】每次执行可以看到：
#   · 图结构可视化（节点 + 边，含 ReAct 循环）
#   · 每个节点执行前后的 State 快照对比
#   · 工具调用的输入参数和返回结果
#   · 每条消息的完整内容（AIMessage 含 tool_calls）
#   · 可以在任意步骤"时间旅行"（查看历史快照）

# ── 云端部署（可选）─────────────────────────────────
# $ langgraph build -t my-travel-agent  # 构建 Docker 镜像
# $ langgraph up                        # 启动生产服务