【三个月 AI Agent 实战学习】Day 41：真实世界工具与客服实战 —

Day 41：真实世界工具与客服实战 —— 让 Agent 从“能聊”到“能办事”

欢迎来到第四十一天！过去几天我们构建了完整的记忆系统，Agent 已经能记住用户信息、自动抽取和更新记忆。但一个真正的客服 Agent 不能只会“聊天”，它还需要查询订单、追踪物流、创建工单——也就是与真实世界的业务系统交互。今天我们将为客服 Agent 添加这些工具，让它从“能聊”升级为“能办事”。我们将使用 SQLite 模拟订单数据库，用 Python 函数模拟物流 API，并实现创建工单的功能。最终，Agent 将能够处理一个完整的客服场景：用户提供订单号 → 查询订单 → 查询物流 → 投诉并创建工单 → 记住用户问题。

在这里插入图片描述

一、今日学习目标

设计一个电商客服 Agent 的完整工具集：订单查询、物流追踪、工单创建。
学会使用 SQLite 创建模拟订单数据库，并编写安全的参数化查询工具。
掌握如何将业务函数封装为 LangChain Tool，并编写清晰的工具描述。
将真实工具与记忆系统整合，构建一个“记忆感知 + 能办事”的客服 Agent。
通过多轮对话测试，验证 Agent 在复杂场景下的工具调用和记忆利用能力。

二、详细实现步骤

步骤 1：设计客服 Agent 的功能与工具

场景：用户咨询订单问题，可能涉及：

查询订单状态（已发货？待付款？）
查询物流信息（快递到哪了？）
投诉问题（物流太慢、商品破损）并创建工单
记住用户的历史问题，下次优先处理

工具列表：

工具名	功能	参数
`search_order`	根据订单号查询订单详情	`order_id: str`
`get_express`	根据订单号查询物流轨迹	`order_id: str`
`create_ticket`	创建客服工单	`order_id: str, issue: str`
`recall_memory`	检索长期记忆（已有）	`query: str`
`save_memory`	保存长期记忆（已有）	`content: str, memory_type: str`

记忆整合：

对话前自动注入用户相关记忆（如历史投诉、偏好）。
对话后自动抽取新记忆（如“用户投诉了订单 12345 的物流”）。
Agent 也可以主动调用 recall_memory 查询特定记忆。

步骤 2：环境准备与数据库初始化

确保已安装所需库：langchain、langchain-openai、langchain-community、chromadb。新建 customer_service_agent.py，导入模块并初始化。

import os
import uuid
import sqlite3
from datetime import datetime, timedelta
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate, PromptTemplate
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_core.documents import Document
from langchain_core.output_parsers import StrOutputParser
from langchain.agents import create_react_agent, AgentExecutor
from typing import Annotated

load_dotenv()

llm = ChatOpenAI(
    model="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com",
    temperature=0.1
)

embeddings = OpenAIEmbeddings(
    model="embedding-2",
    api_key=os.getenv("ZHIPU_API_KEY"),
    base_url="https://open.bigmodel.cn/api/paas/v4"
)

vectorstore = Chroma(
    collection_name="customer_service_memories",
    embedding_function=embeddings,
    persist_directory="./cs_memory_db"
)

初始化 SQLite 数据库，创建 orders 表和 tickets 表：

def init_db():
    conn = sqlite3.connect("customer_service.db")
    cursor = conn.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS orders (
            order_id TEXT PRIMARY KEY,
            user_id TEXT,
            product_name TEXT,
            status TEXT,
            express_no TEXT,
            created_at TEXT
        )
    """)
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS tickets (
            ticket_id TEXT PRIMARY KEY,
            order_id TEXT,
            issue TEXT,
            status TEXT,
            created_at TEXT
        )
    """)
    # 插入模拟数据
    cursor.execute("SELECT COUNT(*) FROM orders")
    if cursor.fetchone()[0] == 0:
        orders = [
            ("12345", "user_001", "三体全集", "已发货", "SF123456789", "2025-03-15"),
            ("67890", "user_001", "小王子", "待付款", None, "2025-03-18"),
            ("11111", "user_002", "百年孤独", "已签收", "YT987654321", "2025-03-10"),
        ]
        cursor.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?, ?)", orders)
        conn.commit()
    conn.close()

