如何把 OpenAI Assistants 封装为 AutoGen Core 智能体并处理流式输出？

OpenAI Assistants 是运行在服务端的 API：你通过 assistant_id 和 thread_id 引用一个已有状态的助手，对话记忆保存在 OpenAI 侧的 thread 中。如果你希望这个助手像 AutoGen 框架里的普通智能体一样参与消息传递——接收 TextMessage、返回响应、按需上传文件或重置记忆——就需要把它封装成一个 AutoGen Core 的 RoutedAgent，并通过 OpenAI 客户端的 assistant event handler 拿到流式输出。

AutoGen 的 cookbook 提供了完整的参考实现 OpenAI Assistant Agent，本文按该实现拆解一条可执行路径：定义消息协议 → 封装智能体类 → 实现流式事件处理器 → 注册到 Runtime → 发送消息并验证流式输出。

前提条件

按 Core 安装指南 完成环境准备：Python 3.10 或更高版本，并用 pip 安装 autogen-core：

pip install "autogen-core"

cookbook 代码额外依赖 openai（提供 AsyncClient 和 AsyncAssistantEventHandler）、aiofiles（智能体上传文件时异步读取本地文件）、requests（示例中下载演示数据），以及一个可用的 OpenAI API key（openai.AsyncClient() 按默认凭证机制读取，需自行配置）。

注意：cookbook 中的示例代码使用顶层 await，即在 Jupyter 一类异步上下文中直接运行；如果用普通 Python 脚本跑，需要把消息发送部分放进异步入口（如 asyncio.run）。

定义消息协议

AutoGen Core 中智能体之间的通信基于你自定义的消息类型。参考实现定义了 4 种消息（见 cookbook）：

消息类型	用途	字段
TextMessage	与智能体对话	content（消息内容）、source（发送方标识）
Reset	重置助手记忆（清空 thread 中的消息）	无
UploadForCodeInterpreter	上传数据文件给 code interpreter	file_path
UploadForFileSearch	上传文档给 file search	file_path、vector_store_id

from dataclasses import dataclass


@dataclass
class TextMessage:
    content: str
    source: str


@dataclass
class Reset:
    pass


@dataclass
class UploadForCodeInterpreter:
    file_path: str


@dataclass
class UploadForFileSearch:
    file_path: str
    vector_store_id: str

封装智能体类：OpenAIAssistantAgent

智能体类继承 AutoGen Core 的 RoutedAgent，用 @message_handler 为每种消息类型注册处理函数。构造参数有 5 个：description（智能体描述）、client（openai.AsyncClient 实例）、assistant_id、thread_id，以及 assistant_event_handler_factory——一个创建 AsyncAssistantEventHandler 的工厂函数，用于产生流式输出；按代码文档说明，提供了该工厂则走 streaming 模式，不提供则用阻塞模式生成响应。

完整的参考实现如下（与 cookbook 一致，四个 handler 分别对应上面四种消息）：

import asyncio
import os
from typing import Any, Callable, List

import aiofiles
from autogen_core import AgentId, MessageContext, RoutedAgent, message_handler
from openai import AsyncAssistantEventHandler, AsyncClient
from openai.types.beta.thread import ToolResources, ToolResourcesFileSearch


