OpenAIAssistantBase Class

  • Reference

OpenAI Assistant Base class.

Manages the interaction with OpenAI Assistants.

Note: This class is experimental and may change in the future.

Initialize an OpenAIAssistant Base.

Inheritance
Agent
OpenAIAssistantBase

Constructor

OpenAIAssistantBase(ai_model_id: str, client: AsyncOpenAI, service_id: str, *, kernel: Kernel | None = None, id: str | None = None, name: str | None = None, description: str | None = None, instructions: str | None = None, enable_code_interpreter: bool | None = None, enable_file_search: bool | None = None, enable_json_response: bool | None = None, code_interpreter_file_ids: list[str] | None = [], temperature: float | None = None, top_p: float | None = None, vector_store_id: str | None = None, metadata: dict[str, Any] | None = {}, max_completion_tokens: int | None = None, max_prompt_tokens: int | None = None, parallel_tool_calls_enabled: bool | None = True, truncation_message_count: int | None = None, assistant: Assistant | None = None, polling_options: RunPollingOptions = None, file_search_file_ids: Annotated[list[str] | None, MaxLen(max_length=20)] = None)

Parameters

Name Description
ai_model_id
Required

The AI model id. Defaults to None.
client
Required

The client, either AsyncOpenAI or AsyncAzureOpenAI.
service_id
Required

The service id.
kernel
Required

The kernel. (optional)
id
Required

The id. Defaults to None. (optional)
name
Required

The name. Defaults to None. (optional)
description
Required

The description. Defaults to None. (optional)
default_headers
Required

The default headers. Defaults to None. (optional)
instructions
Required

The instructions. Defaults to None. (optional)
enable_code_interpreter
Required

Enable code interpreter. Defaults to False. (optional)
enable_file_search
Required

Enable file search. Defaults to False. (optional)
enable_json_response
Required

Enable JSON response. Defaults to False. (optional)
code_interpreter_file_ids
Required

The file ids. Defaults to []. (optional)
temperature
Required

The temperature. Defaults to None. (optional)
top_p
Required

The top p. Defaults to None. (optional)
vector_store_id
Required

The vector store id. Defaults to None. (optional)
metadata
Required

The metadata. Defaults to {}. (optional)
max_completion_tokens
Required

The max completion tokens. Defaults to None. (optional)
max_prompt_tokens
Required

The max prompt tokens. Defaults to None. (optional)
parallel_tool_calls_enabled
Required

Enable parallel tool calls. Defaults to True. (optional)
truncation_message_count
Required

The truncation message count. Defaults to None. (optional)
kwargs
Required

The keyword arguments.

Keyword-Only Parameters

Name Description
kernel
Required
id
Required
name
Required
description
Required
instructions
Required
enable_code_interpreter
Required
enable_file_search
Required
enable_json_response
Required
code_interpreter_file_ids
Required
temperature
Required
top_p
Required
vector_store_id
Required
metadata
Required
max_completion_tokens
Required
max_prompt_tokens
Required
parallel_tool_calls_enabled
Default value: True
truncation_message_count
Required
assistant
Required
polling_options
Required
file_search_file_ids
Required

Methods

add_chat_message

Add a chat message.
add_file

Add a file for use with the Assistant.
create_assistant

Create the assistant.
create_channel

Create a channel.
create_thread

Create a thread.
create_vector_store

Create a vector store.
delete

Delete the assistant.
delete_file

Delete a file.
delete_thread

Delete a thread.
delete_vector_store

Delete a vector store.
get_channel_keys

Get the channel keys.
get_thread_messages

Get the messages for the specified thread.
invoke

Invoke the chat assistant.

The supplied arguments will take precedence over the specified assistant level attributes.
invoke_stream

Invoke the chat assistant with streaming.
model_post_init

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that's what pydantic-core passes when calling it.
modify_assistant

Modify the assistant.

add_chat_message

Add a chat message.

async add_chat_message(thread_id: str, message: ChatMessageContent) -> Message

Parameters

Name Description
thread_id
Required

The thread id.
message
Required

The chat message.

Returns

Type Description
<xref:Message>

The message.

add_file

Add a file for use with the Assistant.

async add_file(file_path: str, purpose: Literal['assistants', 'vision']) -> str

Parameters

Name Description
file_path
Required

The file path.
purpose
Required

The purpose. Can be "assistants" or "vision".

Returns

Type Description
str

The file id.

Exceptions

Type Description
AgentInitializationError

If the client has not been initialized or the file is not found.

create_assistant

Create the assistant.

async create_assistant(ai_model_id: str | None = None, description: str | None = None, instructions: str | None = None, name: str | None = None, enable_code_interpreter: bool | None = None, code_interpreter_file_ids: list[str] | None = None, enable_file_search: bool | None = None, vector_store_id: str | None = None, metadata: dict[str, str] | None = {}, **kwargs: Any) -> Assistant

Parameters

Name Description
ai_model_id

