OpenAI Agent SDK

设计原则

• 具有足够的功能，但不增加过多的新东西
• 开箱即用，可以自定义内部细节

主要功能

• 代理循环：内置，负责调用工具，结果给LLM并循环直到完成
• 交接：多个代理之间的协调和委派
• 护栏：与代理并运行输入验证和检查，如果检查失败则提前中断
• Python优先：使用内置语言功能来协调和链接代理，不需要学习新的抽象概念
• 函数工具：将任何Python函数转为工具
• 跟踪：跟踪调试，监控，支持可视化

安装

pip install openai-agents

基础使用

from openai import AsyncOpenAI
from agents import (
    set_default_openai_client,
    set_default_openai_api,
    set_tracing_disabled
)

# Ollama 配置
OPENAI_BASE_URL = "http://localhost:11434/v1"
OPENAI_API_KEY = "ollama"
OPENAI_MODEL_NAME = "qwen2.5:7b-instruct-q4_K_M"

client = AsyncOpenAI(
    base_url=OPENAI_BASE_URL,
    api_key=OPENAI_API_KEY,
)

# 使用自定义客户端
set_default_openai_client(client=client, use_for_tracing=False)
# 使用兼容的API模式
set_default_openai_api("chat_completions")
# 禁用OpenAI的跟踪服务
set_tracing_disabled(disabled=True)

from agents import Agent, Runner

# 关键：在 Agent 构造函数中直接指定 model 参数
agent = Agent(
    name="AI助手",
    instructions="你是一个友好的AI助手，使用中文交流",
    model=OPENAI_MODEL_NAME  # 添加这一行，明确指定模型名称
)

result = Runner.run_sync(agent, "你是谁")
print(result.final_output)

异步写法

from openai import AsyncOpenAI
from agents import (
    set_default_openai_client,
    set_default_openai_api,
    set_tracing_disabled
)
from agents import Agent, Runner
import asyncio

# Ollama 配置
OPENAI_BASE_URL = "http://localhost:11434/v1"
OPENAI_API_KEY = "ollama"
OPENAI_MODEL_NAME = "qwen2.5:7b-instruct-q4_K_M"

client = AsyncOpenAI(
    base_url=OPENAI_BASE_URL,
    api_key=OPENAI_API_KEY,
)

# 使用自定义客户端
set_default_openai_client(client=client, use_for_tracing=False)
# 使用兼容的API模式
set_default_openai_api("chat_completions")
# 禁用OpenAI的跟踪服务
set_tracing_disabled(disabled=True)

agent = Agent(
    name="AI助手",
    instructions="你是一个友好的AI助手，使用中文交流",
    model=OPENAI_MODEL_NAME  # 添加这一行，明确指定模型名称
)

async def main():
    result = await Runner.run(agent, "你是谁")
    print(result.final_output)

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

流式传输

from openai import AsyncOpenAI
from agents import (
    set_default_openai_client,
    set_default_openai_api,
    set_tracing_disabled
)
from agents import Agent, Runner
import asyncio

# Ollama 配置
OPENAI_BASE_URL = "http://localhost:11434/v1"
OPENAI_API_KEY = "ollama"
OPENAI_MODEL_NAME = "qwen2.5:7b-instruct-q4_K_M"

client = AsyncOpenAI(
    base_url=OPENAI_BASE_URL,
    api_key=OPENAI_API_KEY,
)

# 使用自定义客户端
set_default_openai_client(client=client, use_for_tracing=False)
# 使用兼容的API模式
set_default_openai_api("chat_completions")
# 禁用OpenAI的跟踪服务
set_tracing_disabled(disabled=True)

agent = Agent(
    name="AI助手",
    instructions="你是一个友好的AI助手，使用中文交流",
    model=OPENAI_MODEL_NAME  # 添加这一行，明确指定模型名称
)


async def main():
    # 注意：不需要 await，直接调用
    result = Runner.run_streamed(agent, "讲一个300字的故事")

    # 遍历流式事件
    async for event in result.stream_events():
        if event.type == "raw_response_event":
            # raw_response_event 来自LLM的响应
            if hasattr(event.data, 'delta') and event.data.delta:
                print(event.data.delta, end='', flush=True)
        elif event.type == "run_item_stream_event":
            # run_item_stream_event 来自runner事件（工具调用，交接）
            # 可以选择打印工具调用等信息
            pass

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

