autogen_ext.agents.openai#

类 OpenAIAgent(name: str, description: str, client: AsyncOpenAI | AsyncAzureOpenAI, model: str, instructions: str, tools: Iterable[Literal['web_search_preview', 'image_generation', 'local_shell'] | FileSearchToolConfig | WebSearchToolConfig | ComputerUseToolConfig | MCPToolConfig | CodeInterpreterToolConfig | ImageGenerationToolConfig | LocalShellToolConfig] | None = None, temperature: float | None = 1, max_output_tokens: int | None = None, json_mode: bool = False, store: bool = True, truncation: str = 'disabled')[来源]#

基类：BaseChatAgent, Component[OpenAIAgentConfig]

一个使用 OpenAI 响应 API 生成响应的代理实现。

安装

pip install "autogen-ext[openai]"
# pip install "autogen-ext[openai,azure]"  # For Azure OpenAI Assistant

该代理利用响应 API 生成具有以下功能的响应：

多轮对话
内置工具支持（文件搜索、代码解释器、网络搜索预览等）

目前不支持自定义工具。

v0.7.0 版本更新：增加了对内置工具类型的支持，例如文件搜索、网络搜索预览、代码解释器、计算机使用预览、图像生成和 MCP。增加了对具有必需和可选参数的工具配置的支持。

内置工具分为两类

可以使用字符串格式的工具（无必需参数）

web_search_preview：可以用作“web_search_preview”或使用可选配置（user_location、search_context_size）

image_generation：可以用作“image_generation”或使用可选配置（background、input_image_mask）

local_shell：可以用作“local_shell”（警告：仅适用于 codex-mini-latest 模型）

需要字典配置的工具（具有必需参数）

file_search：必须使用带有 vector_store_ids (List[str]) 的字典

computer_use_preview：必须使用带有 display_height (int)、display_width (int)、environment (str) 的字典

code_interpreter：必须使用带有 container (str) 的字典

mcp：必须使用带有 server_label (str)、server_url (str) 的字典

以字符串格式使用必需参数工具将引发 ValueError 并附带 F 有用的错误消息。工具参数类型注释仅接受不需要参数的工具的字符串值。

注意

此代理不支持自定义工具（autogen FunctionTool 或其他用户定义的工具）。仅支持通过 Responses API 提供的 OpenAI 内置工具。

参数:

name (str) – 代理的名称
description (str) – 代理目的的描述
client (Union[AsyncOpenAI, AsyncAzureOpenAI]) – OpenAI 客户端实例
model (str) – 要使用的模型（例如“gpt-4.1”）
instructions (str) – 代理的系统指令
tools (Optional[Iterable[Union[str, BuiltinToolConfig]]]) – 代理可以使用的工具。支持的字符串值（无必需参数）：“web_search_preview”、“image_generation”、“local_shell”。字典值可以为带参数的内置工具提供配置。内置工具的必需参数：- file_search：vector_store_ids (List[str]) - computer_use_preview：display_height (int)、display_width (int)、environment (str) - code_interpreter：container (str) - mcp：server_label (str)、server_url (str) 内置工具的可选参数：- file_search：max_num_results (int)、ranking_options (dict)、filters (dict) - web_search_preview：user_location (str 或 dict)、search_context_size (int) - image_generation：background (str)、input_image_mask (str) - mcp：allowed_tools (List[str])、headers (dict)、require_approval (bool) 具有模型限制的特殊工具：- local_shell：仅适用于“codex-mini-latest”模型（警告：支持非常有限）。不支持自定义工具。
temperature (Optional[float]) – 响应生成的温度（默认值：1）
max_output_tokens (Optional[int]) – 最大输出令牌数
json_mode (bool) – 是否使用 JSON 模式（默认值：False）
store (bool) – 是否存储对话（默认值：True）
truncation (str) – 截断策略（默认值：“disabled”）

示例

内置工具的基本用法

import asyncio

from autogen_agentchat.ui import Console
from autogen_ext.agents.openai import OpenAIAgent
from openai import AsyncOpenAI


async def example():
    client = AsyncOpenAI()
    agent = OpenAIAgent(
        name="SimpleAgent",
        description="A simple OpenAI agent using the Responses API",
        client=client,
        model="gpt-4.1",
        instructions="You are a helpful assistant.",
        tools=["web_search_preview"],  # Only tools without required params
    )
    await Console(agent.run_stream(task="Search for recent AI developments"))


asyncio.run(example())

配置内置工具的用法

import asyncio

from autogen_agentchat.ui import Console
from autogen_ext.agents.openai import OpenAIAgent
from openai import AsyncOpenAI


