LangChain / LangGraph 零基础手把手

每一步结构固定：
🎯 本步目标  →  📚 前置  →  🧠 概念（白话）
  →  💻 完整代码  →  ✅ 验证  →  ⚠️ 常见报错
  →  🧪 5 分钟小练习  →  📍 检查点

mkdir langchain-practice && cd langchain-practice
python -m venv .venv
source .venv/bin/activate          # Windows: .venv\Scripts\activate
pip install --upgrade pip
pip install \
  "langchain>=0.3" \
  "langchain-openai>=0.2" \
  "langchain-community>=0.3" \
  "langgraph>=0.2" \
  "python-dotenv"

OPENAI_API_KEY=sk-xxxxxxxxxxxxxx
# 如果用 DeepSeek：
# OPENAI_API_KEY=sk-xxxxxxxxxxxxxx
# OPENAI_BASE_URL=https://api.deepseek.com/v1

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
 
load_dotenv()  # 从 .env 读 OPENAI_API_KEY
 
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# 如果用 DeepSeek：model="deepseek-chat"
 
resp = llm.invoke("用一句话介绍 LangChain")
print(resp.content)

python step01_hello.py

LangChain 是一个用于构建基于大语言模型的应用程序的框架...

for chunk in llm.stream("讲个冷笑话"):
    print(chunk.content, end="", flush=True)

prompt  |  llm  |  parser
↓
(字典) → 提示词 → AI 消息 → 字符串

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
load_dotenv()
 
# 1. Prompt：把字典变成一条消息
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是简洁的助手，只用一句话回答。"),
    ("user", "{question}"),
])
 
# 2. LLM：把消息变成 AI 消息
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
 
# 3. Parser：把 AI 消息变成字符串
parser = StrOutputParser()
 
# 4. 串起来
chain = prompt | llm | parser
 
# 5. 用字典喂入
print(chain.invoke({"question": "什么是 RAG？"}))
print("-" * 40)
print(chain.invoke({"question": "Python 的装饰器是什么？"}))

from langchain_core.output_parsers import JsonOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "你把问题分类。必须严格返回 JSON：{{\"title\":\"...\",\"tags\":[\"...\"]}}"),
    ("user", "{question}"),
])
chain = prompt | llm | JsonOutputParser()
print(chain.invoke({"question": "怎么学好 Python？"}))

from dotenv import load_dotenv
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
 
load_dotenv()
 
class IssueTicket(BaseModel):
    """把用户抱怨抽成工单。"""
    title: str = Field(description="问题标题，不超过 20 字")
    severity: str = Field(description="low / medium / high")
    tags: list[str] = Field(description="相关标签，2-5 个")
    summary: str = Field(description="一句话摘要")
 
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_llm = llm.with_structured_output(IssueTicket)
 
complaint = "我买的耳机才用了一个月左声道就没声音了，花了 800 块，你们客服还不回我消息！"
ticket = structured_llm.invoke(complaint)
 
print(type(ticket))
print(ticket.title)
print(ticket.severity)
print(ticket.tags)
print(ticket.summary)

第 1 轮：[user: "我叫张三"]  →  AI: "你好张三"
第 2 轮：[user: "我叫张三", AI: "你好张三", user: "我叫什么？"]  →  AI: "你叫张三"
                ↑ 全部历史重塞

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
 
load_dotenv()
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是友好的中文助手。"),
    MessagesPlaceholder("history"),
    ("user", "{input}"),
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm
 
# 每个 session_id 对应一份独立的历史
store: dict[str, InMemoryChatMessageHistory] = {}
 
def get_history(session_id: str):
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]
 
with_memory = RunnableWithMessageHistory(
    chain,
    get_history,
    input_messages_key="input",
    history_messages_key="history",
)
 
cfg = {"configurable": {"session_id": "u-001"}}
 
print(with_memory.invoke({"input": "我叫张三，28 岁。"}, cfg).content)
print(with_memory.invoke({"input": "我多大？"}, cfg).content)
print(with_memory.invoke({"input": "我叫什么？"}, cfg).content)
 
# 换一个 session，应该"不认识"你
cfg2 = {"configurable": {"session_id": "u-002"}}
print(with_memory.invoke({"input": "我叫什么？"}, cfg2).content)

pip install "langchain-chroma>=0.1" "langchain-text-splitters>=0.3" pypdf