init_db()

步骤 3：定义真实世界工具

3.1 订单查询工具

@tool
def search_order(
    order_id: Annotated[str, "订单号，例如 12345"]
) -> str:
    """根据订单号查询订单详情，包括商品名称、订单状态、快递单号等。"""
    conn = sqlite3.connect("customer_service.db")
    cursor = conn.cursor()
    cursor.execute(
        "SELECT product_name, status, express_no, created_at FROM orders WHERE order_id = ?",
        (order_id,)
    )
    row = cursor.fetchone()
    conn.close()
    if row:
        product, status, express_no, created = row
        return f"订单 {order_id}：商品「{product}」，状态：{status}，快递单号：{express_no or '无'}，下单时间：{created}"
    else:
        return f"未找到订单 {order_id}，请确认订单号是否正确。"

3.2 物流查询工具

@tool
def get_express(
    order_id: Annotated[str, "订单号，例如 12345"]
) -> str:
    """根据订单号查询物流轨迹。返回最新的物流状态和预计送达时间。"""
    conn = sqlite3.connect("customer_service.db")
    cursor = conn.cursor()
    cursor.execute("SELECT express_no, status FROM orders WHERE order_id = ?", (order_id,))
    row = cursor.fetchone()
    conn.close()
    if not row:
        return f"未找到订单 {order_id}。"
    express_no, status = row
    if not express_no:
        return f"订单 {order_id} 尚未发货，暂无物流信息。"
    # 模拟物流轨迹
    if "已签收" in status:
        return f"订单 {order_id} 的快递 {express_no} 已签收。"
    else:
        return f"订单 {order_id} 的快递 {express_no} 正在运输中，预计 2 天后送达。最新状态：已到达北京分拣中心。"

3.3 创建工单工具

@tool
def create_ticket(
    order_id: Annotated[str, "关联的订单号"],
    issue: Annotated[str, "问题描述，例如：物流太慢、商品破损"]
) -> str:
    """为指定订单创建客服工单，用于记录和跟进用户投诉。"""
    ticket_id = f"T{datetime.now().strftime('%Y%m%d%H%M%S')}"
    conn = sqlite3.connect("customer_service.db")
    cursor = conn.cursor()
    cursor.execute(
        "INSERT INTO tickets VALUES (?, ?, ?, ?, ?)",
        (ticket_id, order_id, issue, "待处理", datetime.now().isoformat())
    )
    conn.commit()
    conn.close()
    return f"已为您创建工单，工单号：{ticket_id}。问题「{issue}」已记录，客服会尽快跟进。"

3.4 记忆工具（复用 Day 40）

@tool
def recall_memory(
    query: Annotated[str, "查询关键词或问题，用于检索相关记忆"]
) -> str:
    """从长期记忆中检索与查询相关的信息。当你需要回忆用户之前提到过的偏好、事实或事件时使用此工具。"""
    results = vectorstore.similarity_search(
        query, k=3, filter={"status": "active"}
    )
    if not results:
        return "未找到相关记忆。"
    lines = []
    for doc in results:
        mem_type = doc.metadata.get("memory_type", "unknown")
        lines.append(f"[{mem_type}] {doc.page_content}")
    return "\n".join(lines)

@tool
def save_memory(
    content: Annotated[str, "要保存的记忆内容，简洁明确"],
    memory_type: Annotated[str, "记忆类型：fact、preference、event"]
) -> str:
    """将重要信息保存到长期记忆中。当你发现用户提供了值得记住的新信息时使用此工具。"""
    memory_id = str(uuid.uuid4())
    doc = Document(
        page_content=content,
        metadata={
            "user_id": "current_user",
            "memory_type": memory_type,
            "importance": 3,
            "timestamp": datetime.now().isoformat(),
            "status": "active",
            "memory_id": memory_id,
            "access_count": 0
        }
    )
    vectorstore.add_documents([doc])
    return f"已保存记忆：{content}"

步骤 4：构建记忆感知型客服 Agent