class OpenAIAssistantAgent(RoutedAgent):
    """An agent implementation that uses the OpenAI Assistant API to generate
    responses.

    Args:
        description (str): The description of the agent.
        client (openai.AsyncClient): The client to use for the OpenAI API.
        assistant_id (str): The assistant ID to use for the OpenAI API.
        thread_id (str): The thread ID to use for the OpenAI API.
        assistant_event_handler_factory (Callable[[], AsyncAssistantEventHandler], optional):
            A factory function to create an async assistant event handler. Defaults to None.
            If provided, the agent will use the streaming mode with the event handler.
            If not provided, the agent will use the blocking mode to generate responses.
    """

    def __init__(
        self,
        description: str,
        client: AsyncClient,
        assistant_id: str,
        thread_id: str,
        assistant_event_handler_factory: Callable[[], AsyncAssistantEventHandler],
    ) -> None:
        super().__init__(description)
        self._client = client
        self._assistant_id = assistant_id
        self._thread_id = thread_id
        self._assistant_event_handler_factory = assistant_event_handler_factory

    @message_handler
    async def handle_message(self, message: TextMessage, ctx: MessageContext) -> TextMessage:
        """Handle a message. This method adds the message to the thread and publishes a response."""
        # Save the message to the thread.
        await ctx.cancellation_token.link_future(
            asyncio.ensure_future(
                self._client.beta.threads.messages.create(
                    thread_id=self._thread_id,
                    content=message.content,
                    role="user",
                    metadata={"sender": message.source},
                )
            )
        )
        # Generate a response.
        async with self._client.beta.threads.runs.stream(
            thread_id=self._thread_id,
            assistant_id=self._assistant_id,
            event_handler=self._assistant_event_handler_factory(),
        ) as stream:
            await ctx.cancellation_token.link_future(asyncio.ensure_future(stream.until_done()))

        # Get the last message.
        messages = await ctx.cancellation_token.link_future(
            asyncio.ensure_future(self._client.beta.threads.messages.list(self._thread_id, order="desc", limit=1))
        )
        last_message_content = messages.data[0].content

        # Get the text content from the last message.
        text_content = [content for content in last_message_content if content.type == "text"]
        if not text_content:
            raise ValueError(f"Expected text content in the last message: {last_message_content}")

        return TextMessage(content=text_content[0].text.value, source=self.metadata["type"])

    @message_handler()
    async def on_reset(self, message: Reset, ctx: MessageContext) -> None:
        """Handle a reset message. This method deletes all messages in the thread."""
        # Get all messages in this thread.
        all_msgs: List[str] = []
        while True:
            if not all_msgs:
                msgs = await ctx.cancellation_token.link_future(
                    asyncio.ensure_future(self._client.beta.threads.messages.list(self._thread_id))
                )
            else:
                msgs = await ctx.cancellation_token.link_future(
                    asyncio.ensure_future(self._client.beta.threads.messages.list(self._thread_id, after=all_msgs[-1]))
                )
            for msg in msgs.data:
                all_msgs.append(msg.id)
            if not msgs.has_next_page():
                break
        # Delete all the messages.
        for msg_id in all_msgs:
            status = await ctx.cancellation_token.link_future(
                asyncio.ensure_future(
                    self._client.beta.threads.messages.delete(message_id=msg_id, thread_id=self._thread_id)
                )
            )
            assert status.deleted is True

    @message_handler()
    async def on_upload_for_code_interpreter(self, message: UploadForCodeInterpreter, ctx: MessageContext) -> None:
        """Handle an upload for code interpreter. This method uploads a file and updates the thread with the file."""
        # Get the file content.
        async with aiofiles.open(message.file_path, mode="rb") as f:
            file_content = await ctx.cancellation_token.link_future(asyncio.ensure_future(f.read()))
        file_name = os.path.basename(message.file_path)
        # Upload the file.
        file = await ctx.cancellation_token.link_future(
            asyncio.ensure_future(self._client.files.create(file=(file_name, file_content), purpose="assistants"))
        )
        # Get existing file ids from tool resources.
        thread = await ctx.cancellation_token.link_future(
            asyncio.ensure_future(self._client.beta.threads.retrieve(thread_id=self._thread_id))
        )
        tool_resources: ToolResources = thread.tool_resources if thread.tool_resources else ToolResources()
        assert tool_resources.code_interpreter is not None
        if tool_resources.code_interpreter.file_ids:
            file_ids = tool_resources.code_interpreter.file_ids
        else:
            file_ids = [file.id]
        # Update thread with new file.
        await ctx.cancellation_token.link_future(
            asyncio.ensure_future(
                self._client.beta.threads.update(
                    thread_id=self._thread_id,
                    tool_resources={
                        "code_interpreter": {"file_ids": file_ids},
                    },
                )
            )
        )

    @message_handler()
    async def on_upload_for_file_search(self, message: UploadForFileSearch, ctx: MessageContext) -> None:
        """Handle an upload for file search. This method uploads a file and updates the vector store."""
        # Get the file content.
        async with aiofiles.open(message.file_path, mode="rb") as file:
            file_content = await ctx.cancellation_token.link_future(asyncio.ensure_future(file.read()))
        file_name = os.path.basename(message.file_path)
        # Upload the file.
        await ctx.cancellation_token.link_future(
            asyncio.ensure_future(
                self._client.vector_stores.file_batches.upload_and_poll(
                    vector_store_id=message.vector_store_id,
                    files=[(file_name, file_content)],
                )
            )
        )