问题 ──embed──► 向量 ──检索──► Top-K 相似文档块 ──拼进 prompt──► LLM 回答

# 公司请假制度
 
- 普通病假：提前 1 小时通知直属上级，填写 OA 单据。
- 年假：员工入职满 1 年后每年 10 天。
- 婚假：法定 3 天 + 公司额外 5 天，共 8 天。
- 产假：按国家标准 158 天。
- 调休：加班满 8 小时可兑换 1 天调休，3 个月内有效。

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_chroma import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
 
load_dotenv()
 
# 1. 加载 & 切分
loader = TextLoader("kb.md", encoding="utf-8")
docs = loader.load()
splitter = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=20)
chunks = splitter.split_documents(docs)
print(f"切成了 {len(chunks)} 块")
 
# 2. 存进向量库
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(chunks, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
 
# 3. 组 RAG chain
def format_docs(docs):
    return "\n\n".join(d.page_content for d in docs)
 
prompt = ChatPromptTemplate.from_template("""
只能基于下面【参考资料】回答问题。如果资料里没有就说"资料中未提到"。
 
【参考资料】
{context}
 
【问题】
{question}
""")
 
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
 
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)
 
# 4. 测试
for q in ["婚假有几天？", "产假多少天？", "午休时间是多久？"]:
    print(f"\nQ: {q}")
    print(f"A: {rag_chain.invoke(q)}")

Q: 婚假有几天？
A: 婚假共 8 天（法定 3 天 + 公司额外 5 天）。
 
Q: 产假多少天？
A: 产假 158 天。
 
Q: 午休时间是多久？
A: 资料中未提到。

from langchain_core.runnables import RunnableParallel
 
rag_with_source = RunnableParallel(
    answer=rag_chain,
    source=retriever,
)
result = rag_with_source.invoke("产假多少天？")
print(result["answer"])
print("--- 依据 ---")
for d in result["source"]:
    print(d.page_content[:60])

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage
 
load_dotenv()
 
@tool
def add(a: float, b: float) -> float:
    """两数相加。"""
    return a + b
 
@tool
def multiply(a: float, b: float) -> float:
    """两数相乘。"""
    return a * b
 
