手把手教你用 OpenAI Agents SDK 实现客服Multi Agent

手把手教你用 OpenAI Agents SDK 实现客服Multi Agent 手把手教你用 OpenAI Agents SDK 实现客服Multi Agent Modified July 21, 2025 🔌 Step 4 实现Tool（或者如何mock） 对于录单agent的tool，add waybill，这里简化了参数和校验，实际就是请求了上游业务TMS系统的接口 Code block Python Copy 运单添加工具 @function tool def add waybill( 运单号: str, 发货地: str, 卸货地: str, 货物类型: str, ) str: """添加运单到系统 Args: 运单号: 运单编号 发货地: 货物发货地址 卸货地: 货物卸货地址 货物类型: 运输的货物类型 """ logger.info(f"[运单添加] 准备添加运单 运单号: {运单号}") 构建运单数据 waybill data = { '运单号': 运单号, '发货地': 发货地, '卸货地': 卸货地, '货物类型': 货物类型, '状态': '运输中' } 记录到专用请求响应日志 req res logger.log raw request({ "url": "http://localhost:3001/api/waybills", "method": "POST", "headers": {"Content Type": "application/json"}, "data": waybill data, }, "waybill add") try: 使用同步的httpx客户端，避免事件循环冲突 import httpx with httpx.Client() as client: response = client.post( "http://localhost:3001/api/waybills", json=waybill data, headers={'Content Type': 'application/json'}, timeout=30.0 ) 记录响应 logger.info(f"[运单添加] 响应状态码: {response.status code}") logger.info(f"[运单添加] 响应内容: {response.text}") if response.status code == 200: result = f"""✅ 运单添加成功！ 如果没有上游系统，完全可以写一段mock，随机一个运单号之类返回，这样可以快速吧客服Agent的能力跑起来 查单也是类似的 Code block Python Copy 查询物流订单工具 @function tool def query logistics orders(发货地: str) str: """根据发货地查询物流订单 Args: 发货地: 发货地址，支持模糊匹配 """ logger.info(f"[订单查询] 查询发货地: {发货地}") try: URL编码中文参数 import urllib.parse encoded 发货地 = urllib.parse.quote(发货地) encoded param name = urllib.parse.quote('发货地') query url = f"http://localhost:3001/api/orders?{encoded param name}={encoded 发货地}" 使用同步的httpx客户端发送请求 import httpx with httpx.Client() as client: response = client.get(query url, timeout=30.0) 记录响应 logger.info(f"[订单查询] 响应状态码: {response.status code}") logger.info(f"[订单查询] 响应内容: {response.text}") if response.status code == 200: try: orders data = response.json() ..... 参数有效性 在请求接口前，还应该注意增加参数处理的环节 比如类似单独定义一个 process logistics order，用JSON mode之类去提取有效参数，规范请求API的参数 Code block Python Copy @function tool def process logistics order(user input: str) str: """智能处理物流订单：使用AI分析用户输入并提取参数 Args: user input: 用户输入的订单信息 """ logger.info(f"[智能处理] 开始AI分析物流订单参数") 使用AI来提取参数 extraction prompt = f""" 请从以下用户输入中提取物流订单的4个关键参数，并以JSON格式返回。如果某个参数缺失，请在JSON中将该字段设为null。 用户输入："{user input}" 需要提取的参数： 1. 发货地：货物的起始地址、装货地、出发地等 2. 卸货地：货物的目的地址、终点、送货地等 3. 货物类型：运输的货物种类或名称 4. 所需车辆：需要的车辆类型或规格 请严格按照以下JSON格式返回，不要包含任何其他文字： {{ "发货地": "提取的发货地址或null", "卸货地": "提取的卸货地址或null", "货物类型": "提取的货物类型或null", "所需车辆": "提取的车辆类型或null" }} """ try: 使用同步的OpenAI客户端进行参数提取 import openai import json 创建OpenAI客户端 client = openai.OpenAI() response = client.chat.completions.create( model="gpt 3.5 turbo", messages=[ {"role": "system", "content": "你是专业的物流信息提取助手，请严格按照要求以JSON格式返回提取的参数。"}, {"role": "user", "content": extraction prompt} ], temperature=0.1, max tokens=200 ) 解析AI返回的JSON ai response = response.choices[0].message.content.strip() logger.info(f"[智能处理] AI提取结果: {ai response}") 解析JSON响应 try: extracted params = json.loads(ai response) except json.JSONDecodeError: 如果AI返回的不是有效JSON，尝试提取JSON部分 import re json match = re.search(r'\{. \}', ai response, re.DOTALL) if json match: extracted params = json.loads(json match.group()) else: raise ValueError("AI返回的不是有效的JSON格式") 检查哪些参数缺失 required params = ["发货地", "卸货地", "货物类型", "所需车辆"] missing params = [] found params = {} for param in required params: value = extracted params.get(param) if value and value.lower() != "null" and value.strip(): found params[param] = value.strip() logger.info(f"[智能处理] {param}: AI提取到 '{value}'") else: missing params.append(param) 构建返回结果 if missing params: missing list = "、".join(missing params) result = f"📋 物流订单信息不完整，还需要以下信息：\n\n{missing list}\n\n请提供缺失的信息，我将帮您完成订单提交。" logger.info(f"[智能处理] 缺少参数: {missing params}") else: 参数齐全，明确指示需要提交订单，并提供参数 发货地 = found params.get('发货地', '未知') 卸货地 = found params.get('卸货地', '未知') 货物类型 = found params.get('货物类型', '未知') 所需车辆 = found params.get('所需车辆', '未知') result = f"🎉 物流订单信息完整！\n\n提取到的参数：\n✅ 发货地: {发货地}\n✅ 卸货地: {卸货地}\n✅ 货物类型: {货物类型}\n✅ 所需车辆: {所需车辆}\n\n 现在立即调用submit logistics order工具提交订单！ \n\n参数：发货地='{发货地}', 卸货地='{卸货地}', 货物类型='{货物类型}', 所需车辆='{所需车辆}'" logger.info(f"[智能处理] 参数齐全，准备提交: {found params}") except Exception as e: logger.error(f"[智能处理] AI参数提取失败: {e}") result = f"❌ 参数提取失败：{str(e)}\n\n请手动提供以下信息：\n 发货地\n 卸货地\n 货物类型\n 所需车辆" return result 🧠 Tracing SDK自带 完备的Tracing 访问 。可以显式的开启 Code block Python Copy set tracing disabled(disabled=False) 开启tracing 主agent调用 def gen trace id() str: """Generate a new trace ID.""" return get trace provider().gen trace id() 运行后，访问 https://platform.openai.com/traces 就可以看到一轮llm的chat之后，先进入了主Agent，然后被handoff到了运单查询agent 然后可以看到tool的具体参数，返回，以及最后给用户的返回 🎨 自定义LLM SDK提供了快速替代OpenAI的模型的方法 比如想用 siliconflow 的接口 只需要先定义一个 CustomModelProvider Code block Python Copy 🔌 Step 4 实现Tool（或者如何mock） 对于录单agent的tool，add waybill，这里简化了参数和校验，实际就是请求了上游业务TMS系统的接口 如果没有上游系统，完全可以写一段mock，随机一个运单号之类返回，这样可以快速吧客服Agent的能力跑起来 查单也是类似的 参数有效性 在请求接口前，还应该注意增加参数处理的环节 比如类似单独定义一个 process logistics order，用JSON mode之类去提取有效参数，规范请求API的参数 🧠 Tracing SDK自带 完备的Tracing 访问 。可以显式的开启 运行后，访问 https://platform.openai.com/traces 就可以看到一轮llm的chat之后，先进入了主Agent，然后被handoff到了运单查询agent 然后可以看到tool的具体参数，返回，以及最后给用户的返回 🎨 自定义LLM SDK提供了快速替代OpenAI的模型的方法 比如想用 siliconflow 的接口 只需要先定义一个 CustomModelProvider 然后在run Agent时候，使用就可以了 🎁 Todos 实际Demo需求如下，不过考虑篇幅只抽象了核心的内容。但是终归离生产还有很远的距离 包括并不限于 • 基于用户的 Context Engineering 模块：用于构建用户状态和上下文，提升智能体响应的个性化与准确性。 • Access Control 权限控制：如在录单等操作中，需要精细化控制不同角色的访问与操作权限，确保系统安全合规。 • HITL（Human in the Loop）机制：尤其在投诉等复杂场景中，引入人工审阅或介入，能显著提升处理质量与用户满意度。 官方 SDK 中已经提供了许多快速实现的示例（GitHub 示例仓库），可供深入学习与原型开发参考。 GitHub 示例仓库 在此基础上，建议根据自身业务场景、团队技术栈与系统需求，设计一套符合实际需求的智能体系统架构，灵活集成模型、接口与权限策略，形成具备可扩展性与稳定性的解决方案 以上就是 手把手教你用 OpenAI Agents SDK 实现客服Multi Agent 的内容啦，希望对你有帮助 如果需要完整demo代码，或者有问题欢迎联系AI二八酱沟通 🧠 持续分享 AI相关 / Agent / 提效工具 / 一起学习一起变强 📚 Engineering & 经济学双硕士 💻 独立开发者 🧚‍♂️ ISTJ to ISTP 💬 欢迎评论交流 🌟 大家好啊，我是喜欢吃麻酱的AI二八酱😊 前几天用OpenAI 的 Agents SDK和物流客户共创了一个客服Agent的Demo，非常快速的实现了想要的效果。今天就来分享一下 大家好啊，我是喜欢吃麻酱的AI二八酱😊 前几天用OpenAI 的 Agents SDK和物流客户共创了一个客服Agent的Demo，非常快速的实现了想要的效果。今天就来分享一下 主要内容 SDK核心组件解析 Demo实现的拆解：如何定义Agent，设置handoff；如何上游服务API对接或mock Tracing 和 真实生产的Todos SDK核心组件解析 Demo实现的拆解：如何定义Agent，设置handoff；如何上游服务API对接或mock Tracing 和 真实生产的Todos 先来看下最后实现的效果 📖Demo场景和框架说明 业务场景是一个运输服务的司机端的智能客服，用户常见的问题场景一般都是关于资质审批，运单查询，录单之类 所以除了需要基础的Caht能力之外，还需要有稳定的ToolUse能力，当然还有成熟的Guardrails和Tracing组件 其实可以用的框架有很多，LangChain / AutoGPT / AutoGen等等。 这次用的 OpenAI Agents SDK 可以 ✅方便快速根据业务场景，实现demo ✅代码可读性强 ✅可以快速更换LLM模型 下文用的是SDK的Pyhon版本 https://github.com/openai/openai agents python 还有JavaScript/TypeScript的可以参考 https://github.com/openai/openai agents js/tree/main?tab=readme ov file 🛠️ 核心组件 Agent & Handoffs 核心当然是Agent啦，可以定义 instructions，以及在这个Demo中，实现 Multi Agent 的Handoffs 比如在这个简化的版本中，定义了一个主Agent，以及两个Handoffs的Agent • 录单的 logistics add agent • 查询运单的 order query agent Guardrails & Sessions & Tracing SDK还提供了丰富的可观测性（Observability）生产必要组件，可以快速实现 日志以及调用链路（工具调用、模型推理）的追踪 安全护栏（Guardrails） 功能：可配置的输入输出安全检查机制 作用：过滤有害内容、验证数据格式、防止越权操作，确保Agent行为符合合规要求和伦理准则 会话管理（Sessions） 功能：跨运行周期的对话历史自动管理 作用：维持上下文连续性，支持长对话记忆，避免重复提问，提升交互连贯性 追踪系统（Tracing） 功能：Agent运行全流程记录与可视化 作用：提供执行链路追踪、错误定位、性能分析，支持工作流优化与调试 不过除了Tracing，其他内容Demo中暂时没有涉及。另外还有很多Agent设计模式，还有example可以参考哈，比如人工介入的处理 🎓 Demo结构 整体框架如下图。主agent识别意图并分发给专业agent处理，实际功能实现由agent调用原有系统API 💬 Step 1 对话接入 一开始视频的效果，接入用户端使用了微信客服 当然其他任何用户端都是ok的，web / IM / 都可以，以下代码是直接在控制台对话 用户的对话通过接入层直接到main agent 🎯 Step 2 主Agent和意图识别 这里main agent就是上文提到的主Agent。可以通过instructions来明确所有能力，以及什么情况下交由哪个专业Agent完成的handoffs 🔧 Step 3 分Agent和tools 然后就可以分别定义handoff的agent，以及实现API调用的tool 比如录单的Agent，定义好需要什么参数，如何检查参数完整性。并且明确了什么时候使用 add waybill的tool 查单Agent也一样