用OpenAI Agents SDK，1小时搞定客服Demo

💬 Step 1 对话接入

一开始视频的效果，接入用户端使用了微信客服
当然其他任何用户端都是ok的，web / IM / 都可以，以下代码是直接在控制台对话

用户的对话通过接入层直接到main_agent

class AgentManager:
    """芝麻酱客服系统管理器"""
    
    def__init__(self):
        self.console = Console()
    
    asyncdefrun_interactive(self):
        """运行交互式客服系统"""
        self.print_welcome()
        
        whileTrue:
            try:
                # 获取用户输入
                user_input = input("\n💬 请问有什么可以帮助您的？> ").strip()
                
                    
                ifnot user_input:
                    continue
                
                # 生成trace ID并开始处理
                trace_id = gen_trace_id()
                with trace("芝麻酱客服对话", trace_id=trace_id):
                    self.console.print(f"\n[dim]🔍 Trace ID: {trace_id}[/dim]")
                    self.console.print("[bold blue]🤖 芝麻酱正在思考中...[/bold blue]")
                    
                    start_time = time.time()
                    
                    # 使用streaming模式运行agent
                    result = Runner.run_streamed(
                        main_agent,
                        input=user_input
                    )

                .....


asyncdefmain():
    """主入口函数"""
    import sys

    manager = AgentManager()
    await manager.run_interactive()

if __name__ == "__main__":
    asyncio.run(main()) 

🔧 Step 3 分 Agent 与 Tool 实现

然后就可以分别定义handoff的agent，以及实现API调用的tool
比如录单的Agent，定义好需要什么参数，如何检查参数完整性。并且明确了什么时候使用 add_waybill的tool

录单 Agent

# 运单添加专员Agent
logistics_add_agent = Agent(
    name="Logistics_Add_Agent",
    instructions="""你是专业的运单添加专员。你的主要职责是：
1. 收集完整的运单信息，包括运单号、货物详情、车辆和司机信息
2. 确保所有必需的运单信息都已提供：运单号、发货地、卸货地、货物类型、货物重量、车牌号、司机姓名、司机电话、运费
3. 如果信息不完整，询问用户补充缺失的信息
4. 信息齐全后立即添加运单到系统

工作流程：
1. 接收用户的运单添加请求
2. 逐一收集和确认运单信息
3. 信息齐全后使用 add_waybill 工具添加运单
4. 向用户反馈添加结果

必需参数说明：
- 运单号：唯一的运单编号（如：WB20241225001）
- 发货地：货物发货地址
- 卸货地：货物卸货地址  
- 货物类型：运输的货物类型（如：钢材、电子产品等）
- 货物重量：货物重量（如：10吨、500公斤）
- 车牌号：运输车辆车牌号（如：京A12345）
- 司机姓名：司机姓名
- 司机电话：司机联系电话
- 运费：运输费用（如：5000元）
- 预计到达时间：可选参数

重要：收集齐全所有必需信息后，立即使用add_waybill工具添加运单！""",
    tools=[add_waybill],
    handoff_description="专门处理运单的添加和录入",
    model_settings=ModelSettings(temperature=0.1)
)

查单 Agent

查单Agent也一样

# 订单查询专员Agent
order_query_agent = Agent(
    name="Order_Query_Agent",
    instructions="""你是专业的订单查询专员。你的主要职责是：
1. 根据用户提供的发货地信息查询物流订单
2. 支持模糊匹配查询，如用户输入"北京"可以匹配"北京市"、"北京朝阳区"等
3. 将查询结果清晰地展示给用户
4. 如果查询结果为空，提示用户尝试其他关键词

工作流程：
1. 接收用户的查询请求，提取发货地关键词
2. 使用 query_logistics_orders 工具进行查询
3. 格式化并展示查询结果
4. 如果没有结果，建议用户尝试简化关键词或使用其他关键词

查询说明：
- 支持省份查询：如"北京"、"江苏"、"广东"等
- 支持城市查询：如"上海"、"深圳"、"杭州"等  
- 支持区域查询：如"朝阳"、"浦东"等
- 查询不区分大小写，支持部分匹配""",
    tools=[query_logistics_orders],
    handoff_description="专门处理物流订单的查询和检索",
    model_settings=ModelSettings(temperature=0.1)
)

🔌 Step 4 Tool 实现（可 Mock）

对于录单agent的tool，add_waybill，这里简化了参数和校验，实际就是请求了上游业务TMS系统的接口

运单添加工具

# 运单添加工具
@function_tool
defadd_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 能力。

订单查询工具

查单也是类似的

# 查询物流订单工具
@function_tool
defquery_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的参数

@function_tool
defprocess_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 访问 。可以显式的开启

set_tracing_disabled(disabled=False)  # 开启 tracing

主 Agent 调用：

def gen_trace_id() -> str:
    return get_trace_provider().gen_trace_id()

运行后可在 OpenAI Tracing 平台

就可以看到一轮llm的chat之后，先进入了主Agent，然后被handoff到了运单查询agent

然后可以看到tool的具体参数，返回，以及最后给用户的返回

🎨 自定义 LLM

可快速替换为自定义模型（如 siliconflow）：
然后在run Agent时候，使用就可以了

BASE_URL = os.getenv("OPENAI_BASE_URL", "https://api.siliconflow.cn/v1")
API_KEY = os.getenv("API_KEY")
MODEL_NAME = os.getenv("OPENAI_MODEL_NAME", "Qwen/Qwen3-30B-A3B")

client = AsyncOpenAI(base_url=BASE_URL, api_key=API_KEY)

classCustomModelProvider(ModelProvider):
    defget_model(self, model_name: str | None) -> Model:
        return OpenAIChatCompletionsModel(model=model_name or MODEL_NAME, openai_client=client)

CUSTOM_MODEL_PROVIDER = CustomModelProvider()

result = Runner.run_streamed(
    main_agent,
    input=user_input,
    run_config=RunConfig(model_provider=CUSTOM_MODEL_PROVIDER),
)