Day 41真实世界工具与客服实战 —— 让 Agent 从“能聊”到“能办事”欢迎来到第四十一天过去几天我们构建了完整的记忆系统Agent 已经能记住用户信息、自动抽取和更新记忆。但一个真正的客服 Agent 不能只会“聊天”它还需要查询订单、追踪物流、创建工单——也就是与真实世界的业务系统交互。今天我们将为客服 Agent 添加这些工具让它从“能聊”升级为“能办事”。我们将使用 SQLite 模拟订单数据库用 Python 函数模拟物流 API并实现创建工单的功能。最终Agent 将能够处理一个完整的客服场景用户提供订单号 → 查询订单 → 查询物流 → 投诉并创建工单 → 记住用户问题。一、今日学习目标设计一个电商客服 Agent 的完整工具集订单查询、物流追踪、工单创建。学会使用 SQLite 创建模拟订单数据库并编写安全的参数化查询工具。掌握如何将业务函数封装为 LangChain Tool并编写清晰的工具描述。将真实工具与记忆系统整合构建一个“记忆感知 能办事”的客服 Agent。通过多轮对话测试验证 Agent 在复杂场景下的工具调用和记忆利用能力。二、详细实现步骤步骤 1设计客服 Agent 的功能与工具场景用户咨询订单问题可能涉及查询订单状态已发货待付款查询物流信息快递到哪了投诉问题物流太慢、商品破损并创建工单记住用户的历史问题下次优先处理工具列表工具名功能参数search_order根据订单号查询订单详情order_id: strget_express根据订单号查询物流轨迹order_id: strcreate_ticket创建客服工单order_id: str, issue: strrecall_memory检索长期记忆已有query: strsave_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()llmChatOpenAI(modeldeepseek-chat,api_keyos.getenv(DEEPSEEK_API_KEY),base_urlhttps://api.deepseek.com,temperature0.1)embeddingsOpenAIEmbeddings(modelembedding-2,api_keyos.getenv(ZHIPU_API_KEY),base_urlhttps://open.bigmodel.cn/api/paas/v4)vectorstoreChroma(collection_namecustomer_service_memories,embedding_functionembeddings,persist_directory./cs_memory_db)初始化 SQLite 数据库创建orders表和tickets表definit_db():connsqlite3.connect(customer_service.db)cursorconn.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:根据订单号查询订单详情包括商品名称、订单状态、快递单号等。connsqlite3.connect(customer_service.db)cursorconn.cursor()cursor.execute(SELECT product_name, status, express_no, created_at FROM orders WHERE order_id ?,(order_id,))rowcursor.fetchone()conn.close()ifrow:product,status,express_no,createdrowreturnf订单{order_id}商品「{product}」状态{status}快递单号{express_noor无}下单时间{created}else:returnf未找到订单{order_id}请确认订单号是否正确。3.2 物流查询工具tooldefget_express(order_id:Annotated[str,订单号例如 12345])-str:根据订单号查询物流轨迹。返回最新的物流状态和预计送达时间。connsqlite3.connect(customer_service.db)cursorconn.cursor()cursor.execute(SELECT express_no, status FROM orders WHERE order_id ?,(order_id,))rowcursor.fetchone()conn.close()ifnotrow:returnf未找到订单{order_id}。express_no,statusrowifnotexpress_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_idfT{datetime.now().strftime(%Y%m%d%H%M%S)}connsqlite3.connect(customer_service.db)cursorconn.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 40tooldefrecall_memory(query:Annotated[str,查询关键词或问题用于检索相关记忆])-str:从长期记忆中检索与查询相关的信息。当你需要回忆用户之前提到过的偏好、事实或事件时使用此工具。resultsvectorstore.similarity_search(query,k3,filter{status:active})ifnotresults:return未找到相关记忆。lines[]fordocinresults:mem_typedoc.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_idstr(uuid.uuid4())docDocument(page_contentcontent,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构建记忆感知型客服 Agent4.1 记忆注入函数defretrieve_and_format_memories(user_input:str,user_id:strcurrent_user,k:int3)-str:resultsvectorstore.similarity_search(user_input,kk,filter{user_id:user_id,status:active})ifnotresults:return暂无相关记忆。lines[]fordocinresults:mem_typedoc.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_textretrieve_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}promptPromptTemplate.from_template(react_template)defcreate_customer_service_agent(user_input:str):system_promptbuild_system_prompt_with_memories(user_input)agentcreate_react_agent(llm,tools,prompt)executorAgentExecutor(agentagent,toolstools,verboseTrue,handle_parsing_errorsTrue,max_iterations8)resultexecutor.invoke({input:user_input,system_prompt:system_prompt})returnresult[output]4.3 后处理记忆抽取extract_promptChatPromptTemplate.from_messages([(system,从以下对话中抽取值得长期记住的用户信息。只输出 JSON 数组每个元素包含 content、memory_type、importance。 memory_type 可选fact、preference、event。 importance 为 1-5。 如果没有值得记住的信息输出 []。),(human,用户{user_input}\nAI{ai_response})])extract_chainextract_prompt|llm|StrOutputParser()defpost_process_memories(user_input:str,ai_response:str):try:resultextract_chain.invoke({user_input:user_input,ai_response:ai_response})importjson memoriesjson.loads(result)formeminmemories:# 简单去重检索相似度existingvectorstore.similarity_search_with_score(mem[content],k1,filter{status:active})ifexistingandexisting[0][1]0.15:continuedocDocument(page_contentmem[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_responsecreate_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 的状态。)预期观察场景 1Agent 调用search_order和get_express返回物流信息。场景 2Agent 调用create_ticket创建工单并可能调用save_memory记录投诉事件。场景 3Agent 从记忆或工单中检索到之前的投诉记录。场景 4Agent 调用search_order查询新订单。三、常见问题与调试Q1SQLite 数据库文件被锁定或无法写入。→ 确保每次操作后关闭连接。多线程环境下需要使用连接池或check_same_threadFalse但要注意线程安全。生产环境建议使用 PostgreSQL 或 MySQL。Q2Agent 调用工具时参数错误例如订单号带了引号。→ 在工具内部对参数做清洗如order_id.strip().strip().strip()。同时在工具描述中给出示例帮助模型正确填充。Q3Agent 不调用工具直接编造订单信息。→ 在 System Prompt 中强调“必须调用工具查询不得编造”。降低温度至 0.1。如果仍不调用可以在用户输入中显式提醒“请使用订单查询工具”。Q4工单创建后Agent 忘记告诉用户工单号。→ 在工具返回信息中包含工单号Agent 通常会将其转述给用户。可以在 System Prompt 中要求“创建工单后务必告知用户工单号”。Q5记忆注入的检索结果不相关干扰 Agent。→ 限制k3并设置相似度阈值。也可以只注入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 即将变得更加智能和人性化有任何问题欢迎随时提问