async def example_with_configs():
    client = AsyncOpenAI()
    # Configure tools with required and optional parameters
    tools = [
        # {
        #     "type": "file_search",
        #     "vector_store_ids": ["vs_abc123"],  # required
        #     "max_num_results": 10,  # optional
        # },
        # {
        #     "type": "computer_use_preview",
        #     "display_height": 1024,  # required
        #     "display_width": 1280,  # required
        #     "environment": "linux",  # required
        # },
        {
            "type": "code_interpreter",
            "container": {"type": "auto"},  # required
        },
        # {
        #     "type": "mcp",
        #     "server_label": "my-mcp-server",  # required
        #     "server_url": "https://:3000",  # required
        # },
        {
            "type": "web_search_preview",
            "user_location": {  # optional - structured location
                "type": "approximate",  # required: "approximate" or "exact"
                "country": "US",  # optional
                "region": "CA",  # optional
                "city": "San Francisco",  # optional
            },
            "search_context_size": "low",  # optional
        },
        # "image_generation",  # Simple tools can still use string format
    ]

    agent = OpenAIAgent(
        name="ConfiguredAgent",
        description="An agent with configured tools",
        client=client,
        model="gpt-4.1",
        instructions="You are a helpful assistant with specialized tools.",
        tools=tools,  # type: ignore
    )
    await Console(agent.run_stream(task="Search for recent AI developments"))


asyncio.run(example_with_configs())

注意: OpenAIAgent 不支持自定义工具。仅使用 Responses API 中的内置工具。

component_config_schema#: 别名 OpenAIAgentConfig

component_provider_override: ClassVar[str | None] = 'autogen_ext.agents.openai.OpenAIAgent'#: 覆盖组件的提供者字符串。这应该用于防止内部模块名称成为模块名称的一部分。

属性 produced_message_types: Sequence[Type[TextMessage] | Type[MultiModalMessage] | Type[StopMessage] | Type[ToolCallSummaryMessage] | Type[HandoffMessage]]#: 返回此代理可以生成的消息类型。

异步 on_messages(messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) → Response[来源]#: 处理传入消息并返回响应。

注意

代理是有状态的，传递给此方法的消息应该是自上次调用此方法以来的新消息。代理应在调用此方法之间保持其状态。例如，如果代理需要记住以前的消息才能响应当前消息，它应该将以前的消息存储在代理状态中。

异步 on_messages_stream(messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) → AsyncGenerator[Annotated[ToolCallRequestEvent | ToolCallExecutionEvent | MemoryQueryEvent | UserInputRequestedEvent | ModelClientStreamingChunkEvent | ThoughtEvent | SelectSpeakerEvent | CodeGenerationEvent | CodeExecutionEvent, FieldInfo(annotation=NoneType, required=True,discriminator='type')] | TextMessage | MultiModalMessage | StopMessage | ToolCallSummaryMessage | HandoffMessage | Response, None][来源]#: 处理传入消息并返回消息流，最后一个项目是响应。BaseChatAgent 中的基本实现只是调用 on_messages() 并生成响应中的消息。

注意

代理是有状态的，传递给此方法的消息应该是自上次调用此方法以来的新消息。代理应在调用此方法之间保持其状态。例如，如果代理需要记住以前的消息才能响应当前消息，它应该将以前的消息存储在代理状态中。

异步 on_reset(cancellation_token: CancellationToken) → None[来源]#: 将代理重置为其初始化状态。

异步 save_state() → Mapping[str, Any][来源]#: 导出状态。无状态代理的默认实现。

异步 load_state(state: Mapping[str, Any]) → None[来源]#: 从保存的状态恢复代理。无状态代理的默认实现。

to_config() → OpenAIAgentConfig[来源]#: 私有 _to_config 方法的公共包装器。

类方法 from_config(config: OpenAIAgentConfig) → OpenAIAgent[来源]#: 私有 _from_config 类方法的公共包装器。

属性 tools: list[Any]#: 公共访问代理的工具。

属性 model: str#: 公共访问代理的模型。

基类：BaseChatAgent

一个使用 Assistant API 生成响应的代理实现。

安装

pip install "autogen-ext[openai]"  # For OpenAI Assistant
# pip install "autogen-ext[openai,azure]"  # For Azure OpenAI Assistant

该代理利用 Assistant API 创建具有以下功能的 AI 助手：

代码解释和执行
文件处理和搜索
自定义函数调用
多轮对话

代理维护一个对话线程，并且可以使用各种工具，包括

代码解释器：用于执行代码和处理文件
文件搜索：用于搜索上传的文档
自定义函数：用于通过用户定义的工具扩展功能

主要特点

支持多种文件格式，包括代码、文档、图像
每个助手最多可以处理 128 个工具
在线程中维护对话上下文
支持为代码解释器和搜索上传文件
向量存储集成，实现高效的文件搜索
自动文件解析和嵌入

您可以通过提供 thread_id 或 assistant_id 参数来使用现有线程或助手。

示例

使用助手分析 CSV 文件中的数据

from openai import AsyncOpenAI
from autogen_core import CancellationToken
import asyncio
from autogen_ext.agents.openai import OpenAIAssistantAgent
from autogen_agentchat.messages import TextMessage


