如何把 OpenAI Assistants 封装为 AutoGen Core 智能体并处理流式输出【免费下载链接】autogenA programming framework for agentic AI项目地址: https://gitcode.com/GitHub_Trending/au/autogenOpenAI 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-corepip install autogen-corecookbook 代码额外依赖openai提供AsyncClient和AsyncAssistantEventHandler、aiofiles智能体上传文件时异步读取本地文件、requests示例中下载演示数据以及一个可用的 OpenAI API keyopenai.AsyncClient()按默认凭证机制读取需自行配置。注意cookbook 中的示例代码使用顶层await即在 Jupyter 一类异步上下文中直接运行如果用普通 Python 脚本跑需要把消息发送部分放进异步入口如asyncio.run。定义消息协议AutoGen Core 中智能体之间的通信基于你自定义的消息类型。参考实现定义了 4 种消息见 cookbook消息类型用途字段TextMessage与智能体对话content消息内容、source发送方标识Reset重置助手记忆清空 thread 中的消息无UploadForCodeInterpreter上传数据文件给 code interpreterfile_pathUploadForFileSearch上传文档给 file searchfile_path、vector_store_idfrom 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智能体描述、clientopenai.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_idself._thread_id, contentmessage.content, roleuser, metadata{sender: message.source}, ) ) ) # Generate a response. async with self._client.beta.threads.runs.stream( thread_idself._thread_id, assistant_idself._assistant_id, event_handlerself._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, orderdesc, limit1)) ) 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(fExpected text content in the last message: {last_message_content}) return TextMessage(contenttext_content[0].text.value, sourceself.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, afterall_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_idmsg_id, thread_idself._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, moderb) 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), purposeassistants)) ) # Get existing file ids from tool resources. thread await ctx.cancellation_token.link_future( asyncio.ensure_future(self._client.beta.threads.retrieve(thread_idself._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_idself._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, moderb) 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_idmessage.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, flushTrue) 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\npython) 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, flushTrue) 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( modelgpt-4o-mini, descriptionAn AI assistant that helps with everyday tasks., instructionsHelp 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( descriptionOpenAI Assistant Agent, clientopenai.AsyncClient(), assistant_idoai_assistant.id, thread_idthread.id, assistant_event_handler_factorylambda: EventHandler(), ), ) agent AgentId(assistant, default)发送前先打开autogen_core的 DEBUG 日志可以看到消息在 Runtime 内部的流转import logging logging.basicConfig(levellogging.WARNING) logging.getLogger(autogen_core).setLevel(logging.DEBUG)然后发送一条TextMessage并等待 Runtime 空闲runtime.start() await runtime.send_message(TextMessage(contentHello, how are you today!, sourceuser), agent) await runtime.stop_when_idle()验证流式输出是否生效一次成功的交互会同时出现三类输出以下均为 cookbook 中的文档示例输出实际内容会随模型响应不同stderr 上的 Runtime 日志说明消息已被路由到 handlerINFO: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 Unknownstdout 上的流式内容由事件处理器打印先出现Assistant:分隔线随后是逐增量打印的回复文本-------------------------------------------------------------------------------- Assistant: Hello! Im here and ready to assist you. How can I help you today?stderr 上的响应解析日志说明 handler 已返回TextMessageINFO:autogen_core:Resolving response with message type TextMessage for recipient None from assistant: {content: Hello! Im 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 会把文件以purposeassistants上传到 OpenAI 并写入 thread 的tool_resources.code_interpreter.file_ids之后用普通TextMessage提问即可基于该文件作答。File search 文档UploadForFileSearch(file_path..., vector_store_idvector_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 cookbookRoutedAgent与message_handler的机制可进一步参阅 autogen-core 源码。【免费下载链接】autogenA programming framework for agentic AI项目地址: https://gitcode.com/GitHub_Trending/au/autogen创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考