大家好，我是讯享网，很高兴认识大家。这里提供最前沿的Ai技术和互联网信息。



本文基于最新稳定版 openai-agents >= 0.0.15 编写，覆盖从入门到生产级的所有常用场景，是Anthropic官方推出的智能体开发工具包，封装了工具调用、记忆管理、多Agent流转、执行调度等通用能力，无需从零实现Agent调度逻辑，可快速搭建基于OpenAI模型的智能体应用。

# 基础安装 pip install openai-agents # 若需要使用内置代码执行工具，安装额外依赖 pip install openai-agents[code-execution] 

import os # 配置OpenAI API密钥，支持GPT-3.5-turbo、GPT-4o、GPT-4o-mini等模型 os.environ["OPENAI_API_KEY"] = "你的OpenAI API Key" # 国内用户需配置中转服务地址，否则无法直接访问接口 # os.environ["OPENAI_BASE_URL"] = "你的中转接口地址" 

---

---

适合本地脚本快速测试、简单交互场景：

#!/usr/bin/env python3 # -*- coding:utf-8 -*- """ 验证 openai-agent 库接入自定义大模型 (Doubao 中转接口) """ import base64 import os from dotenv import load_dotenv, find_dotenv # 加载环境变量 load_dotenv(find_dotenv()) # --- 1. 本地/中转大模型配置信息 --- APP_ID = '' APP_KEY = 'XiZZkFqiKhQNUOts' # 自定义 URL CUSTOM_BASE_URL = "http://chatgpt-api-pre.vmic.xyz/v1" # 重新组装中转平台所需的鉴权 Key CUSTOM_API_KEY = "sk-xuanji-" + APP_ID + "-" + base64.b64encode(APP_KEY.encode("utf-8")).decode() # 【关键修复】直接强制覆盖环境变量，不要用 os.getenv! # 避免系统 .env 里缓存的官方 OpenAI_API_KEY 和你自定义的 URL 混用导致 401 报错 os.environ["OPENAI_API_KEY"] = CUSTOM_API_KEY os.environ["OPENAI_BASE_URL"] = CUSTOM_BASE_URL print("👉 正在连接的 BASE URL:", os.environ["OPENAI_BASE_URL"]) print("👉 使用的 API KEY:", os.environ["OPENAI_API_KEY"]) # 【重要：必须在环境变量彻底设置完毕后，再导入 agents】 from agents import Agent, Runner def main(): # --- 2. 初始化 Agent --- agent = Agent( name="通用小助手", instructions="你是一个友好的答疑助手，回答尽量简洁清晰、通俗易懂。", model="Doubao-Seed-1.6" # 确保你自定义的接口支持此模型名 ) print(" ⏳ 正在向模型发起请求，请稍候... ") # --- 3. 执行对话并增加防崩溃捕获 --- try: # 同步执行调用 result = Runner.run_sync( agent, input="Python里生成随机数的3种常见方法是什么？" ) # 输出成功结果 print("🤖 最终回答： " + "-" * 40) print(result.final_output) print("-" * 40) except Exception as e: # 异常排查提示 print(f"❌ 执行失败，捕获到错误: {e}") print(" 💡【可能的原因及排查建议】:") print("1. 接口连通性: 如果提示 ConnectionRefused，请检查 Vmic 接口是否需要内网环境代理。") print("2. 鉴权问题: 如果提示 401，请检查 APP_ID 和 APP_KEY 是否过期。") print( "3. 模型兼容性: 'Doubao-Seed-1.6' 若不支持 Function/Tool Calling 参数，Agent 解析器可能会报错，建议尝试更换为高版本模型测试。") if __name__ == "__main__": main() 

Agent可自动调用你定义的外部能力，比如查询天气、查询订单等：

from agents import Agent, Runner, function_tool

用装饰器将普通函数注册为工具，SDK会自动解析函数注释、参数规则告知Agent

@function_tool def get_weather(city: str) -> str:

""" 查询指定城市的实时天气 Args: city: 要查询天气的城市名称，比如"北京" """ # 此处可替换为真实天气API的调用逻辑 return f"{city}今日天气：晴，气温18-25℃，风力2级" 

