OpenAI Assistant Agent#

Open AI Assistant and Azure OpenAI Assistant are server-side APIs for building agents. They can be used to build agents in AutoGen. This cookbook demonstrates how to to use OpenAI Assistant to create an agent that can run code and Q&A over document.

Message Protocol#

First, we need to specify the message protocol for the agent backed by OpenAI Assistant. The message protocol defines the structure of messages handled and published by the agent. For illustration, we define a simple message protocol of 4 message types: Message, Reset, UploadForCodeInterpreter and 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

The TextMessage message type is used to communicate with the agent. It has a content field that contains the message content, and a source field for the sender. The Reset message type is a control message that resets the memory of the agent. It has no fields. This is useful when we need to start a new conversation with the agent.

The UploadForCodeInterpreter message type is used to upload data files for the code interpreter and UploadForFileSearch message type is used to upload documents for file search. Both message types have a file_path field that contains the local path to the file to be uploaded.

Defining the Agent#

Next, we define the agent class. The agent class constructor has the following arguments: description, client, assistant_id, thread_id, and assistant_event_handler_factory. The client argument is the OpenAI async client object, and the assistant_event_handler_factory is for creating an assistant event handler to handle OpenAI Assistant events. This can be used to create streaming output from the assistant.

The agent class has the following message handlers:

  • handle_message: Handles the TextMessage message type, and sends back the response from the assistant.

  • handle_reset: Handles the Reset message type, and resets the memory of the assistant agent.

  • handle_upload_for_code_interpreter: Handles the UploadForCodeInterpreter message type, and uploads the file to the code interpreter.

  • handle_upload_for_file_search: Handles the UploadForFileSearch message type, and uploads the document to the file search.

The memory of the assistant is stored inside a thread, which is kept in the server side. The thread is referenced by the thread_id argument.

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

import aiofiles
from autogen_core.base import AgentId, MessageContext
from autogen_core.components import 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.beta.vector_stores.file_batches.upload_and_poll(
                    vector_store_id=message.vector_store_id,
                    files=[(file_name, file_content)],
                )
            )
        )

The agent class is a thin wrapper around the OpenAI Assistant API to implement the message protocol. More features, such as multi-modal message handling, can be added by extending the message protocol.

Assistant Event Handler#

The assistant event handler provides call-backs for handling Assistant API specific events. This is useful for handling streaming output from the assistant and further user interface integration.

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

Using the Agent#

First we need to use the openai client to create the actual assistant, thread, and vector store. Our AutoGen agent will be using these.

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.beta.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]}},
)

Then, we create a runtime, and register an agent factory function for this agent with the runtime.

from autogen_core.application 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")

Let’s turn on logging to see what’s happening under the hood.

import logging

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

Let’s send a greeting message to the agent, and see the response streamed back.

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'}

Assistant with Code Interpreter#

Let’s ask some math question to the agent, and see it uses the code interpreter to answer the question.

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'}

Let’s get some data from Seattle Open Data portal. We will be using the City of Seattle Wage Data. Let’s download it first.

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)

Let’s send the file to the agent using an UploadForCodeInterpreter message.

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

We can now ask some questions about the data to the agent.

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'}
On this page
Edit on GitHub
Show Source