OpenAI 助手代理#
OpenAI 助手和Azure OpenAI 助手是用于构建代理的服务器端 API。它们可以用于在 AutoGen 中构建代理。本指南演示如何使用 OpenAI 助手创建一个可以运行代码和对文档进行问答的代理。
消息协议#
首先，我们需要为由 OpenAI 助手支持的代理指定消息协议。消息协议定义了代理处理和发布的消息结构。为了说明，我们定义了一个包含 4 种消息类型的简单消息协议：Message、Reset、UploadForCodeInterpreter 和 UploadForFileSearch。
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
TextMessage 消息类型用于与代理通信。它有一个包含消息内容的 content 字段，以及一个用于发送者的 source 字段。Reset 消息类型是一个控制消息，用于重置代理的内存。它没有字段。当我们需要与代理开始新的对话时，这很有用。
UploadForCodeInterpreter 消息类型用于上传数据文件供代码解释器使用，而 UploadForFileSearch 消息类型用于上传文档以进行文件搜索。两种消息类型都有一个 file_path 字段，其中包含要上传文件的本地路径。
定义代理#
接下来，我们定义代理类。代理类构造函数具有以下参数：description、client、assistant_id、thread_id 和 assistant_event_handler_factory。client 参数是 OpenAI 异步客户端对象，而 assistant_event_handler_factory 用于创建助手事件处理程序来处理 OpenAI 助手事件。这可以用于从助手创建流式输出。
代理类具有以下消息处理程序
handle_message：处理TextMessage消息类型，并发送助手的响应。handle_reset：处理Reset消息类型，并重置助手代理的内存。handle_upload_for_code_interpreter：处理UploadForCodeInterpreter消息类型，并将文件上传到代码解释器。handle_upload_for_file_search：处理UploadForFileSearch消息类型，并将文档上传到文件搜索。
助手的内存存储在一个线程中，该线程保存在服务器端。该线程由 thread_id 参数引用。
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)],
)
)
)
代理类是 OpenAI Assistant API 的一个薄包装器，用于实现消息协议。可以通过扩展消息协议来添加更多功能，例如多模态消息处理。
助手事件处理程序#
助手事件处理程序提供用于处理助手 API 特定事件的回调。这对于处理助手的流式输出和进一步的用户界面集成非常有用。
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))
使用代理#
首先，我们需要使用 openai 客户端来创建实际的助手、线程和向量存储。我们的 AutoGen 代理将使用这些。
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]}},
)
然后，我们创建一个运行时，并为该代理注册一个代理工厂函数。
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")
让我们打开日志记录，看看幕后发生了什么。
import logging
logging.basicConfig(level=logging.WARNING)
logging.getLogger("autogen_core").setLevel(logging.DEBUG)
让我们向代理发送一条问候消息，看看流式返回的响应。
runtime.start()
await runtime.send_message(TextMessage(content="Hello, how are you today!", source="user"), agent)
await runtime.stop_when_idle()
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
--------------------------------------------------------------------------------
Assistant:
Hello! I'm here and ready to assist you. How can I help you today?
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'}
带代码解释器的助手#
让我们向代理提出一些数学问题，看看它如何使用代码解释器来回答问题。
runtime.start()
await runtime.send_message(TextMessage(content="What is 1332322 x 123212?", source="user"), agent)
await runtime.stop_when_idle()
INFO:autogen_core:Sending message of type TextMessage to assistant: {'content': 'What is 1332322 x 123212?', 'source': 'user'}
INFO:autogen_core:Calling message handler for assistant:default with message type TextMessage sent by Unknown
# Calculating the product of 1332322 and 123212
result = 1332322 * 123212
result
```
Executing code...
--------------------------------------------------------------------------------
Assistant:
The product of 1,332,322 and 123,212 is 164,158,058,264.
INFO:autogen_core:Resolving response with message type TextMessage for recipient None from assistant: {'content': 'The product of 1,332,322 and 123,212 is 164,158,058,264.', 'source': 'assistant'}
让我们从西雅图开放数据门户获取一些数据。我们将使用西雅图市工资数据。让我们先下载它。
import requests
response = requests.get("https://data.seattle.gov/resource/2khk-5ukd.csv")
with open("seattle_city_wages.csv", "wb") as file:
file.write(response.content)
让我们使用 UploadForCodeInterpreter 消息将文件发送给代理。
runtime.start()
await runtime.send_message(UploadForCodeInterpreter(file_path="seattle_city_wages.csv"), agent)
await runtime.stop_when_idle()
INFO:autogen_core:Sending message of type UploadForCodeInterpreter to assistant: {'file_path': 'seattle_city_wages.csv'}
INFO:autogen_core:Calling message handler for assistant:default with message type UploadForCodeInterpreter sent by Unknown
INFO:autogen_core:Resolving response with message type NoneType for recipient None from assistant: None
我们现在可以向代理询问有关数据的一些问题。
runtime.start()
await runtime.send_message(TextMessage(content="Take a look at the uploaded CSV file.", source="user"), agent)
await runtime.stop_when_idle()
INFO:autogen_core:Sending message of type TextMessage to assistant: {'content': 'Take a look at the uploaded CSV file.', 'source': 'user'}
INFO:autogen_core:Calling message handler for assistant:default with message type TextMessage sent by Unknown
import pandas as pd
# Load the uploaded CSV file to examine its contents
file_path = '/mnt/data/file-oEvRiyGyHc2jZViKyDqL8aoh'
csv_data = pd.read_csv(file_path)
# Display the first few rows of the dataframe to understand its structure
csv_data.head()
```
Executing code...
--------------------------------------------------------------------------------
Assistant:
The uploaded CSV file contains the following columns:
1. **department**: The department in which the individual works.
2. **last_name**: The last name of the employee.
3. **first_name**: The first name of the employee.
4. **job_title**: The job title of the employee.
5. **hourly_rate**: The hourly rate for the employee's position.
Here are the first few entries from the file:
| department | last_name | first_name | job_title | hourly_rate |
|--------------------------------|-----------|------------|------------------------------------|-------------|
| Police Department | Aagard | Lori | Pol Capt-Precinct | 112.70 |
| Police Department | Aakervik | Dag | Pol Ofcr-Detective | 75.61 |
| Seattle City Light | Aaltonen | Evan | Pwrline Clear Tree Trimmer | 53.06 |
| Seattle Public Utilities | Aar | Abdimallik | Civil Engrng Spec,Sr | 64.43 |
| Seattle Dept of Transportation | Abad | Abigail | Admin Spec II-BU | 37.40 |
If you need any specific analysis or information from this data, please let me know!
INFO:autogen_core:Resolving response with message type TextMessage for recipient None from assistant: {'content': "The uploaded CSV file contains the following columns:\n\n1. **department**: The department in which the individual works.\n2. **last_name**: The last name of the employee.\n3. **first_name**: The first name of the employee.\n4. **job_title**: The job title of the employee.\n5. **hourly_rate**: The hourly rate for the employee's position.\n\nHere are the first few entries from the file:\n\n| department | last_name | first_name | job_title | hourly_rate |\n|--------------------------------|-----------|------------|------------------------------------|-------------|\n| Police Department | Aagard | Lori | Pol Capt-Precinct | 112.70 |\n| Police Department | Aakervik | Dag | Pol Ofcr-Detective | 75.61 |\n| Seattle City Light | Aaltonen | Evan | Pwrline Clear Tree Trimmer | 53.06 |\n| Seattle Public Utilities | Aar | Abdimallik | Civil Engrng Spec,Sr | 64.43 |\n| Seattle Dept of Transportation | Abad | Abigail | Admin Spec II-BU | 37.40 |\n\nIf you need any specific analysis or information from this data, please let me know!", 'source': 'assistant'}
runtime.start()
await runtime.send_message(TextMessage(content="What are the top-10 salaries?", source="user"), agent)
await runtime.stop_when_idle()
INFO:autogen_core:Sending message of type TextMessage to assistant: {'content': 'What are the top-10 salaries?', 'source': 'user'}
INFO:autogen_core:Calling message handler for assistant:default with message type TextMessage sent by Unknown
# Sorting the data by hourly_rate in descending order and selecting the top 10 salaries
top_10_salaries = csv_data[['first_name', 'last_name', 'job_title', 'hourly_rate']].sort_values(by='hourly_rate', ascending=False).head(10)
top_10_salaries.reset_index(drop=True, inplace=True)
top_10_salaries
```
Executing code...
--------------------------------------------------------------------------------
Assistant:
Here are the top 10 salaries based on the hourly rates from the CSV file:
| First Name | Last Name | Job Title | Hourly Rate |
|------------|-----------|------------------------------------|-------------|
| Eric | Barden | Executive4 | 139.61 |
| Idris | Beauregard| Executive3 | 115.90 |
| Lori | Aagard | Pol Capt-Precinct | 112.70 |
| Krista | Bair | Pol Capt-Precinct | 108.74 |
| Amy | Bannister | Fire Chief, Dep Adm-80 Hrs | 104.07 |
| Ginger | Armbruster| Executive2 | 102.42 |
| William | Andersen | Executive2 | 102.42 |
| Valarie | Anderson | Executive2 | 102.42 |
| Paige | Alderete | Executive2 | 102.42 |
| Kathryn | Aisenberg | Executive2 | 100.65 |
If you need any further details or analysis, let me know!
INFO:autogen_core:Resolving response with message type TextMessage for recipient None from assistant: {'content': 'Here are the top 10 salaries based on the hourly rates from the CSV file:\n\n| First Name | Last Name | Job Title | Hourly Rate |\n|------------|-----------|------------------------------------|-------------|\n| Eric | Barden | Executive4 | 139.61 |\n| Idris | Beauregard| Executive3 | 115.90 |\n| Lori | Aagard | Pol Capt-Precinct | 112.70 |\n| Krista | Bair | Pol Capt-Precinct | 108.74 |\n| Amy | Bannister | Fire Chief, Dep Adm-80 Hrs | 104.07 |\n| Ginger | Armbruster| Executive2 | 102.42 |\n| William | Andersen | Executive2 | 102.42 |\n| Valarie | Anderson | Executive2 | 102.42 |\n| Paige | Alderete | Executive2 | 102.42 |\n| Kathryn | Aisenberg | Executive2 | 100.65 |\n\nIf you need any further details or analysis, let me know!', 'source': 'assistant'}
带文件搜索的助手#
让我们尝试文档问答功能。我们首先下载关于第三次英阿战争的维基百科页面。
response = requests.get("https://en.wikipedia.org/wiki/Third_Anglo-Afghan_War")
with open("third_anglo_afghan_war.html", "wb") as file:
file.write(response.content)
使用 UploadForFileSearch 消息将文件发送给代理。
runtime.start()
await runtime.send_message(
UploadForFileSearch(file_path="third_anglo_afghan_war.html", vector_store_id=vector_store.id), agent
)
await runtime.stop_when_idle()
INFO:autogen_core:Sending message of type UploadForFileSearch to assistant: {'file_path': 'third_anglo_afghan_war.html', 'vector_store_id': 'vs_h3xxPbJFnd1iZ9WdjsQwNdrp'}
INFO:autogen_core:Calling message handler for assistant:default with message type UploadForFileSearch sent by Unknown
INFO:autogen_core:Resolving response with message type NoneType for recipient None from assistant: None
让我们向代理询问有关文档的一些问题。在提问之前，我们重置代理内存以开始新的对话。
runtime.start()
await runtime.send_message(Reset(), agent)
await runtime.send_message(
TextMessage(
content="When and where was the treaty of Rawalpindi signed? Answer using the document provided.", source="user"
),
agent,
)
await runtime.stop_when_idle()
INFO:autogen_core:Sending message of type Reset to assistant: {}
INFO:autogen_core:Calling message handler for assistant:default with message type Reset sent by Unknown
INFO:autogen_core:Resolving response with message type NoneType for recipient None from assistant: None
INFO:autogen_core:Sending message of type TextMessage to assistant: {'content': 'When and where was the treaty of Rawalpindi signed? Answer using the document provided.', 'source': 'user'}
INFO:autogen_core:Calling message handler for assistant:default with message type TextMessage sent by Unknown
--------------------------------------------------------------------------------
Assistant:
The Treaty of Rawalpindi was signed on **8 August 1919**. The location of the signing was in **Rawalpindi**, which is in present-day Pakistan【6:0†source】.
INFO:autogen_core:Resolving response with message type TextMessage for recipient None from assistant: {'content': 'The Treaty of Rawalpindi was signed on **8 August 1919**. The location of the signing was in **Rawalpindi**, which is in present-day Pakistan【6:0†source】.', 'source': 'assistant'}
[0] third_anglo_afghan_war.html
就是这样！我们已经成功构建了一个由 OpenAI 助手支持的代理。