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)[source]#

基类: BaseChatAgent

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

安装

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

该代理利用 Assistant API 创建具有以下功能的人工智能助手

  • 代码解释和执行

  • 文件处理和搜索

  • 自定义函数调用

  • 多轮对话

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

  • 代码解释器：用于执行代码和处理文件

  • 文件搜索：用于搜索上传的文档

  • 自定义函数：用于通过用户定义的工具扩展功能

主要特性

  • 支持多种文件格式，包括代码、文档、图像

  • 每个助手最多可以处理 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="Python Helper",
        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="Python Helper",
        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 采样参数

async delete_assistant(cancellation_token: CancellationToken) → None[source]#

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

async delete_uploaded_files(cancellation_token: CancellationToken) → None[source]#

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

async delete_vector_store(cancellation_token: CancellationToken) → None[source]#

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

async handle_incoming_message(message: BaseChatMessage, cancellation_token: CancellationToken) → None[source]#

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

async load_state(state: Mapping[str, Any]) → None[source]#

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

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

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

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

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

async on_reset(cancellation_token: CancellationToken) → None[source]#

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

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

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

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

property produced_message_types: Sequence[type[BaseChatMessage]]#

助手代理生成的邮件类型。

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

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

property threads: AsyncThreads#