连续对话

from openai import AsyncOpenAI
from agents import (
    set_default_openai_client,
    set_default_openai_api,
    set_tracing_disabled
)
from agents import Agent, Runner
import asyncio

# Ollama 配置
OPENAI_BASE_URL = "http://localhost:11434/v1"
OPENAI_API_KEY = "ollama"
OPENAI_MODEL_NAME = "qwen2.5:7b-instruct-q4_K_M"

client = AsyncOpenAI(
    base_url=OPENAI_BASE_URL,
    api_key=OPENAI_API_KEY,
)

# 使用自定义客户端
set_default_openai_client(client=client, use_for_tracing=False)
# 使用兼容的API模式
set_default_openai_api("chat_completions")
# 禁用OpenAI的跟踪服务
set_tracing_disabled(disabled=True)

agent = Agent(
    name="AI助手",
    instructions="你是一个友好的AI助手，使用中文交流",
    model=OPENAI_MODEL_NAME  # 添加这一行，明确指定模型名称
)

# 使用列表存储对话历史
conversation_history = []


async def main(msg):
    global conversation_history
    conversation_history.append({
        "role": "user",
        "content": msg,
    })
    result = Runner.run_streamed(agent, conversation_history)
    # 收集助手的完整回复
    full_response = ""
    # 遍历流式事件
    async for event in result.stream_events():
        if event.type == "raw_response_event":
            # raw_response_event 来自LLM的响应
            if hasattr(event.data, 'delta') and event.data.delta:
                print(event.data.delta, end='', flush=True)
                full_response += event.data.delta
        elif event.type == "run_item_stream_event":
            # run_item_stream_event 来自runner事件（工具调用，交接）
            # 可以选择打印工具调用等信息
            pass

    # 添加助手的回复到历史
    conversation_history.append({
        "role": "assistant",
        "content": full_response,
    })


async def run_conversation():
    await main("1+1=?")
    await asyncio.sleep(3)
    print()
    await main("再加1呢")
    await asyncio.sleep(3)
    print()
    await main("再加1呢")


if __name__ == '__main__':
    asyncio.run(run_conversation())

多模态

import base64
from openai import AsyncOpenAI
from agents import (
    set_default_openai_client,
    set_default_openai_api,
    set_tracing_disabled
)
from agents import Agent, Runner
import asyncio

# Ollama 配置
OPENAI_BASE_URL = "http://localhost:11434/v1"
OPENAI_API_KEY = "ollama"
OPENAI_MODEL_NAME = "qwen3-vl:2b"  # 改成支持多模态的模型！

client = AsyncOpenAI(
    base_url=OPENAI_BASE_URL,
    api_key=OPENAI_API_KEY,
)

# 使用自定义客户端
set_default_openai_client(client=client, use_for_tracing=False)
# 使用兼容的API模式
set_default_openai_api("chat_completions")
# 禁用OpenAI的跟踪服务
set_tracing_disabled(disabled=True)

agent = Agent(
    name="AI助手",
    instructions="你是一个友好的AI助手，使用中文交流",
    model=OPENAI_MODEL_NAME
)


# 图片转base64
def encode_image(image_path):
    with open(image_path, 'rb') as image_file:
        bin_content = image_file.read()
        return base64.b64encode(bin_content).decode("utf-8")


async def main():
    base64_image = encode_image("1.png")

    message = {
        "role": "user",
        "content": [
            {
                "type": "image_url",
                "image_url": {
                    "url": f"data:image/png;base64,{base64_image}"
                }
            },
            {
                "type": "text",
                "text": "图片中有什么内容？"
            }
        ]
    }

    result = Runner.run_streamed(agent, message)

    # 遍历流式事件
    async for event in result.stream_events():
        if event.type == "raw_response_event":
            # raw_response_event 来自LLM的响应
            if hasattr(event.data, 'delta') and event.data.delta:
                print(event.data.delta, end='', flush=True)
        elif event.type == "run_item_stream_event":
            # run_item_stream_event 来自runner事件（工具调用，交接）
            pass


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