@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气。返回一句话描述。"""
    # 真实项目这里调外部 API；这里假造
    return f"{city} 今天多云，22 度。"
 
tools = [add, multiply, get_weather]
tool_map = {t.name: t for t in tools}
 
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_with_tools = llm.bind_tools(tools)
 
def run(user_input: str):
    messages = [HumanMessage(user_input)]
    # 第一步：问 LLM
    ai_msg = llm_with_tools.invoke(messages)
    messages.append(ai_msg)
 
    # 第二步：如果 LLM 想调工具，就执行
    while ai_msg.tool_calls:
        for call in ai_msg.tool_calls:
            tool_fn = tool_map[call["name"]]
            result = tool_fn.invoke(call["args"])
            print(f"  [调用 {call['name']}({call['args']}) → {result}]")
            messages.append({
                "role": "tool",
                "tool_call_id": call["id"],
                "content": str(result),
            })
        # 第三步：把结果再塞给 LLM，让它基于结果回答
        ai_msg = llm_with_tools.invoke(messages)
        messages.append(ai_msg)
 
    return ai_msg.content
 
for q in [
    "3.14 乘以 2.5 等于多少？",
    "上海今天天气怎么样？",
    "先算 100 加 200，再乘以 3，告诉我结果。",
]:
    print(f"\nQ: {q}")
    print(f"A: {run(q)}")

Q: 3.14 乘以 2.5 等于多少？
  [调用 multiply({'a': 3.14, 'b': 2.5}) → 7.85]
A: 3.14 乘以 2.5 等于 7.85。
 
Q: 上海今天天气怎么样？
  [调用 get_weather({'city': '上海'}) → 上海 今天多云，22 度。]
A: 上海今天多云，22 度。
 
Q: 先算 100 加 200，再乘以 3，告诉我结果。
  [调用 add({'a': 100, 'b': 200}) → 300]
  [调用 multiply({'a': 300, 'b': 3}) → 900]
A: 结果是 900。

@tool
def transfer_money(from_account: str, to_account: str, amount: float) -> str:
    """转账。注意 amount 必须大于 0。"""
    if amount <= 0:
        return "错误：金额必须大于 0"
    if amount > 10000:
        return "错误：单笔限额 10000"
    return f"已从 {from_account} 向 {to_account} 转 {amount} 元"

State（状态）：一个字典，整张图共享
  ↓
Node（节点）：一个函数，读 State，返回对 State 的修改
  ↓
Edge（边）：规定节点之间怎么跳转，可以是直连也可以是条件

START  ──►  [load_user]  ──►  [answer]  ──►  END
              ↓ 读 state["user_id"]
              ↓ 写 state["profile"]
                         ↓ 读 state["profile"] + state["question"]
                         ↓ 写 state["answer"]

from typing import TypedDict, Annotated
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
import operator
 
load_dotenv()
 
# 1. 定义 State
class ChatState(TypedDict):
    question: str
    classification: str
    answer: str
    log: Annotated[list[str], operator.add]   # 每个节点追加，自动合并
 
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
 
# 2. 定义节点（每个节点是函数）
def classify(state: ChatState) -> dict:
    prompt = f"把下面问题归类为 'tech' / 'life' / 'other' 之一，只输出类别词：\n{state['question']}"
    cat = llm.invoke(prompt).content.strip().lower()
    if cat not in {"tech", "life", "other"}:
        cat = "other"
    return {"classification": cat, "log": [f"classify -> {cat}"]}
 
def answer(state: ChatState) -> dict:
    style = {
        "tech":  "用工程师的口吻回答",
        "life":  "用亲切的口吻回答",
        "other": "用简洁的口吻回答",
    }[state["classification"]]
    prompt = f"{style}：{state['question']}"
    ans = llm.invoke(prompt).content
    return {"answer": ans, "log": ["answer done"]}
 
def done(state: ChatState) -> dict:
    return {"log": ["done"]}
 
# 3. 组装图
graph = StateGraph(ChatState)
graph.add_node("classify", classify)
graph.add_node("answer", answer)
graph.add_node("done", done)
 
graph.add_edge(START, "classify")
graph.add_edge("classify", "answer")
graph.add_edge("answer", "done")
graph.add_edge("done", END)
 
app = graph.compile()
 
# 4. 跑
for q in ["Python 的 GIL 是什么？", "今天午饭吃什么好？"]:
    result = app.invoke({"question": q, "classification": "", "answer": "", "log": []})
    print(f"\nQ: {q}")
    print(f"  分类: {result['classification']}")
    print(f"  回答: {result['answer'][:60]}...")
    print(f"  日志: {result['log']}")

Q: Python 的 GIL 是什么？
  分类: tech
  回答: GIL（Global Interpreter Lock）是 CPython...
  日志: ['classify -> tech', 'answer done', 'done']
 
Q: 今天午饭吃什么好？
  分类: life
  回答: 今天天气不错，推荐试试...
  日志: ['classify -> life', 'answer done', 'done']

from typing import TypedDict
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel, Field
 
load_dotenv()
 
class State(TypedDict):
    message: str
    sentiment: str            # positive / neutral / negative
    answer: str
    handoff_to_human: bool
 
class Sentiment(BaseModel):
    label: str = Field(description="positive / neutral / negative")
 
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
sentiment_llm = llm.with_structured_output(Sentiment)
 
def detect_sentiment(state: State) -> dict:
    out = sentiment_llm.invoke(f"判断情绪，只返回 positive/neutral/negative：{state['message']}")
    label = out.label.strip().lower()
    if label not in {"positive", "neutral", "negative"}:
        label = "neutral"
    return {"sentiment": label}
 
def ai_answer(state: State) -> dict:
    ans = llm.invoke(f"友好回答：{state['message']}").content
    return {"answer": ans, "handoff_to_human": False}
 
def human_handoff(state: State) -> dict:
    return {
        "answer": "非常抱歉让您不满意，我已为您转接人工客服，请稍候。",
        "handoff_to_human": True,
    }
 
# 条件边的"路由函数"：返回下一个节点名
def route_by_sentiment(state: State) -> str:
    return "handoff" if state["sentiment"] == "negative" else "ai_answer"
 
graph = StateGraph(State)
graph.add_node("detect", detect_sentiment)
graph.add_node("ai_answer", ai_answer)
graph.add_node("handoff", human_handoff)
 
graph.add_edge(START, "detect")
graph.add_conditional_edges(
    "detect",
    route_by_sentiment,
    {"ai_answer": "ai_answer", "handoff": "handoff"},
)
graph.add_edge("ai_answer", END)
graph.add_edge("handoff", END)
 
app = graph.compile()
 
for msg in [
    "你们的产品真的不错！",
    "这什么垃圾系统，用了三次崩三次，退钱！",
    "请问怎么导出数据？",
]:
    r = app.invoke({"message": msg, "sentiment": "", "answer": "", "handoff_to_human": False})
    print(f"\n{msg}")
    print(f"  情绪: {r['sentiment']}")
    print(f"  回答: {r['answer'][:60]}")
    print(f"  转人工: {r['handoff_to_human']}")

from typing import TypedDict, Annotated
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import InMemorySaver
 
load_dotenv()
 
class State(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]   # 官方 reducer，自动去重合并
 
llm = ChatOpenAI(model="gpt-4o-mini")
 
def chat(state: State) -> dict:
    resp = llm.invoke(state["messages"])
    return {"messages": [resp]}
 
graph = StateGraph(State)
graph.add_node("chat", chat)
graph.add_edge(START, "chat")
graph.add_edge("chat", END)
 
# 关键：compile 时传入 checkpointer
checkpointer = InMemorySaver()
app = graph.compile(checkpointer=checkpointer)
 
# 第一次对话
cfg = {"configurable": {"thread_id": "u-001"}}
r1 = app.invoke({"messages": [HumanMessage("我叫王小明，学 Python 3 个月了。")]}, cfg)
print("AI:", r1["messages"][-1].content)
 
# 再问（同一 thread）
r2 = app.invoke({"messages": [HumanMessage("我叫什么？学了多久？")]}, cfg)
print("AI:", r2["messages"][-1].content)
 
# 看看完整 State
snapshot = app.get_state(cfg)
print(f"\n[thread u-001 共有 {len(snapshot.values['messages'])} 条消息]")
 
# 换个 thread，不认识
cfg2 = {"configurable": {"thread_id": "u-002"}}
r3 = app.invoke({"messages": [HumanMessage("我叫什么？")]}, cfg2)
print("\n不同 thread AI:", r3["messages"][-1].content)

from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3
 
conn = sqlite3.connect("chat.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)
# 剩下一模一样

app.invoke({...})  →  跑到 interrupt  →  返回，thread 状态 = 等待中
                                  ↓ 人类审批
app.invoke(Command(resume={"approved": True}))  →  从断点继续

from typing import TypedDict
from dotenv import load_dotenv
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command
 
load_dotenv()
 
class State(TypedDict):
    order_id: str
    amount: float
    reason: str
    decision: str
    status: str
 
def propose_refund(state: State) -> dict:
    # AI 决定发起退款（这里简化为直接来到这步）
    print(f"  [AI 判断：订单 {state['order_id']} 需退款 {state['amount']} 元，理由：{state['reason']}]")
    return {}
 
def human_approval(state: State) -> dict:
    # 在这里暂停
    decision = interrupt({
        "action": "refund",
        "order_id": state["order_id"],
        "amount": state["amount"],
        "reason": state["reason"],
        "prompt": "请审批：是否同意退款？",
    })
    # decision 是人类 resume 进来的值
    return {"decision": decision.get("verdict", "reject"),
            "status": "approved" if decision.get("verdict") == "approve" else "rejected"}
 
def execute(state: State) -> dict:
    if state["status"] == "approved":
        print(f"  [执行退款：{state['amount']} 元已退]")
    else:
        print(f"  [退款已拒绝]")
    return {}
 
graph = StateGraph(State)
graph.add_node("propose", propose_refund)
graph.add_node("approval", human_approval)
graph.add_node("execute", execute)
graph.add_edge(START, "propose")
graph.add_edge("propose", "approval")
graph.add_edge("approval", "execute")
graph.add_edge("execute", END)
 
checkpointer = InMemorySaver()
app = graph.compile(checkpointer=checkpointer)
 
cfg = {"configurable": {"thread_id": "order-123"}}
 
# --- 第一轮：跑到 interrupt 停住 ---
print("== 第一轮：AI 发起 ==")
state = app.invoke(
    {"order_id": "ORD-001", "amount": 580.0, "reason": "商品破损", "decision": "", "status": ""},
    cfg,
)
print(f"当前节点状态：{app.get_state(cfg).next}")   # 会显示 ('approval',)
print(f"interrupt payload: {app.get_state(cfg).tasks[0].interrupts[0].value}")
 
# --- 第二轮：人类审批 ---
print("\n== 人类审批中... 模拟同意 ==")
state = app.invoke(Command(resume={"verdict": "approve"}), cfg)
print(f"最终 status: {state['status']}")
 
# --- 另一个订单：模拟拒绝 ---
print("\n== 新订单，拒绝路径 ==")
cfg2 = {"configurable": {"thread_id": "order-124"}}
app.invoke(
    {"order_id": "ORD-002", "amount": 9999.0, "reason": "不喜欢", "decision": "", "status": ""},
    cfg2,
)
app.invoke(Command(resume={"verdict": "reject"}), cfg2)

== 第一轮：AI 发起 ==
  [AI 判断：订单 ORD-001 需退款 580.0 元，理由：商品破损]
当前节点状态：('approval',)
interrupt payload: {'action': 'refund', ...}
 
== 人类审批中... 模拟同意 ==
  [执行退款：580.0 元已退]
最终 status: approved
 
== 新订单，拒绝路径 ==
  [AI 判断：订单 ORD-002 需退款 9999.0 元，理由：不喜欢]
  [退款已拒绝]

START
  │
  ▼
[intent_router]   ◄── 把用户消息分类：order_query / policy_ask / refund / complaint
  │
  ├── order_query ──► [call_tools(get_order)] ──► [reply] ──► END
  │
  ├── policy_ask ───► [rag_search] ──► [reply] ──► END
  │
  ├── refund ───────► [propose_refund] ──► [approval(interrupt)] ──► [execute_refund] ──► [reply] ──► END
  │
  └── complaint ────► [handoff_human] ──► END
 
全程共享 State；全程走 SqliteSaver 持久化。

from typing import TypedDict, Annotated, Literal
import sqlite3
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate
from langchain_chroma import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.types import interrupt, Command
from pydantic import BaseModel, Field
 
load_dotenv()
 
# ---------- State ----------
class State(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    intent: str
    order_info: str
    kb_answer: str
    refund_amount: float
    refund_status: str
 
# ---------- 工具 ----------
FAKE_ORDERS = {
    "ORD-001": {"product": "无线耳机", "amount": 580, "status": "已发货"},
    "ORD-002": {"product": "蓝牙音箱", "amount": 399, "status": "已签收"},
}
 
@tool
def get_order(order_id: str) -> str:
    """根据订单号查订单详情。"""
    o = FAKE_ORDERS.get(order_id)
    if not o:
        return "订单不存在"
    return f"{order_id} - {o['product']}，{o['amount']} 元，{o['status']}"
 
# ---------- RAG ----------
with open("kb.md", "w", encoding="utf-8") as f:
    f.write("""
