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，导入模块并初始化。
importosimportuuidimportsqlite3fromdatetimeimportdatetime,timedeltafromdotenvimportload_dotenvfromlangchain_openaiimportChatOpenAI,OpenAIEmbeddingsfromlangchain_community.vectorstoresimportChromafromlangchain_core.toolsimporttoolfromlangchain_core.promptsimportChatPromptTemplate,PromptTemplatefromlangchain_core.messagesimportHumanMessage,AIMessage,SystemMessagefromlangchain_core.documentsimportDocumentfromlangchain_core.output_parsersimportStrOutputParserfromlangchain.agentsimportcreate_react_agent,AgentExecutorfromtypingimportAnnotated 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表：
definit_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")ifcursor.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 订单查询工具
@tooldefsearch_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()ifrow:product,status,express_no,created=rowreturnf"订单{order_id}：商品「{product}」，状态：{status}，快递单号：{express_noor'无'}，下单时间：{created}"else:returnf"未找到订单{order_id}，请确认订单号是否正确。"3.2 物流查询工具
@tooldefget_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()ifnotrow:returnf"未找到订单{order_id}。"express_no,status=rowifnotexpress_no:returnf"订单{order_id}尚未发货，暂无物流信息。"# 模拟物流轨迹if"已签收"instatus:returnf"订单{order_id}的快递{express_no}已签收。"else:returnf"订单{order_id}的快递{express_no}正在运输中，预计 2 天后送达。最新状态：已到达北京分拣中心。"3.3 创建工单工具
@tooldefcreate_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()returnf"已为您创建工单，工单号：{ticket_id}。问题「{issue}」已记录，客服会尽快跟进。"3.4 记忆工具（复用 Day 40）
@tooldefrecall_memory(query:Annotated[str,"查询关键词或问题，用于检索相关记忆"])->str:"""从长期记忆中检索与查询相关的信息。当你需要回忆用户之前提到过的偏好、事实或事件时使用此工具。"""results=vectorstore.similarity_search(query,k=3,filter={"status":"active"})ifnotresults:return"未找到相关记忆。"lines=[]fordocinresults:mem_type=doc.metadata.get("memory_type","unknown")lines.append(f"[{mem_type}]{doc.page_content}")return"\n".join(lines)@tooldefsave_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])returnf"已保存记忆：{content}"步骤 4：构建记忆感知型客服 Agent
4.1 记忆注入函数
defretrieve_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"})ifnotresults:return"暂无相关记忆。"lines=[]fordocinresults:mem_type=doc.metadata.get("memory_type","unknown")lines.append(f"[{mem_type}]{doc.page_content}")return"\n".join(lines)defbuild_system_prompt_with_memories(user_input:str)->str:memories_text=retrieve_and_format_memories(user_input)returnf"""你是一个专业的电商客服助手，名叫小智。你可以查询订单、追踪物流、创建工单，并且能记住用户的历史信息。 以下是你当前知道的关于该用户的信息（来自长期记忆）：{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)defcreate_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})returnresult["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()defpost_process_memories(user_input:str,ai_response:str):try:result=extract_chain.invoke({"user_input":user_input,"ai_response":ai_response})importjson memories=json.loads(result)formeminmemories:# 简单去重：检索相似度existing=vectorstore.similarity_search_with_score(mem["content"],k=1,filter={"status":"active"})ifexistingandexisting[0][1]<0.15:continuedoc=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']}")exceptExceptionase:print(f"记忆抽取失败：{e}")步骤 5：完整对话流程
defchat(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)returnai_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 即将变得更加智能和人性化！
有任何问题欢迎随时提问！