创建Agent并绑定工具

agent = Agent(

name="天气助手", instructions="用户查询天气时必须调用工具获取信息，禁止编造数据", tools=[get_weather], model="gpt-4o-mini" 

)

执行查询，Agent会自动判断是否需要调用工具

result = Runner.run_sync(agent, input=“北京今天天气怎么样？适合出门吗？”) print(result.final_output)

输出示例：北京今日晴，气温18-25℃，风力小，非常适合出行哦~

适合复杂业务场景，多个分工不同的Agent配合完成任务：

from agents import Agent, Runner, Handoff

技术支持Agent：负责处理产品故障、报错类问题

tech_agent = Agent(

name="技术支持", instructions="你负责解决用户的产品使用故障、报错、bug类问题，回答要给出具体可落地的解决步骤" 

)

售前Agent：负责解答产品价格、功能、购买类问题，遇到技术问题自动流转给技术支持

pre_sales_agent = Agent(

name="售前客服", instructions="你负责解答产品购买、价格、功能范围相关问题，如果遇到技术故障类问题，直接转交给技术支持处理", handoffs=[Handoff(agent=tech_agent, description="转交技术支持处理技术故障问题")] 

)

用户问技术问题时，售前Agent会自动流转给技术支持输出结果

result = Runner.run_sync(pre_sales_agent, input=“我买的产品登录时报错403怎么解决？”) print(result.final_output)

---

适合服务端集成、高并发场景，所有同步方法都有对应的异步实现：

4.1.1 基础异步对话

import asyncio from agents import Agent, Runner

async def basic_async_demo():

agent = Agent( name="编程助手", instructions="你是资深Python开发，回答代码问题要带详细注释", model="gpt-4o-mini" ) result = await Runner.run(agent, input="写一个快速排序的Python实现") print(result.final_output) 

if name == “main”:

asyncio.run(basic_async_demo()) 

4.1.2 异步工具调用

支持直接注册异步函数作为工具：

import asyncio from agents import Agent, Runner, function_tool

@function_tool async def get_order_status(order_id: str) -> str:

""" 查询用户订单的物流状态 Args: order_id: 订单号，格式为ORD开头的8位字符串 """ await asyncio.sleep(0.5) # 模拟异步接口请求延迟 # 此处可替换为异步查询订单数据库/接口的逻辑 return f"订单{order_id}状态：已发货，当前在广州转运中心，预计明天送达" 

async def tool_async_demo():

agent = Agent( name="客服助手", instructions="用户查订单必须调用工具，禁止编造信息", tools=[get_order_status], model="gpt-4o-mini" ) result = await Runner.run(agent, input="帮我查下ORD的订单到哪了") print(result.final_output) 

if name == “main”:

asyncio.run(tool_async_demo()) 

---

适合前端交互、过程可视化场景，支持两种流式模式：

4.2.1 结果增量流式（打字机效果）

仅输出Agent给用户的最终回答，增量返回，和普通大模型流式输出体验一致：

import asyncio from agents import Agent, Runner

async def stream_output_demo():

agent = Agent( name="助手", instructions="回答尽量详细，分点说明", model="gpt-4o-mini" ) async with Runner.stream(agent, input="介绍下人工智能的3个常见应用场景") as stream: # 遍历增量内容逐字打印 async for delta in stream.output_deltas: print(delta, end="", flush=True) # flush保证实时输出 

if name == “main”:

asyncio.run(stream_output_demo()) 

4.2.2 全流程事件流式（展示Agent工作过程）

可拿到Agent思考、调用工具、流转任务的全链路事件，适合做交互增强（比如给用户显示“正在调用查询工具”的提示）、调试：

import asyncio from agents import Agent, Runner, function_tool

@function_tool def get_weather(city: str) -> str:

"""查询指定城市天气""" return f"{city}今日晴，24℃" 

async def event_stream_demo():