handle_message 的处理顺序是：把消息写入 thread → 用 threads.runs.stream 带事件处理器发起流式运行并等待完成 → 从 thread 中取最后一条消息，取出其中的 text 内容后封装成新的 TextMessage 返回。如果最后一条消息里没有 text 类型内容，会抛出 ValueError，这是运行时判断"助手本次没有产出文本"的唯一依据。

这个类只是 OpenAI Assistant API 的一个薄封装，cookbook 指出可以通过扩展消息协议（例如多模态消息）增加更多能力。

用事件处理器处理流式输出

流式输出靠 AsyncAssistantEventHandler 的回调实现。参考实现覆盖了 6 个回调，各自对应一类 Assistant 事件：

• on_text_delta：文本增量到达时打印 delta.value，这是流式文本输出的核心；
• on_run_step_created / on_run_step_delta / on_run_step_done：跟踪 run step，识别 code_interpreter 工具调用，在代码生成、代码增量和执行阶段打印分隔标记；
• on_message_created / on_message_done：在消息创建时打印分隔线，在消息完成时处理 file search 的引用标注（把 file_citation 解析为文件名并打印引用列表）。

from openai import AsyncAssistantEventHandler, AsyncClient
from openai.types.beta.threads import Message, Text, TextDelta
from openai.types.beta.threads.runs import RunStep, RunStepDelta
from typing_extensions import override


class EventHandler(AsyncAssistantEventHandler):
    @override
    async def on_text_delta(self, delta: TextDelta, snapshot: Text) -> None:
        print(delta.value, end="", flush=True)

    @override
    async def on_run_step_created(self, run_step: RunStep) -> None:
        details = run_step.step_details
        if details.type == "tool_calls":
            for tool in details.tool_calls:
                if tool.type == "code_interpreter":
                    print("\nGenerating code to interpret:\n\n```python")

    @override
    async def on_run_step_done(self, run_step: RunStep) -> None:
        details = run_step.step_details
        if details.type == "tool_calls":
            for tool in details.tool_calls:
                if tool.type == "code_interpreter":
                    print("\n```\nExecuting code...")

    @override
    async def on_run_step_delta(self, delta: RunStepDelta, snapshot: RunStep) -> None:
        details = delta.step_details
        if details is not None and details.type == "tool_calls":
            for tool in details.tool_calls or []:
                if tool.type == "code_interpreter" and tool.code_interpreter and tool.code_interpreter.input:
                    print(tool.code_interpreter.input, end="", flush=True)

    @override
    async def on_message_created(self, message: Message) -> None:
        print(f"{'-'*80}\nAssistant:\n")

    @override
    async def on_message_done(self, message: Message) -> None:
        # print a citation to the file searched
        if not message.content:
            return
        content = message.content[0]
        if not content.type == "text":
            return
        text_content = content.text
        annotations = text_content.annotations
        citations: List[str] = []
        for index, annotation in enumerate(annotations):
            text_content.value = text_content.value.replace(annotation.text, f"[{index}]")
            if file_citation := getattr(annotation, "file_citation", None):
                client = AsyncClient()
                cited_file = await client.files.retrieve(file_citation.file_id)
                citations.append(f"[{index}] {cited_file.filename}")
        if citations:
            print("\n".join(citations))

这些回调决定了终端里"看到什么"：文本增量实时打印，工具调用阶段打印代码块边界，file search 回答完成时打印引用文件。如果你的场景只需要文本流，on_text_delta 是必选回调，其余可按需保留。

创建 Assistant、Thread 与向量库

智能体本身不创建服务端资源，需要用 openai 客户端先把 assistant、thread 和向量库建好，再把 id 交给智能体。cookbook 的做法是创建带 code_interpreter 和 file_search 两个工具的 assistant（模型为 gpt-4o-mini），再建一个向量库并把它挂到 thread 的 tool_resources 上：

import openai

# Create an assistant with code interpreter and file search tools.
oai_assistant = openai.beta.assistants.create(
    model="gpt-4o-mini",
    description="An AI assistant that helps with everyday tasks.",
    instructions="Help the user with their task.",
    tools=[{"type": "code_interpreter"}, {"type": "file_search"}],
)

# Create a vector store to be used for file search.
vector_store = openai.vector_stores.create()

