autogen_ext.agents.openai#

class OpenAIAssistantAgent(name: str, description: str, client: AsyncOpenAI | AsyncAzureOpenAI, model: str, instructions: str, tools: Iterable[Literal['code_interpreter', 'file_search'] | Tool | Callable[[...], Any] | Callable[[...], Awaitable[Any]]] | None = None, assistant_id: str | None = None, thread_id: str | None = None, metadata: Dict[str, str] | None = None, response_format: Literal['auto'] | ResponseFormatText | ResponseFormatJSONObject | ResponseFormatJSONSchema | None = None, temperature: float | None = None, tool_resources: ToolResources | None = None, top_p: float | None = None)[源代码]#

基类：BaseChatAgent

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

安装:

pip install "autogen-ext[openai]"
# pip install "autogen-ext[openai,azure]"  # 用于 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()

    # 创建 OpenAI 客户端
    client = AsyncOpenAI(api_key="your-api-key", base_url="your-base-url")

    # 创建带代码解释器的助手
    assistant = OpenAIAssistantAgent(
        name="Python Helper",
        description="Helps with Python programming",
        client=client,
        model="gpt-4",
        instructions="You are a helpful Python programming assistant.",
        tools=["code_interpreter"],
    )

    # 上传供助手使用的文件
    await assistant.on_upload_for_code_interpreter("data.csv", cancellation_token)

    # 获取助手响应
    response = await assistant.on_messages(
        [TextMessage(source="user", content="Analyze the data in data.csv")], cancellation_token
    )

    print(response)

    # 清理资源
    await assistant.delete_uploaded_files(cancellation_token)
    await assistant.delete_assistant(cancellation_token)


asyncio.run(example())

使用 Azure OpenAI Assistant 与 AAD 认证:

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()

    # 创建 Azure OpenAI 客户端
    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,
    )

    # 创建带代码解释器的助手
    assistant = OpenAIAssistantAgent(
        name="Python Helper",
        description="Helps with Python programming",
        client=client,
        model="gpt-4o",
        instructions="You are a helpful Python programming assistant.",
        tools=["code_interpreter"],
    )

    # 获取助手响应
    response = await assistant.on_messages([TextMessage(source="user", content="Hello.")], cancellation_token)

    print(response)

    # 清理资源
    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 采样参数

async delete_assistant(cancellation_token: CancellationToken) → None[源代码]#

如果助手是由此实例创建的，则删除该助手。

async delete_uploaded_files(cancellation_token: CancellationToken) → None[源代码]#

删除由此代理实例上传的所有文件。

async delete_vector_store(cancellation_token: CancellationToken) → None[源代码]#

如果向量存储是由该实例创建的，则删除它。

async handle_incoming_message(message: BaseChatMessage, cancellation_token: CancellationToken) → None[源代码]#

通过将常规文本消息添加到线程中来处理它们。

async load_state(state: Mapping[str, Any]) → None[源代码]#

从保存的状态恢复代理。无状态代理的默认实现。

property messages: AsyncMessages#
async on_messages(messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) → Response[源代码]#

处理传入的消息并返回响应。

async on_messages_stream(messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) → AsyncGenerator[BaseAgentEvent | BaseChatMessage | Response, None][源代码]#

处理传入的消息并返回响应。

async on_reset(cancellation_token: CancellationToken) → None[源代码]#

通过删除初始化后的新消息和运行记录来处理重置命令。

async on_upload_for_code_interpreter(file_paths: str | Iterable[str], cancellation_token: CancellationToken) → None[源代码]#

为代码解释器处理文件上传。

处理文件搜索的文件上传。

property produced_message_types: Sequence[type[BaseChatMessage]]#

助手代理产生的消息类型。

property runs: AsyncRuns#
async save_state() → Mapping[str, Any][源代码]#

导出状态。无状态代理的默认实现。

property threads: AsyncThreads#