agent = Agent( name="助手", tools=[get_weather], model="gpt-4o-mini" ) async with Runner.stream(agent, input="上海今天天气怎么样，适合出门吗") as stream: async for event in stream.events: # 不同事件类型做不同处理 if event.type == "tool_call.started": print(f" 

🤖 正在调用工具：{event.tool_call.function.name}，参数：{event.tool_call.function.arguments}“)

 elif event.type == "tool_call.result": print(f" 

🔧 工具返回结果：{event.result}”)

 elif event.type == "response.content.delta": # 最终回答的增量内容 print(event.delta.text, end="", flush=True) elif event.type == "handoff.started": print(f" 

🔄 正在转接给处理”)

if name == “main”:

asyncio.run(event_stream_demo()) 

常见事件类型说明：

---

SDK默认自动维护会话历史，无需手动存储上下文，多轮对话直接传入上一次的执行结果即可：

import asyncio from agents import Agent, Runner

async def multi_turn_demo():

agent = Agent( name="助手", model="gpt-4o-mini" ) # 第一轮对话 first_result = await Runner.run(agent, input="我叫张三，今年28岁") print("第一轮回答：", first_result.final_output) # 第二轮对话，传入previous_result即可携带上下文 second_result = await Runner.run(agent, input="我叫什么名字，今年多大？", previous_result=first_result) print("第二轮回答：", second_result.final_output) # 会正确输出你叫张三，28岁 

if name == “main”:

asyncio.run(multi_turn_demo()) 

---

支持Pydantic模型作为参数，适合复杂参数校验场景：

from pydantic import BaseModel, Field from agents import function_tool, Agent, Runner

class UserInfo(BaseModel):

name: str = Field(description="用户姓名") age: int = Field(description="用户年龄，必须是正整数") phone: str = Field(description="用户手机号，11位数字") 

@function_tool def register_user(user_info: UserInfo) -> str:

""" 注册新用户 Args: user_info: 用户的注册信息，包含姓名、年龄、手机号 """ return f"用户{user_info.name}注册成功，手机号：{user_info.phone}" 

---

SDK自带Python代码运行能力，无需自己实现工具，Agent可自动写代码、运行代码解决问题：

import asyncio from agents import Agent, Runner from agents.tools import PythonCodeExecutionTool

async def code_exec_demo():

agent = Agent( name="数据分析师", tools=[PythonCodeExecutionTool()], # 绑定内置代码执行工具 instructions="遇到计算、数据处理问题可以直接写Python代码运行解决" ) result = await Runner.run(agent, input="计算2的100次方是多少") print(result.final_output) 

if name == “main”:

asyncio.run(code_exec_demo()) 

---

执行完成后可通过steps属性查看Agent的每一步操作，方便排查问题：

result = Runner.run_sync(agent, input=“查北京天气”) for idx, step in enumerate(result.steps):

print(f" 

第{idx+1}步：”)

print("Agent思考内容：", step.input) if step.tool_calls: print("调用的工具：", [tc.function.name for tc in step.tool_calls]) print("工具返回结果：", [tc.result for tc in step.tool_calls]) 

---

agent = Agent(

name="助手", tools=[get_weather], max_tool_calls=5, # 限制最多调用5次工具，防止无限循环调用 model="gpt-4o-mini" 

)

工具调用出错时SDK会自动把错误信息返回给Agent，让Agent自行调整重试，也可以手动捕获异常

try:

result = Runner.run_sync(agent, input="查火星天气") 

except Exception as e:

print("执行出错：", e) 

---

1. 国内使用必须配置OPENAI_BASE_URL为合法的中转服务地址，否则无法连接OpenAI接口；
2. 工具的参数注释、字段描述一定要写清晰，否则Agent可能无法正确识别调用规则；
3. 多轮工具调用、大上下文会消耗更多Token，建议根据场景合理设置max_tool_calls和上下文长度，控制成本；
4. 涉及敏感数据的工具建议增加额外参数校验、权限判断，避免Agent传入非法参数导致安全问题；
5. SDK仍在快速迭代阶段，建议固定版本使用，避免API变动导致代码报错。