# 退换货政策
- 7 天无理由退货。
- 商品破损须提供照片。
- 退款 1-3 个工作日到账。
# 配送政策
- 江浙沪 48 小时达。
- 其它地区 3-5 天。
""")
loader = TextLoader("kb.md", encoding="utf-8")
chunks = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=20).split_documents(loader.load())
vs = Chroma.from_documents(chunks, OpenAIEmbeddings(model="text-embedding-3-small"))
retriever = vs.as_retriever(search_kwargs={"k": 3})
 
# ---------- LLM ----------
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
 
class Intent(BaseModel):
    label: Literal["order_query", "policy_ask", "refund", "complaint"] = Field(
        description="分类结果"
    )
intent_llm = llm.with_structured_output(Intent)
 
# ---------- 节点 ----------
def intent_router(state: State) -> dict:
    last = state["messages"][-1].content
    out = intent_llm.invoke(f"""把用户消息分类到以下之一并只输出字段：
- order_query: 问订单状态、物流（带订单号 ORD-）
- policy_ask: 问政策、规则（退货、配送、价保）
- refund: 明确要求退款退货
- complaint: 愤怒、投诉、辱骂
 
消息：{last}""")
    return {"intent": out.label}
 
def call_tool_node(state: State) -> dict:
    last = state["messages"][-1].content
    llm_with_tools = llm.bind_tools([get_order])
    ai = llm_with_tools.invoke([HumanMessage(last)])
    info = ""
    if ai.tool_calls:
        for c in ai.tool_calls:
            info = get_order.invoke(c["args"])
    return {"order_info": info}
 
def rag_node(state: State) -> dict:
    last = state["messages"][-1].content
    docs = retriever.invoke(last)
    context = "\n".join(d.page_content for d in docs)
    ans = llm.invoke(f"基于资料回答，资料外不要编造。\n资料：{context}\n问题：{last}").content
    return {"kb_answer": ans}
 
def propose_refund(state: State) -> dict:
    last = state["messages"][-1].content
    # 假设从消息中抽金额（简化，真实用结构化输出）
    return {"refund_amount": 580.0}
 
def approval_node(state: State) -> dict:
    verdict = interrupt({
        "action": "refund",
        "amount": state["refund_amount"],
        "prompt": "请审批退款。",
    })
    return {"refund_status": "approved" if verdict.get("ok") else "rejected"}
 
def execute_refund(state: State) -> dict:
    return {}
 
def handoff_node(state: State) -> dict:
    return {}
 
def reply_node(state: State) -> dict:
    # 根据 intent 组织最终回复
    if state["intent"] == "order_query":
        text = f"您的订单信息：{state['order_info']}"
    elif state["intent"] == "policy_ask":
        text = state["kb_answer"]
    elif state["intent"] == "refund":
        text = (f"退款 {state['refund_amount']} 元已批准，预计 1-3 个工作日到账。"
                if state["refund_status"] == "approved"
                else f"退款申请已拒绝。")
    else:
        text = "已为您转接人工客服，请稍候。"
    return {"messages": [AIMessage(text)]}
 
# ---------- 路由 ----------
def route(state: State) -> str:
    return {
        "order_query": "call_tool",
        "policy_ask": "rag",
        "refund": "propose_refund",
        "complaint": "handoff",
    }[state["intent"]]
 
# ---------- 组装 ----------
g = StateGraph(State)
g.add_node("intent", intent_router)
g.add_node("call_tool", call_tool_node)
g.add_node("rag", rag_node)
g.add_node("propose_refund", propose_refund)
g.add_node("approval", approval_node)
g.add_node("execute_refund", execute_refund)
g.add_node("handoff", handoff_node)
g.add_node("reply", reply_node)
 
g.add_edge(START, "intent")
g.add_conditional_edges("intent", route, {
    "call_tool": "call_tool",
    "rag": "rag",
    "propose_refund": "propose_refund",
    "handoff": "handoff",
})
g.add_edge("call_tool", "reply")
g.add_edge("rag", "reply")
g.add_edge("propose_refund", "approval")
g.add_edge("approval", "execute_refund")
g.add_edge("execute_refund", "reply")
g.add_edge("handoff", "reply")
g.add_edge("reply", END)
 
conn = sqlite3.connect("agent.db", check_same_thread=False)
app = g.compile(checkpointer=SqliteSaver(conn))
 
# ---------- 跑几个场景 ----------
def run(thread: str, msg: str, resume=None):
    cfg = {"configurable": {"thread_id": thread}}
    if resume is not None:
        out = app.invoke(Command(resume=resume), cfg)
    else:
        out = app.invoke({
            "messages": [HumanMessage(msg)],
            "intent": "", "order_info": "", "kb_answer": "",
            "refund_amount": 0.0, "refund_status": "",
        }, cfg)
    if out.get("messages"):
        print(f"  AI: {out['messages'][-1].content}")
    state = app.get_state(cfg)
    if state.next:
        print(f"  [暂停中，等待节点: {state.next}]")
    return state
 
print("\n=== 场景 1：查订单 ===")
run("t1", "帮我查一下 ORD-001 的订单状态")
 
print("\n=== 场景 2：问政策 ===")
run("t2", "你们多久能退款到账？")
 
print("\n=== 场景 3：投诉 ===")
run("t3", "你们这什么破玩意，赶紧给我退钱！")
 
print("\n=== 场景 4：退款（走人审）===")
run("t4", "我要退 580 元")
print("  [人类点击同意]")
run("t4", "", resume={"ok": True})

=== 场景 1：查订单 ===
  AI: 您的订单信息：ORD-001 - 无线耳机，580 元，已发货
 
=== 场景 2：问政策 ===
  AI: 退款 1-3 个工作日到账。
 
=== 场景 3：投诉 ===
  AI: 已为您转接人工客服，请稍候。
 
=== 场景 4：退款（走人审）===
  [暂停中，等待节点: ('approval',)]
  [人类点击同意]
  AI: 退款 580.0 元已批准，预计 1-3 个工作日到账。