# Create a thread which is used as the memory for the assistant.
thread = openai.beta.threads.create(
    tool_resources={"file_search": {"vector_store_ids": [vector_store.id]}},
)

thread 即助手记忆的载体，保存在服务端，智能体只持有 thread.id 作为引用。

注册到 Runtime 并发送消息

创建 SingleThreadedAgentRuntime，把智能体的工厂函数以类型名 "assistant" 注册进去，然后用 AgentId("assistant", "default") 定位实例：

from autogen_core import SingleThreadedAgentRuntime

runtime = SingleThreadedAgentRuntime()
await OpenAIAssistantAgent.register(
    runtime,
    "assistant",
    lambda: OpenAIAssistantAgent(
        description="OpenAI Assistant Agent",
        client=openai.AsyncClient(),
        assistant_id=oai_assistant.id,
        thread_id=thread.id,
        assistant_event_handler_factory=lambda: EventHandler(),
    ),
)
agent = AgentId("assistant", "default")

发送前先打开 autogen_core 的 DEBUG 日志，可以看到消息在 Runtime 内部的流转：

import logging

logging.basicConfig(level=logging.WARNING)
logging.getLogger("autogen_core").setLevel(logging.DEBUG)

然后发送一条 TextMessage 并等待 Runtime 空闲：

runtime.start()
await runtime.send_message(TextMessage(content="Hello, how are you today!", source="user"), agent)
await runtime.stop_when_idle()

验证流式输出是否生效

一次成功的交互会同时出现三类输出（以下均为 cookbook 中的文档示例输出，实际内容会随模型响应不同）：

1. stderr 上的 Runtime 日志，说明消息已被路由到 handler：

INFO:autogen_core:Sending message of type TextMessage to assistant: {'content': 'Hello, how are you today!', 'source': 'user'}
INFO:autogen_core:Calling message handler for assistant:default with message type TextMessage sent by Unknown

2. stdout 上的流式内容，由事件处理器打印，先出现 Assistant: 分隔线，随后是逐增量打印的回复文本：

--------------------------------------------------------------------------------
Assistant:

Hello! I'm here and ready to assist you. How can I help you today?

3. stderr 上的响应解析日志，说明 handler 已返回 TextMessage：

INFO:autogen_core:Resolving response with message type TextMessage for recipient None from assistant: {'content': "Hello! I'm here and ready to assist you. How can I help you today?", 'source': 'assistant'}

如果助手动用了 code interpreter，流里还会插入 Generating code to interpret: 与 Executing code... 的边界标记（例如问数学题时，文档示例显示生成了 result = 1332322 * 123212 的代码并返回乘积结果）。

可选能力：文件上传与记忆重置

• Code interpreter 文件：把本地文件路径包成 UploadForCodeInterpreter(file_path=...) 发给智能体，handler 会把文件以 purpose="assistants" 上传到 OpenAI 并写入 thread 的 tool_resources.code_interpreter.file_ids；之后用普通 TextMessage 提问即可基于该文件作答。
• File search 文档：UploadForFileSearch(file_path=..., vector_store_id=vector_store.id) 会把文件经 vector_stores.file_batches.upload_and_poll 上传到指定向量库；提问后 on_message_done 会解析引用并打印形如 [0] third_anglo_afghan_war.html 的引用文件名（文档示例）。
• 重置记忆：发送 Reset()，handler 会分页列出 thread 内全部消息并逐条删除，可用于开启一段全新对话；文档示例在切换 file search 场景前就是这样重置的。

限制与注意点

• 记忆完全在服务端：thread 由 OpenAI 侧保存，重置或清理只能走 API 删除 thread 内消息，本地没有持久化状态可操作。
• handle_message 假定运行结束后 thread 最后一条消息包含 text 内容，否则抛 ValueError；如果你的助手配置可能产生非文本结果，需要自行扩展该 handler。
• 流式行为依赖 assistant_event_handler_factory；参考实现中该参数为必传项，不提供时按代码文档说明走阻塞模式。
• 示例代码是异步风格且使用顶层 await，直接在同步脚本中运行会失败，需放入 Jupyter 或 asyncio.run 入口。

完整可运行代码与更多交互示例见 openai-assistant-agent cookbook，RoutedAgent 与 message_handler 的机制可进一步参阅 autogen-core 源码。