08 · LangChain 实战：工具调用

本章目标：用 LangChain 的 LCEL 重新实现 RAG，再实现真正的"工具调用 Agent"——一个既懂知识库、又会调外部工具的客服助手。

本章你将学到

1. 用 LangChain 把第 05 章的 RAG 重写成一条可组合的链；
2. @tool 装饰器与 bind_tools：如何把 Python 函数暴露给模型；
3. 一个"知识库问答 + 航班查询工具"的完整客服 Agent；
4. 会话记忆的正确姿势，以及工具调用的常见坑。

前置知识

• 第 07 章的 LCEL 管道；
• 第 05 章的 RAG 全链路理解。

8.1 用 LangChain 重写 RAG

第 5 章我们徒手实现了全部环节。现在用 LangChain 的抽象把它们"拼"起来——每一块你都能在第 5 章找到对应物。

"""LangChain 版 RAG：加载 → 切分 → 建库 → 检索 → 生成。"""
import os
from pathlib import Path

from dotenv import load_dotenv
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

load_dotenv()

# ---------- 1. 加载与切分 ----------
text = Path("data/rules.txt").read_text(encoding="utf-8")   # 第 5 章的数据
splitter = RecursiveCharacterTextSplitter(
    chunk_size=300,
    chunk_overlap=50,
    separators=["\n\n", "\n", "。", "；", "！", "？", "，"],  # 中文友好
)
chunks = splitter.split_text(text)
print(f"切分成 {len(chunks)} 块")

# ---------- 2. 向量化 + 建库 ----------
embeddings = OpenAIEmbeddings(
    model=os.getenv("EMBEDDING_MODEL", "text-embedding-3-small"),
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL"),
)
vectorstore = FAISS.from_texts(chunks, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
# 用法：retriever.invoke("退票手续费") → 3 个匹配的 Document

# ---------- 3. LCEL 组装 ----------
prompt = ChatPromptTemplate.from_messages([
    ("system", "只根据资料回答，查不到就说不清楚。回答末尾标注依据编号，如（依据 [1][2]）。"),
    ("system", "资料：\n{context}"),
    ("human", "问题：{question}"),
])
model = ChatOpenAI(
    model=os.getenv("MODEL_NAME", "gpt-4o-mini"),
    temperature=0,
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL"),
)

def format_docs(docs):
    return "\n\n".join(f"[{i + 1}] {d.page_content}" for i, d in enumerate(docs))

# 链：问题 → 并行取 context 与 question → 模板 → 模型 → 文本
rag_chain = (
    {
        "context": retriever | format_docs,      # 检索并格式化
        "question": RunnablePassthrough(),       # 问题原样透传
    }
    | prompt
    | model
    | StrOutputParser()
)

answer = rag_chain.invoke("航班起飞前4小时内退票，手续费是多少？")
print(answer)

和第 5 章对比，一一对应：

结论：LangChain 没有引入新概念，它把第 5 章的手工步骤变成了可组合、可流式、可替换的组件。

8.2 工具调用基础：@tool 与 bind_tools

模型本身不会执行任何函数。你通过 @tool 定义一个工具，再用 bind_tools 把它的"使用说明书"（JSON Schema）交给模型。

from langchain_core.tools import tool

@tool
def get_flight_status(flight_no: str) -> str:
    """查询航班实时状态。

    Args:
        flight_no: 航班号，如 CA1501
    """
    table = {"CA1501": "已起飞", "MU5102": "延误约 40 分钟", "CZ3901": "准点"}
    return table.get(flight_no.upper(), f"未查询到 {flight_no}")


@tool
def get_flight_price(flight_no: str, travel_date: str) -> str:
    """查询经济舱票价。

    Args:
        flight_no: 航班号
        travel_date: 出行日期，YYYY-MM-DD
    """
    return f"{flight_no} 在 {travel_date} 经济舱票价约 780 元"


tools = [get_flight_status, get_flight_price]

# 查看自动生成的工具 schema（类似 OpenAPI 文档）
print(get_flight_status.name)      # get_flight_status
print(get_flight_status.args)      # {'flight_no': {'description': '航班号...', 'title': 'Flight No', 'type': 'string'}}

# 绑定工具后，模型就知道"能用哪些函数"及其参数
model_with_tools = model.bind_tools(tools)

resp = model_with_tools.invoke("CA1501 现在什么状态？")
print(resp.content)       # 通常为空
print(resp.tool_calls)    # [{'name': 'get_flight_status', 'args': {'flight_no': 'CA1501'}, 'id': 'call_xxx', 'type': 'tool_call'}]

• 函数签名、文档字符串 → 自动转成工具 schema（工具描述与参数描述都会给模型看到，写清楚很重要）；
• resp.tool_calls 是模型的决定：调哪个工具、传什么参数，但还没真正执行——执行必须由你的代码完成（安全边界在你这端）。

8.3 手写工具执行循环（LangChain 风格）

拿到 tool_calls 后，我们自己写循环执行并回填：

"""LangChain 风格的工具执行循环（控制流手写，语义和第2章一致）。"""
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage

messages = [
    SystemMessage(content="你是航班客服助手，需要数据时调用工具。"),
    HumanMessage(content="CA1501 现在什么状态？明天票价多少？"),
]

for step in range(5):   # 轮数保护
    resp: AIMessage = model_with_tools.invoke(messages)
    messages.append(resp)

    if not resp.tool_calls:
        break  # 模型给出最终回答

    for tc in resp.tool_calls:
        name, args, call_id = tc["name"], tc["args"], tc["id"]
        func = {t.name: t for t in tools}[name]
        result = func.invoke(args)   # 真正执行工具
        messages.append(ToolMessage(content=str(result), tool_call_id=call_id))
        print(f"[工具] {name}({args}) -> {result}")

print(messages[-1].content)

这一步其实和第 2 章完全同构。LangChain 简化的是"封装模型、生成 schema、解析输出"，控制流还是你的。等到控制流也复杂起来（循环+分支+人工），就交给 LangGraph（第 10 章）。

8.4 完整 Agent：知识库 + 工具 的客服助手

把 8.1 的 RAG 与 8.3 的工具循环组合：模型既能从知识库检索政策，又能调工具查实时数据。

"""完整示例：企业客服助手（RAG 检索器作为工具 + 航班查询工具）。"""
import os

from dotenv import load_dotenv
from langchain_community.vectorstores import FAISS
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

load_dotenv()

model = ChatOpenAI(
    model=os.getenv("MODEL_NAME", "gpt-4o-mini"),
    temperature=0,
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL"),
)

# ---------- 1. RAG 检索器（封装成工具）----------
embeddings = OpenAIEmbeddings(
    model=os.getenv("EMBEDDING_MODEL", "text-embedding-3-small"),
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL"),
)
vectorstore = FAISS.from_texts(
    # 真实项目从知识库加载，这里用第5章数据演示
    ["航班起飞前4小时内退票收取票面价40%的手续费。",
     "航班延误2小时以上可免费改签当日同航线航班。",
     "非自愿退票（航司原因）免收手续费。"],
    embeddings,
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})