4.1 记忆注入函数

def retrieve_and_format_memories(user_input: str, user_id: str = "current_user", k: int = 3) -> str:
    results = vectorstore.similarity_search(
        user_input, k=k, filter={"user_id": user_id, "status": "active"}
    )
    if not results:
        return "暂无相关记忆。"
    lines = []
    for doc in results:
        mem_type = doc.metadata.get("memory_type", "unknown")
        lines.append(f"[{mem_type}] {doc.page_content}")
    return "\n".join(lines)

def build_system_prompt_with_memories(user_input: str) -> str:
    memories_text = retrieve_and_format_memories(user_input)
    return f"""你是一个专业的电商客服助手，名叫小智。你可以查询订单、追踪物流、创建工单，并且能记住用户的历史信息。

以下是你当前知道的关于该用户的信息（来自长期记忆）：
{memories_text}

请根据这些信息以及用户当前的问题，提供准确、个性化的回复。如果需要查询订单或物流，请调用相应工具。如果记忆中没有相关信息，不要编造。
"""

4.2 创建 Agent 执行器

tools = [search_order, get_express, create_ticket, recall_memory, save_memory]

react_template = """{system_prompt}

你可以使用以下工具：

{tools}

工具名称列表：{tool_names}

使用以下格式进行推理和行动：

问题：用户提出的问题
思考：你应该思考接下来要做什么
行动：要使用的工具名称，必须是 [{tool_names}] 中的一个
行动输入：传递给工具的输入
观察：工具返回的结果
...（这个思考/行动/行动输入/观察可以重复 N 次）
思考：我现在知道最终答案了
最终答案：对用户问题的最终回答

开始！

问题：{input}
思考：{agent_scratchpad}"""

prompt = PromptTemplate.from_template(react_template)

def create_customer_service_agent(user_input: str):
    system_prompt = build_system_prompt_with_memories(user_input)
    agent = create_react_agent(llm, tools, prompt)
    executor = AgentExecutor(
        agent=agent,
        tools=tools,
        verbose=True,
        handle_parsing_errors=True,
        max_iterations=8
    )
    result = executor.invoke({
        "input": user_input,
        "system_prompt": system_prompt
    })
    return result["output"]

4.3 后处理记忆抽取

extract_prompt = ChatPromptTemplate.from_messages([
    ("system", """从以下对话中抽取值得长期记住的用户信息。只输出 JSON 数组，每个元素包含 content、memory_type、importance。
memory_type 可选：fact、preference、event。
importance 为 1-5。
如果没有值得记住的信息，输出 []。"""),
    ("human", "用户：{user_input}\nAI：{ai_response}")
])

extract_chain = extract_prompt | llm | StrOutputParser()

def post_process_memories(user_input: str, ai_response: str):
    try:
        result = extract_chain.invoke({
            "user_input": user_input,
            "ai_response": ai_response
        })
        import json
        memories = json.loads(result)
        for mem in memories:
            # 简单去重：检索相似度
            existing = vectorstore.similarity_search_with_score(
                mem["content"], k=1, filter={"status": "active"}
            )
            if existing and existing[0][1] < 0.15:
                continue
            doc = Document(
                page_content=mem["content"],
                metadata={
                    "user_id": "current_user",
                    "memory_type": mem.get("memory_type", "fact"),
                    "importance": mem.get("importance", 3),
                    "timestamp": datetime.now().isoformat(),
                    "status": "active",
                    "memory_id": str(uuid.uuid4()),
                    "access_count": 0
                }
            )
            vectorstore.add_documents([doc])
            print(f"  [记忆更新] 存入：{mem['content']}")
    except Exception as e:
        print(f"记忆抽取失败：{e}")

步骤 5：完整对话流程

def chat(user_input: str):
    print(f"\n用户：{user_input}")
    ai_response = create_customer_service_agent(user_input)
    print(f"小智：{ai_response}")
    post_process_memories(user_input, ai_response)
    return ai_response

步骤 6：测试完整客服场景