async def example():
    cancellation_token = CancellationToken()

    # Create an OpenAI client
    client = AsyncOpenAI(api_key="your-api-key", base_url="your-base-url")

    # Create an assistant with code interpreter
    assistant = OpenAIAssistantAgent(
        name="PythonHelper",
        description="Helps with Python programming",
        client=client,
        model="gpt-4",
        instructions="You are a helpful Python programming assistant.",
        tools=["code_interpreter"],
    )

    # Upload files for the assistant to use
    await assistant.on_upload_for_code_interpreter("data.csv", cancellation_token)

    # Get response from the assistant
    response = await assistant.on_messages(
        [TextMessage(source="user", content="Analyze the data in data.csv")], cancellation_token
    )

    print(response)

    # Clean up resources
    await assistant.delete_uploaded_files(cancellation_token)
    await assistant.delete_assistant(cancellation_token)


asyncio.run(example())

使用带有 AAD 身份验证的 Azure OpenAI Assistant

from openai import AsyncAzureOpenAI
import asyncio
from azure.identity import DefaultAzureCredential, get_bearer_token_provider
from autogen_core import CancellationToken
from autogen_ext.agents.openai import OpenAIAssistantAgent
from autogen_agentchat.messages import TextMessage


async def example():
    cancellation_token = CancellationToken()

    # Create an Azure OpenAI client
    token_provider = get_bearer_token_provider(DefaultAzureCredential())
    client = AsyncAzureOpenAI(
        azure_deployment="YOUR_AZURE_DEPLOYMENT",
        api_version="YOUR_API_VERSION",
        azure_endpoint="YOUR_AZURE_ENDPOINT",
        azure_ad_token_provider=token_provider,
    )

    # Create an assistant with code interpreter
    assistant = OpenAIAssistantAgent(
        name="PythonHelper",
        description="Helps with Python programming",
        client=client,
        model="gpt-4o",
        instructions="You are a helpful Python programming assistant.",
        tools=["code_interpreter"],
    )

    # Get response from the assistant
    response = await assistant.on_messages([TextMessage(source="user", content="Hello.")], cancellation_token)

    print(response)

    # Clean up resources
    await assistant.delete_assistant(cancellation_token)


asyncio.run(example())

参数:

name (str) – 助手的名称
description (str) – 助手目的的描述
client (AsyncOpenAI | AsyncAzureOpenAI) – OpenAI 客户端或 Azure OpenAI 客户端实例
model (str) – 要使用的模型（例如“gpt-4”）
instructions (str) – 助手的系统指令
tools (Optional[Iterable[Union[Literal["code_interpreter", "file_search"], Tool | Callable[..., Any] | Callable[..., Awaitable[Any]]]]]) – 助手可以使用的工具
assistant_id (Optional[str]) – 要使用的现有助手的 ID
thread_id (Optional[str]) – 要使用的现有线程的 ID
metadata (Optional[Dict[str, str]]) – 助手的额外元数据。
response_format (Optional[AssistantResponseFormatOptionParam]) – 响应格式设置
temperature (Optional[float]) – 响应生成的温度
tool_resources (Optional[ToolResources]) – 额外的工具配置
top_p (Optional[float]) – Top p 采样参数

属性 produced_message_types: Sequence[type[BaseChatMessage]]#: 助手代理生成的消息类型。

属性 threads: AsyncThreads#

属性 runs: AsyncRuns#

属性 messages: AsyncMessages#

异步 on_messages(messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) → Response[来源]#: 处理传入消息并返回响应。

异步 on_messages_stream(messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) → AsyncGenerator[BaseAgentEvent | BaseChatMessage | Response, None][来源]#: 处理传入消息并返回响应。

异步 handle_incoming_message(message: BaseChatMessage, cancellation_token: CancellationToken) → None[来源]#: 通过将常规文本消息添加到线程来处理它们。

异步 on_reset(cancellation_token: CancellationToken) → None[来源]#: 通过删除自初始化以来的新消息和运行来处理重置命令。

异步 on_upload_for_code_interpreter(file_paths: str | Iterable[str], cancellation_token: CancellationToken) → None[来源]#: 处理代码解释器的文件上传。

异步 on_upload_for_file_search(file_paths: str | Iterable[str], cancellation_token: CancellationToken) → None[来源]#: 处理文件搜索的文件上传。

异步 delete_uploaded_files(cancellation_token: CancellationToken) → None[来源]#: 删除此代理实例上传的所有文件。

异步 delete_assistant(cancellation_token: CancellationToken) → None[来源]#: 如果助手是由此实例创建的，则删除助手。

异步 delete_vector_store(cancellation_token: CancellationToken) → None[来源]#: 如果向量存储是由此实例创建的，则删除向量存储。

异步 save_state() → Mapping[str, Any][来源]#: 导出状态。无状态代理的默认实现。

异步 load_state(state: Mapping[str, Any]) → None[来源]#: 从保存的状态恢复代理。无状态代理的默认实现。