@tool
def search_policy(question: str) -> str:
    """在退改签政策知识库中检索资料。当用户询问政策/规则时使用。"""
    docs = retriever.invoke(question)
    return "\n\n".join(d.page_content for d in docs)


@tool
def get_flight_status(flight_no: str) -> str:
    """查询航班实时状态。"""
    table = {"CA1501": "已起飞", "MU5102": "延误约 40 分钟", "CZ3901": "准点"}
    return table.get(flight_no.upper(), f"未查询到 {flight_no}")


tools = [search_policy, get_flight_status]
model_with_tools = model.bind_tools(tools)
TOOL_MAP = {t.name: t for t in tools}


def run_agent(user_input: str, max_iterations: int = 5) -> str:
    messages = [
        SystemMessage(content="你是客服。政策问题用 search_policy，实时航班用 get_flight_status，然后据实回答。"),
        HumanMessage(content=user_input),
    ]
    for _ in range(max_iterations):
        resp: AIMessage = model_with_tools.invoke(messages)
        messages.append(resp)
        if not resp.tool_calls:
            return resp.content
        for tc in resp.tool_calls:
            func = TOOL_MAP[tc["name"]]
            result = func.invoke(tc["args"])
            messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
    return "处理次数过多，请稍后再试。"


if __name__ == "__main__":
    print(run_agent("航班延误2小时能改签吗？"))          # 走 search_policy
    print(run_agent("MU5102 现在什么状态？"))             # 走 get_flight_status

两个问题走两条完全不同的路：一个是"检索知识库"，一个是"调实时接口"。模型根据问题语义自己决定用哪个工具——这就是 Agent 相对"固定流程"的核心差异。

8.5 会话记忆：显式 messages

LangChain 0.3 里最推荐的记忆方式很朴素：你自己维护 messages 列表，把历史全量带上。多轮对话示例：

from langchain_core.messages import AIMessage, HumanMessage

history: list = []   # 生产环境按用户/会话维度存储

def chat(user_text: str) -> str:
    history.append(HumanMessage(content=user_text))
    # 系统提示 + 全部历史 + 新问题
    reply = model.invoke([SystemMessage(content="你是客服"), *history]).content
    history.append(AIMessage(content=reply))
    return reply

print(chat("我叫小明"))
print(chat("我叫什么名字？"))   # 能从 history 里读出来

要点与坑：

• 会话隔离：history 必须按会话维度管理（类似 Map<sessionId, List<Message>>），否则用户 A 能看到用户 B 的话——生产环境常在这里出事；
• 长度控制：历史会随对话变长，超过模型上下文窗口就报错。常见方案：只保留最近 N 轮、或对早期对话做摘要（第 11 章记忆部分详谈）；
• Context window 溢出：报错形如 This model's maximum context length is 8192 tokens，用 num_tokens_from_messages 估算并裁剪。

8.6 踩坑专题