if __name__ == "__main__":
    # 场景 1：用户提供订单号并查询物流
    chat("你好，我的订单号是 12345，帮我查一下物流到哪了？")

    # 场景 2：用户投诉物流太慢
    chat("这个订单 12345 的物流太慢了，我要投诉！")

    # 场景 3：验证 Agent 是否记住了投诉历史
    chat("我之前投诉过哪个订单？")

    # 场景 4：查询另一个订单
    chat("再帮我查一下订单 67890 的状态。")

预期观察：

场景 1：Agent 调用 search_order 和 get_express，返回物流信息。
场景 2：Agent 调用 create_ticket 创建工单，并可能调用 save_memory 记录投诉事件。
场景 3：Agent 从记忆或工单中检索到之前的投诉记录。
场景 4：Agent 调用 search_order 查询新订单。

三、常见问题与调试

Q1：SQLite 数据库文件被锁定或无法写入。
→ 确保每次操作后关闭连接。多线程环境下需要使用连接池或 check_same_thread=False（但要注意线程安全）。生产环境建议使用 PostgreSQL 或 MySQL。

Q2：Agent 调用工具时参数错误，例如订单号带了引号。
→ 在工具内部对参数做清洗（如 order_id.strip().strip('"').strip("'")）。同时，在工具描述中给出示例，帮助模型正确填充。

Q3：Agent 不调用工具，直接编造订单信息。
→ 在 System Prompt 中强调“必须调用工具查询，不得编造”。降低温度至 0.1。如果仍不调用，可以在用户输入中显式提醒“请使用订单查询工具”。

Q4：工单创建后，Agent 忘记告诉用户工单号。
→ 在工具返回信息中包含工单号，Agent 通常会将其转述给用户。可以在 System Prompt 中要求“创建工单后，务必告知用户工单号”。

Q5：记忆注入的检索结果不相关，干扰 Agent。
→ 限制 k=3，并设置相似度阈值。也可以只注入 fact 和 preference 类型，不注入 event，避免历史事件干扰当前对话。

Q6：多用户场景下，如何确保订单查询只返回当前用户的订单？
→ 在工具中增加 user_id 参数，并在 SQL 查询中加入 AND user_id = ?。当前简化版没有用户隔离，生产环境必须加上。

Q7：如何测试 Agent 的鲁棒性？
→ 输入不存在的订单号、空订单号、特殊字符，观察 Agent 是否优雅处理错误。工具应返回明确的错误信息，而不是抛出异常。

四、今日总结与作业

今天你完成了：

✅ 设计了电商客服 Agent 的完整工具集：订单查询、物流追踪、工单创建。
✅ 使用 SQLite 创建了模拟数据库，并实现了安全的参数化查询工具。
✅ 将真实工具与记忆系统整合，构建了“记忆感知 + 能办事”的客服 Agent。
✅ 通过多轮对话测试，验证了 Agent 在复杂场景下的工具调用和记忆利用能力。

今日作业（必做）：

为 search_order 工具增加用户隔离：在 orders 表中已有 user_id 字段，修改工具签名增加 user_id 参数，并在 SQL 中加上 AND user_id = ?。测试不同用户查询同一订单号时是否返回正确结果。
实现一个 get_ticket_status(ticket_id) 工具，根据工单号查询工单状态。在 Agent 中测试用户询问“我的工单处理得怎么样了？”。
设计一个测试场景：用户先查询订单 12345 的物流，然后说“帮我记住这个订单号”，观察 Agent 是否调用了 save_memory，并在下一轮对话中验证记忆是否生效。
（思考题）在真实客服系统中，订单数据库可能非常庞大，直接让 Agent 调用 SQL 查询存在性能和安全风险。你会如何设计一个安全的查询层？请列出至少三个措施。

明日预告： 我们将继续完善客服 Agent，学习如何处理多轮对话中的上下文切换、如何让 Agent 在用户情绪激动时安抚用户，以及如何生成工单总结。请准备好，我们的客服 Agent 即将变得更加智能和人性化！

有任何问题欢迎随时提问！

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/jjj_web/article/details/167688274

【三个月 AI Agent 实战学习】Day 41：真实世界工具与客服实战 —— 让 Agent 从“能聊”到“能办事”