The AI model id. Defaults to None. (optional)

Default value: None
description

The description. Defaults to None. (optional)

Default value: None
instructions

The instructions. Defaults to None. (optional)

Default value: None
name

The name. Defaults to None. (optional)

Default value: None
enable_code_interpreter

Enable code interpreter. Defaults to None. (optional)

Default value: None
enable_file_search

Enable file search. Defaults to None. (optional)

Default value: None
code_interpreter_file_ids

The file ids. Defaults to None. (optional)

Default value: None
vector_store_id

The vector store id. Defaults to None. (optional)

Default value: None
metadata

The metadata. Defaults to {}. (optional)

Default value: {}
kwargs
Required

Extra keyword arguments.

Returns

Type Description
<xref:Assistant>

The assistant

create_channel

Create a channel.

async create_channel() -> AgentChannel

create_thread

Create a thread.

async create_thread(*, code_interpreter_file_ids: list[str] | None = [], messages: list[ChatMessageContent] | None = [], vector_store_id: str | None = None, metadata: dict[str, str] = {}) -> str

Parameters

Name Description
code_interpreter_file_ids
Required

The code interpreter file ids. Defaults to an empty list. (optional)
messages
Required

The chat messages. Defaults to an empty list. (optional)
vector_store_id
Required

The vector store id. Defaults to None. (optional)
metadata
Required

The metadata. Defaults to an empty dictionary. (optional)

Keyword-Only Parameters

Name Description
code_interpreter_file_ids
Required
messages
Required
vector_store_id
Required
metadata
Required

Returns

Type Description
str

The thread id.

create_vector_store

Create a vector store.

async create_vector_store(file_ids: str | list[str]) -> str

Parameters

Name Description
file_ids
Required

The file ids either as a str of a single file ID or a list of strings of file IDs.

Returns

Type Description

The vector store id.

Exceptions

Type Description
AgentExecutionError

If there is an error creating the vector store.

delete

Delete the assistant.

async delete() -> bool

Returns

Type Description
bool

True if the assistant is deleted.

delete_file

Delete a file.

async delete_file(file_id: str) -> None

Parameters

Name Description
file_id
Required

The file id.

delete_thread

Delete a thread.

async delete_thread(thread_id: str) -> None

Parameters

Name Description
thread_id
Required

The thread id.

delete_vector_store

Delete a vector store.

async delete_vector_store(vector_store_id: str) -> None

Parameters

Name Description
vector_store_id
Required

The vector store id.

Exceptions

Type Description
AgentExecutionError

If there is an error deleting the vector store.

get_channel_keys

Get the channel keys.

get_channel_keys() -> Iterable[str]

Returns

Type Description
Iterable[str]

The channel keys.

get_thread_messages

Get the messages for the specified thread.

async get_thread_messages(thread_id: str) -> AsyncIterable[ChatMessageContent]

Parameters

Name Description
thread_id
Required

The thread id.

invoke

Invoke the chat assistant.

The supplied arguments will take precedence over the specified assistant level attributes.

async invoke(thread_id: str, *, ai_model_id: str | None = None, enable_code_interpreter: bool | None = False, enable_file_search: bool | None = False, enable_json_response: bool | None = None, max_completion_tokens: int | None = None, max_prompt_tokens: int | None = None, parallel_tool_calls_enabled: bool | None = True, truncation_message_count: int | None = None, temperature: float | None = None, top_p: float | None = None, metadata: dict[str, str] | None = None, **kwargs: Any) -> AsyncIterable[ChatMessageContent]

Parameters

Name Description
thread_id
Required

The thread id.
ai_model_id
Required

The AI model id. Defaults to None. (optional)
enable_code_interpreter
Required

Enable code interpreter. Defaults to False. (optional)
enable_file_search
Required

Enable file search. Defaults to False. (optional)
enable_json_response
Required

Enable JSON response. Defaults to False. (optional)
max_completion_tokens
Required

The max completion tokens. Defaults to None. (optional)
max_prompt_tokens
Required

The max prompt tokens. Defaults to None. (optional)
parallel_tool_calls_enabled
Required

Enable parallel tool calls. Defaults to True. (optional)
truncation_message_count
Required

The truncation message count. Defaults to None. (optional)
temperature
Required

The temperature. Defaults to None. (optional)
top_p
Required

The top p. Defaults to None. (optional)
metadata
Required

The metadata. Defaults to {}. (optional)
kwargs
Required

Extra keyword arguments.

Keyword-Only Parameters

Name Description
ai_model_id
Required
enable_code_interpreter
Required
enable_file_search
Required
enable_json_response
Required
max_completion_tokens
Required
max_prompt_tokens
Required
parallel_tool_calls_enabled
Default value: True
truncation_message_count
Required
temperature
Required
top_p
Required
metadata
Required

invoke_stream

Invoke the chat assistant with streaming.

async invoke_stream(thread_id: str, *, messages: list[ChatMessageContent] | None = None, ai_model_id: str | None = None, enable_code_interpreter: bool | None = False, enable_file_search: bool | None = False, enable_json_response: bool | None = None, max_completion_tokens: int | None = None, max_prompt_tokens: int | None = None, parallel_tool_calls_enabled: bool | None = True, truncation_message_count: int | None = None, temperature: float | None = None, top_p: float | None = None, metadata: dict[str, str] | None = None, **kwargs: Any) -> AsyncIterable[ChatMessageContent]

Parameters

Name Description
thread_id
Required

Keyword-Only Parameters

Name Description
messages
Required
ai_model_id
Required
enable_code_interpreter
Required
enable_file_search
Required
enable_json_response
Required
max_completion_tokens
Required
max_prompt_tokens
Required
parallel_tool_calls_enabled
Default value: True
truncation_message_count
Required
temperature
Required
top_p
Required
metadata
Required

model_post_init

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that's what pydantic-core passes when calling it.

model_post_init(context: Any, /) -> None

Positional-Only Parameters

Name Description
context
Required

Parameters

Name Description
self
Required

The BaseModel instance.
context
Required

The context.

modify_assistant

Modify the assistant.

async modify_assistant(assistant_id: str, **kwargs: Any) -> Assistant

Parameters

Name Description
assistant_id
Required

The assistant's current ID.
kwargs
Required

Extra keyword arguments.

Returns

Type Description
<xref:Assistant>

The modified assistant.

Attributes

model_computed_fields

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'populate_by_name': True, 'validate_assignment': True}

model_fields

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.fields from Pydantic V1.

model_fields: ClassVar[Dict[str, FieldInfo]] = {'ai_model_id': FieldInfo(annotation=str, required=True), 'assistant': FieldInfo(annotation=Union[Assistant, NoneType], required=False, default=None), 'client': FieldInfo(annotation=AsyncOpenAI, required=True), 'code_interpreter_file_ids': FieldInfo(annotation=Union[list[str], NoneType], required=False, default_factory=list, metadata=[MaxLen(max_length=20)]), 'description': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'enable_code_interpreter': FieldInfo(annotation=Union[bool, NoneType], required=False, default=False), 'enable_file_search': FieldInfo(annotation=Union[bool, NoneType], required=False, default=False), 'enable_json_response': FieldInfo(annotation=Union[bool, NoneType], required=False, default=False), 'file_search_file_ids': FieldInfo(annotation=Union[list[str], NoneType], required=False, default_factory=list, metadata=[MaxLen(max_length=20)]), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'instructions': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'kernel': FieldInfo(annotation=Kernel, required=False, default_factory=Kernel), 'max_completion_tokens': FieldInfo(annotation=Union[int, NoneType], required=False, default=None), 'max_prompt_tokens': FieldInfo(annotation=Union[int, NoneType], required=False, default=None), 'metadata': FieldInfo(annotation=Union[dict[str, Any], NoneType], required=False, default_factory=dict, metadata=[MaxLen(max_length=16)]), 'name': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'parallel_tool_calls_enabled': FieldInfo(annotation=Union[bool, NoneType], required=False, default=True), 'polling_options': FieldInfo(annotation=RunPollingOptions, required=False, default_factory=RunPollingOptions), 'temperature': FieldInfo(annotation=Union[float, NoneType], required=False, default=None), 'top_p': FieldInfo(annotation=Union[float, NoneType], required=False, default=None), 'truncation_message_count': FieldInfo(annotation=Union[int, NoneType], required=False, default=None), 'vector_store_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None)}

tools

The tools.

Returns

Type Description
list[dict[str, str]]

The tools.

ai_model_id 

ai_model_id: str

allowed_message_roles 

allowed_message_roles: ClassVar[list[str]] = [AuthorRole.USER, AuthorRole.ASSISTANT]

assistant 

assistant: Assistant | None

channel_type 

channel_type: ClassVar[type[AgentChannel]] = None

client 

client: AsyncOpenAI

code_interpreter_file_ids 

code_interpreter_file_ids: list[str] | None

enable_code_interpreter 

enable_code_interpreter: bool | None
enable_file_search: bool | None

enable_json_response 

enable_json_response: bool | None

error_message_states 

error_message_states: ClassVar[list[str]] = ['failed', 'canceled', 'expired']

file_search_file_ids 

file_search_file_ids: list[str] | None

is_experimental 

is_experimental = True

max_completion_tokens 

max_completion_tokens: int | None

max_prompt_tokens 

max_prompt_tokens: int | None

metadata 

metadata: dict[str, Any] | None

parallel_tool_calls_enabled 

parallel_tool_calls_enabled: bool | None

polling_options 

polling_options: RunPollingOptions

polling_status 

polling_status: ClassVar[list[str]] = ['queued', 'in_progress', 'cancelling']

temperature 

temperature: float | None

top_p 

top_p: float | None

truncation_message_count 

truncation_message_count: int | None

vector_store_id 

vector_store_id: str | None