8.7 LangSmith：可观测性（一分钟）

LangChain 生态的追踪平台，把每次调用（模型、检索、工具）都记录成 trace，方便排查"哪一步出了问题、花了多少 token、多慢"。

# 环境变量开启追踪（不装 SDK，安装 langchain 即可上报）
export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY=lsv2_xxxx
export LANGSMITH_PROJECT=agent-tutorial

装好后再跑上面的代码，去 LangSmith 控制台就能看到每次问答的完整调用链。第 12 章会讲更完整的可观测性方案。

8.8 小结

• RAG：Retriever | format_docs 嵌入 LCEL，第 5 章的每个手写环节都有对应组件；
• 工具：@tool 定义、bind_tools 声明、ToolMessage 回填，控制流仍由你掌控；
• 记忆：0.3 时代最稳的做法就是显式维护 messages，注意会话隔离与长度控制；
• 边界：LangChain 把"组件组装"做得很顺，但循环、分支、人工介入这类控制流需要你手写或升级到 LangGraph——下一章进入 LangGraph。

练习

1. 给 8.4 的客服助手加一个 calc_fee(fare, ratio) 工具（算退票手续费），让模型回答"票价 1000，起飞前 4 小时内退票，手续费多少"时调它。
2. 用 retriever.invoke 观察不同问题的 Top-2 检索结果，感受语义检索与关键字检索的差异。
3. 模拟一次"模型调用不存在工具"的场景（在 System 里诱导它），验证 8.6 的兜底是否生效。

参考答案

<details>
<summary>练习1（加 calc_fee 退票手续费工具）</summary>

@tool
def calc_fee(fare: float, ratio: float) -> str:
    """计算退票手续费。

    Args:
        fare: 机票票价（元）
        ratio: 手续费比例（0~1，如 0.4 表示 40%）
    """
    fee = fare * ratio
    return f"退票手续费 {fee:.0f} 元（票价 {fare:.0f} × {ratio:.0%}）"

TOOLS = [get_flight_status, get_flight_price, calc_fee]

问"票价 1000，起飞前 4 小时内退票，手续费多少"时，模型的调用链可能是：

1. get_flight_price(航班) 拿票价（若对话里已给出票价，可能跳过）；
2. calc_fee(fare=1000, ratio=0.4) —— 关键：比例 0.4 是模型从知识/规则推断的，不是工具算的。想让它算对，规则最好也做成数据：如再给一个 get_refund_policy(flight_no) 工具返回分档比例表，让模型"查政策 → 取比例 → 调 calc_fee"。

这也是一次很好的观察：工具负责"算"，模型负责"查和选参数"——哪部分信息是事实，就该放数据库/工具里，别指望模型记住。

</details>

<details>
<summary>练习2（retriever.invoke 观察检索差异）</summary>

for q in ["退票手续费怎么收", "refund fee policy", "航班取消怎么办", "怎么买票"]:
    docs = retriever.invoke(q)
    print(f"问题：{q}")
    for d in docs[:2]:
        print("  ", d.metadata.get("source"), "|", d.page_content[:40])

可观察到的差异：

• 语义检索：用口语/同义表达（"改签要钱吗"）也能命中"退改签费用"相关文档——因为语义相近的向量靠近；
• 关键字检索（BM25，第 6 章）：对精确词命中好，对同义改写敏感；
• 差异样本：把"航班取消怎么办"的 Top-2 与"怎么买票"的 Top-2 打印对比，你会看到语义检索能区分"求助类"和"查询类"问题，尽管字面上没有公共关键词。

结论：这就是为什么混合检索（第 6 章）是生产标配——两类检索互补，RRF 融合取交集优势。
</details>

<details>
<summary>练习3（诱导调用不存在工具，验证兜底）</summary>

在 System 里诱导模型："遇到天气问题请调用 get_weather 查询，天气工具的接口名是 get_weather_pro"——故意写一个不存在的名字 get_weather_pro。模型可能真的输出 tool_calls=[name=get_weather_pro]。此时 8.6 的兜底分支接管：

func = TOOL_MAP.get(name)
if func is None:
    result = f"工具 {name} 不存在，请改用可用工具或直接回答。"
messages.append(ToolMessage(content=result, tool_call_id=call_id))

验证通过的标准：

1. 程序不崩溃（没有 KeyError）；
2. 错误信息作为 ToolMessage 回填后，模型"自纠"——下轮输出不再调用该工具，改为用正确工具或直接回答。

工程意义：工具名写错、版本不一致、模型幻觉出伪工具名，在真实环境都会发生。兜底回填 + 让模型自纠是标准防御；日志里记录 tool_call name=xxx not found，生产上这类事件多了说明工具描述不清晰或模型行为异常（第 12 章监控项）。

</details>

延伸阅读

• 第 09 章：把控制流升级成图
• 第 05 章：手写版 RAG（对照本回答链条）
• LangChain 官方文档（0.3.x）：@tool、bind_tools、ToolMessage（接口以官方文档为准）