目录

第十四章:项目实战 - 多工具 Agent 助手

如果第十三章的知识库问答系统侧重“基于资料回答”,那么本章的多工具 Agent 助手则侧重“围绕目标行动”。典型场景包括:查订单、查库存、查日程、写邮件草稿、汇总日报、触发审批等。

本章的目标是构建一个面向企业内部员工的多工具助手。它既可以回答问题,也可以调用工具完成事实核验和草稿生成;既能展示过程,又要控制风险边界。你会看到:一个真正能落地的 Agent,不是“给模型几十个工具让它自由发挥”,而是 工具设计 + 路由 + 审批 + 轨迹日志 + 明确边界 的组合。

14.1 项目目标

一个企业办公 Agent 助手,希望具备这些能力:

14.2 推荐系统架构

用户入口 -> 请求分类
  -> 知识问答工作流
  -> 多工具 Agent
  -> 审批工作流

这意味着:

14.3 先设计工具集,而不是先写 Prompt

第一版建议控制在 5~6 个工具以内。比如:

工具过多的坏处包括:

14.4 定义核心工具

14.4.1 查询订单

from langchain_core.tools import tool
 
@tool
def query_order(order_id: str) -> dict:
    """根据订单号查询订单状态、物流状态和预计到达时间。"""
    return {
        "order_id": order_id,
        "status": "已出库",
        "logistics_status": "运输中",
        "eta": "2026-04-04"
    }

14.4.2 查询库存

@tool
def query_inventory(sku: str) -> dict:
    """根据 SKU 查询商品可用库存和仓库位置。"""
    return {
        "sku": sku,
        "available": 23,
        "warehouse": "华东一号仓"
    }

14.4.3 检索知识库

@tool
def search_kb(question: str) -> str:
    """查询企业知识库,适用于制度、流程、报销、IT 支持等问题。"""
    # 实际项目中可调用第 13 章的 RAG 服务
    return "知识库命中:报销单需主管审批后再提交财务。"

14.4.4 生成邮件草稿

@tool
def draft_email(to: str, subject: str, body: str) -> dict:
    """生成待确认的邮件草稿,不会真正发送。"""
    return {
        "status": "draft",
        "to": to,
        "subject": subject,
        "body": body,
    }

14.4.5 创建审批单草稿

@tool
def create_approval_draft(title: str, reason: str) -> dict:
    """生成审批单草稿,不会直接提交。"""
    return {
        "status": "draft",
        "title": title,
        "reason": reason,
    }

14.5 构建 Agent

from langchain.agents import create_agent
 
agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[
        query_order,
        query_inventory,
        search_kb,
        draft_email,
        create_approval_draft,
    ],
    system_prompt="""
你是企业办公 Agent。
1. 涉及制度和流程的问题优先查询知识库。
2. 涉及订单、库存等事实信息必须调用工具验证。
3. 缺少关键参数时先追问,不要猜测。
4. 邮件和审批只能生成草稿,不能声称已经发送或提交。
5. 回答时要清楚说明你调用了哪些工具。
"""
)

14.6 一个完整的交互案例

14.6.1 问题:订单是否会延迟

result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "请帮我看订单 ORD20260403001 是否会延迟,如果会,请起草一份给客户的通知。"
    }]
})
 
print(result)

合理的 Agent 行为应是:

14.6.2 问题:报销流程怎么走

result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "差旅报销流程怎么走?需要主管审批吗?"
    }]
})

这类问题应优先走 `search_kb()`,而不是让 Agent 凭常识回答。

14.7 缺少参数时先追问

一个成熟 Agent 与普通“会调用工具的机器人”最大的区别之一,就是它知道什么时候该先追问。

例如:

result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "帮我查一下订单状态。"
    }]
})

合理输出应是:

而不是:

14.8 流式展示 Agent 过程

复杂 Agent 最好把过程展示出来,让用户看到系统当前在做什么。

for update in agent.stream(
    {"messages": [{"role": "user", "content": "帮我查一下北京天气并写个出差提醒。"}]},
    stream_mode="updates",
):
    print(update)

你可以在前端映射成:

14.9 风险控制:高风险动作双阶段设计

对于发送邮件、提交审批、发起退款、写回数据库等动作,推荐采用双阶段模式:

14.9.1 一个简单的确认流程示意

def confirm_and_send(draft: dict, confirmed: bool):
    if not confirmed:
        return {"status": "pending_confirmation", "draft": draft}
    # 真实项目中这里调用受控邮件服务
    return {"status": "sent", "draft": draft}

14.10 一个轻量路由器

并不是所有问题都该进 Agent。你可以先做一个轻量分类器:

def route_request(question: str) -> str:
    if any(word in question for word in ["报销", "制度", "流程", "规定"]):
        return "knowledge"
    if any(word in question for word in ["订单", "库存", "客户", "物流"]):
        return "agent"
    if any(word in question for word in ["审批", "发送", "退款"]):
        return "approval"
    return "agent"

路由的好处是:

14.11 评估这个 Agent 要看什么

除了“回答好不好”,多工具 Agent 还应评估:

14.12 一个简单的回归测试思路

test_cases = [
    {
        "question": "帮我查一下订单状态",
        "expect": "追问订单号"
    },
    {
        "question": "差旅报销流程怎么走?",
        "expect": "知识库"
    },
]

你可以记录:

14.13 本章小结

一个优秀的多工具 Agent 助手,不是“什么都让模型猜”,而是:

练习

1. 为你的团队设计一个不超过 6 个工具的办公 Agent。

2. 将其中一个高风险动作改成“双阶段确认”模式。

3. 写一个问题路由器,区分知识问答、工具查询和审批任务。

4. 为 Agent 设计 10 条回归测试问题,并标出期望工具路径。

5. 实现一个“缺少订单号时必须追问”的规则测试。

14.14 补充案例:审批流集成

对高风险任务,你可以让 Agent 只生成审批草稿,再交给独立审批服务:

def submit_after_approval(draft: dict, approved: bool):
    if not approved:
        return {"status": "waiting_approval", "draft": draft}
    return {"status": "submitted", "draft": draft}

14.15 补充案例:记录 Agent 轨迹

def run_agent_with_trace(agent, question: str):
    trace = []
    for update in agent.stream(
        {"messages": [{"role": "user", "content": question}]},
        stream_mode="updates",
    ):
        trace.append(update)
    return trace
 
trace = run_agent_with_trace(agent, "帮我查一下订单 ORD20260403001 的状态")
for item in trace:
    print(item)

这类轨迹非常适合: