Openai

OpenAIAgent

Bases: AgentRunner

OpenAI agent.

Subclasses AgentRunner with a OpenAIAgentWorker.

For the legacy implementation see:

from llama_index..agent.legacy.openai.base import OpenAIAgent

Source code in llama-index-integrations/agent/llama-index-agent-openai/llama_index/agent/openai/base.py

class OpenAIAgent(AgentRunner):
    """OpenAI agent.

    Subclasses AgentRunner with a OpenAIAgentWorker.

    For the legacy implementation see:
    ```python
    from llama_index..agent.legacy.openai.base import OpenAIAgent
    ```

    """

    def __init__(
        self,
        tools: List[BaseTool],
        llm: OpenAI,
        memory: BaseMemory,
        prefix_messages: List[ChatMessage],
        verbose: bool = False,
        max_function_calls: int = DEFAULT_MAX_FUNCTION_CALLS,
        default_tool_choice: str = "auto",
        callback_manager: Optional[CallbackManager] = None,
        tool_retriever: Optional[ObjectRetriever[BaseTool]] = None,
        tool_call_parser: Optional[Callable[[OpenAIToolCall], Dict]] = None,
    ) -> None:
        """Init params."""
        callback_manager = callback_manager or llm.callback_manager
        step_engine = OpenAIAgentWorker.from_tools(
            tools=tools,
            tool_retriever=tool_retriever,
            llm=llm,
            verbose=verbose,
            max_function_calls=max_function_calls,
            callback_manager=callback_manager,
            prefix_messages=prefix_messages,
            tool_call_parser=tool_call_parser,
        )
        super().__init__(
            step_engine,
            memory=memory,
            llm=llm,
            callback_manager=callback_manager,
            default_tool_choice=default_tool_choice,
        )

    @classmethod
    def from_tools(
        cls,
        tools: Optional[List[BaseTool]] = None,
        tool_retriever: Optional[ObjectRetriever[BaseTool]] = None,
        llm: Optional[LLM] = None,
        chat_history: Optional[List[ChatMessage]] = None,
        memory: Optional[BaseMemory] = None,
        memory_cls: Type[BaseMemory] = ChatMemoryBuffer,
        verbose: bool = False,
        max_function_calls: int = DEFAULT_MAX_FUNCTION_CALLS,
        default_tool_choice: str = "auto",
        callback_manager: Optional[CallbackManager] = None,
        system_prompt: Optional[str] = None,
        prefix_messages: Optional[List[ChatMessage]] = None,
        tool_call_parser: Optional[Callable[[OpenAIToolCall], Dict]] = None,
        **kwargs: Any,
    ) -> "OpenAIAgent":
        """Create an OpenAIAgent from a list of tools.

        Similar to `from_defaults` in other classes, this method will
        infer defaults for a variety of parameters, including the LLM,
        if they are not specified.

        """
        tools = tools or []

        chat_history = chat_history or []
        llm = llm or Settings.llm
        if not isinstance(llm, OpenAI):
            raise ValueError("llm must be a OpenAI instance")

        if callback_manager is not None:
            llm.callback_manager = callback_manager

        memory = memory or memory_cls.from_defaults(chat_history, llm=llm)

        if not llm.metadata.is_function_calling_model:
            raise ValueError(
                f"Model name {llm.model} does not support function calling API. "
            )

        if system_prompt is not None:
            if prefix_messages is not None:
                raise ValueError(
                    "Cannot specify both system_prompt and prefix_messages"
                )
            prefix_messages = [ChatMessage(content=system_prompt, role="system")]

        prefix_messages = prefix_messages or []

        return cls(
            tools=tools,
            tool_retriever=tool_retriever,
            llm=llm,
            memory=memory,
            prefix_messages=prefix_messages,
            verbose=verbose,
            max_function_calls=max_function_calls,
            callback_manager=callback_manager,
            default_tool_choice=default_tool_choice,
            tool_call_parser=tool_call_parser,
        )

from_tools classmethod

from_tools(tools: Optional[List[BaseTool]] = None, tool_retriever: Optional[ObjectRetriever[BaseTool]] = None, llm: Optional[LLM] = None, chat_history: Optional[List[ChatMessage]] = None, memory: Optional[BaseMemory] = None, memory_cls: Type[BaseMemory] = ChatMemoryBuffer, verbose: bool = False, max_function_calls: int = DEFAULT_MAX_FUNCTION_CALLS, default_tool_choice: str = 'auto', callback_manager: Optional[CallbackManager] = None, system_prompt: Optional[str] = None, prefix_messages: Optional[List[ChatMessage]] = None, tool_call_parser: Optional[Callable[[OpenAIToolCall], Dict]] = None, **kwargs: Any) -> OpenAIAgent

Create an OpenAIAgent from a list of tools.

Similar to from_defaults in other classes, this method will infer defaults for a variety of parameters, including the LLM, if they are not specified.

Source code in llama-index-integrations/agent/llama-index-agent-openai/llama_index/agent/openai/base.py

@classmethod
def from_tools(
    cls,
    tools: Optional[List[BaseTool]] = None,
    tool_retriever: Optional[ObjectRetriever[BaseTool]] = None,
    llm: Optional[LLM] = None,
    chat_history: Optional[List[ChatMessage]] = None,
    memory: Optional[BaseMemory] = None,
    memory_cls: Type[BaseMemory] = ChatMemoryBuffer,
    verbose: bool = False,
    max_function_calls: int = DEFAULT_MAX_FUNCTION_CALLS,
    default_tool_choice: str = "auto",
    callback_manager: Optional[CallbackManager] = None,
    system_prompt: Optional[str] = None,
    prefix_messages: Optional[List[ChatMessage]] = None,
    tool_call_parser: Optional[Callable[[OpenAIToolCall], Dict]] = None,
    **kwargs: Any,
) -> "OpenAIAgent":
    """Create an OpenAIAgent from a list of tools.

    Similar to `from_defaults` in other classes, this method will
    infer defaults for a variety of parameters, including the LLM,
    if they are not specified.

    """
    tools = tools or []

    chat_history = chat_history or []
    llm = llm or Settings.llm
    if not isinstance(llm, OpenAI):
        raise ValueError("llm must be a OpenAI instance")

    if callback_manager is not None:
        llm.callback_manager = callback_manager

    memory = memory or memory_cls.from_defaults(chat_history, llm=llm)

    if not llm.metadata.is_function_calling_model:
        raise ValueError(
            f"Model name {llm.model} does not support function calling API. "
        )

    if system_prompt is not None:
        if prefix_messages is not None:
            raise ValueError(
                "Cannot specify both system_prompt and prefix_messages"
            )
        prefix_messages = [ChatMessage(content=system_prompt, role="system")]

    prefix_messages = prefix_messages or []

    return cls(
        tools=tools,
        tool_retriever=tool_retriever,
        llm=llm,
        memory=memory,
        prefix_messages=prefix_messages,
        verbose=verbose,
        max_function_calls=max_function_calls,
        callback_manager=callback_manager,
        default_tool_choice=default_tool_choice,
        tool_call_parser=tool_call_parser,
    )

OpenAIAssistantAgent

Bases: BaseAgent

OpenAIAssistant agent.

Wrapper around OpenAI assistant API: https://platform.openai.com/docs/assistants/overview

Source code in llama-index-integrations/agent/llama-index-agent-openai/llama_index/agent/openai/openai_assistant_agent.py

class OpenAIAssistantAgent(BaseAgent):
    """OpenAIAssistant agent.

    Wrapper around OpenAI assistant API: https://platform.openai.com/docs/assistants/overview

    """

    def __init__(
        self,
        client: Any,
        assistant: Any,
        tools: Optional[List[BaseTool]],
        callback_manager: Optional[CallbackManager] = None,
        thread_id: Optional[str] = None,
        instructions_prefix: Optional[str] = None,
        run_retrieve_sleep_time: float = 0.1,
        file_dict: Dict[str, str] = {},
        verbose: bool = False,
    ) -> None:
        """Init params."""
        from openai import OpenAI
        from openai.types.beta.assistant import Assistant

        self._client = cast(OpenAI, client)
        self._assistant = cast(Assistant, assistant)
        self._tools = tools or []
        if thread_id is None:
            thread = self._client.beta.threads.create()
            thread_id = thread.id
        self._thread_id = thread_id
        self._instructions_prefix = instructions_prefix
        self._run_retrieve_sleep_time = run_retrieve_sleep_time
        self._verbose = verbose
        self.file_dict = file_dict

        self.callback_manager = callback_manager or CallbackManager([])

    @classmethod
    def from_new(
        cls,
        name: str,
        instructions: str,
        tools: Optional[List[BaseTool]] = None,
        openai_tools: Optional[List[Dict]] = None,
        thread_id: Optional[str] = None,
        model: str = "gpt-4-1106-preview",
        instructions_prefix: Optional[str] = None,
        run_retrieve_sleep_time: float = 0.1,
        files: Optional[List[str]] = None,
        callback_manager: Optional[CallbackManager] = None,
        verbose: bool = False,
        file_ids: Optional[List[str]] = None,
        api_key: Optional[str] = None,
    ) -> "OpenAIAssistantAgent":
        """From new assistant.

        Args:
            name: name of assistant
            instructions: instructions for assistant
            tools: list of tools
            openai_tools: list of openai tools
            thread_id: thread id
            model: model
            run_retrieve_sleep_time: run retrieve sleep time
            files: files
            instructions_prefix: instructions prefix
            callback_manager: callback manager
            verbose: verbose
            file_ids: list of file ids
            api_key: OpenAI API key

        """
        from openai import OpenAI

        # this is the set of openai tools
        # not to be confused with the tools we pass in for function calling
        openai_tools = openai_tools or []
        tools = tools or []
        tool_fns = [t.metadata.to_openai_tool() for t in tools]
        all_openai_tools = openai_tools + tool_fns

        # initialize client
        client = OpenAI(api_key=api_key)

        # process files
        files = files or []
        file_ids = file_ids or []

        file_dict = _process_files(client, files)

        # TODO: openai's typing is a bit sus
        all_openai_tools = cast(List[Any], all_openai_tools)
        assistant = client.beta.assistants.create(
            name=name,
            instructions=instructions,
            tools=cast(List[Any], all_openai_tools),
            model=model,
        )
        return cls(
            client,
            assistant,
            tools,
            callback_manager=callback_manager,
            thread_id=thread_id,
            instructions_prefix=instructions_prefix,
            file_dict=file_dict,
            run_retrieve_sleep_time=run_retrieve_sleep_time,
            verbose=verbose,
        )

    @classmethod
    def from_existing(
        cls,
        assistant_id: str,
        tools: Optional[List[BaseTool]] = None,
        thread_id: Optional[str] = None,
        instructions_prefix: Optional[str] = None,
        run_retrieve_sleep_time: float = 0.1,
        callback_manager: Optional[CallbackManager] = None,
        api_key: Optional[str] = None,
        verbose: bool = False,
    ) -> "OpenAIAssistantAgent":
        """From existing assistant id.

        Args:
            assistant_id: id of assistant
            tools: list of BaseTools Assistant can use
            thread_id: thread id
            run_retrieve_sleep_time: run retrieve sleep time
            instructions_prefix: instructions prefix
            callback_manager: callback manager
            api_key: OpenAI API key
            verbose: verbose

        """
        from openai import OpenAI

        # initialize client
        client = OpenAI(api_key=api_key)

        # get assistant
        assistant = client.beta.assistants.retrieve(assistant_id)
        # assistant.tools is incompatible with BaseTools so have to pass from params

        return cls(
            client,
            assistant,
            tools=tools,
            callback_manager=callback_manager,
            thread_id=thread_id,
            instructions_prefix=instructions_prefix,
            run_retrieve_sleep_time=run_retrieve_sleep_time,
            verbose=verbose,
        )

    @property
    def assistant(self) -> Any:
        """Get assistant."""
        return self._assistant

    @property
    def client(self) -> Any:
        """Get client."""
        return self._client

    @property
    def thread_id(self) -> str:
        """Get thread id."""
        return self._thread_id

    @property
    def files_dict(self) -> Dict[str, str]:
        """Get files dict."""
        return self.file_dict

    @property
    def chat_history(self) -> List[ChatMessage]:
        raw_messages = self._client.beta.threads.messages.list(
            thread_id=self._thread_id, order="asc"
        )
        return from_openai_thread_messages(list(raw_messages))

    def reset(self) -> None:
        """Delete and create a new thread."""
        self._client.beta.threads.delete(self._thread_id)
        thread = self._client.beta.threads.create()
        thread_id = thread.id
        self._thread_id = thread_id

    def get_tools(self, message: str) -> List[BaseTool]:
        """Get tools."""
        return self._tools

    def upload_files(self, files: List[str]) -> Dict[str, Any]:
        """Upload files."""
        return _process_files(self._client, files)

    def add_message(
        self,
        message: str,
        file_ids: Optional[List[str]] = None,
        tools: Optional[List[Dict[str, Any]]] = None,
    ) -> Any:
        """Add message to assistant."""
        attachments = format_attachments(file_ids=file_ids, tools=tools)
        return self._client.beta.threads.messages.create(
            thread_id=self._thread_id,
            role="user",
            content=message,
            attachments=attachments,
        )

    def _run_function_calling(self, run: Any) -> List[ToolOutput]:
        """Run function calling."""
        tool_calls = run.required_action.submit_tool_outputs.tool_calls
        tool_output_dicts = []
        tool_output_objs: List[ToolOutput] = []

        for tool_call in tool_calls:
            fn_obj = tool_call.function
            _, tool_output = call_function(self._tools, fn_obj, verbose=self._verbose)
            tool_output_dicts.append(
                {"tool_call_id": tool_call.id, "output": str(tool_output)}
            )
            tool_output_objs.append(tool_output)

        # submit tool outputs
        # TODO: openai's typing is a bit sus
        self._client.beta.threads.runs.submit_tool_outputs(
            thread_id=self._thread_id,
            run_id=run.id,
            tool_outputs=cast(List[Any], tool_output_dicts),
        )
        return tool_output_objs

    async def _arun_function_calling(self, run: Any) -> List[ToolOutput]:
        """Run function calling."""
        tool_calls = run.required_action.submit_tool_outputs.tool_calls
        tool_output_dicts = []
        tool_output_objs: List[ToolOutput] = []
        for tool_call in tool_calls:
            fn_obj = tool_call.function
            _, tool_output = await acall_function(
                self._tools, fn_obj, verbose=self._verbose
            )
            tool_output_dicts.append(
                {"tool_call_id": tool_call.id, "output": str(tool_output)}
            )
            tool_output_objs.append(tool_output)

        # submit tool outputs
        self._client.beta.threads.runs.submit_tool_outputs(
            thread_id=self._thread_id,
            run_id=run.id,
            tool_outputs=cast(List[Any], tool_output_dicts),
        )
        return tool_output_objs

    def run_assistant(
        self, instructions_prefix: Optional[str] = None
    ) -> Tuple[Any, Dict]:
        """Run assistant."""
        instructions_prefix = instructions_prefix or self._instructions_prefix
        run = self._client.beta.threads.runs.create(
            thread_id=self._thread_id,
            assistant_id=self._assistant.id,
            instructions=instructions_prefix,
        )
        from openai.types.beta.threads import Run

        run = cast(Run, run)

        sources = []
        while run.status in ["queued", "in_progress", "requires_action"]:
            run = self._client.beta.threads.runs.retrieve(
                thread_id=self._thread_id, run_id=run.id
            )
            if run.status == "requires_action":
                cur_tool_outputs = self._run_function_calling(run)
                sources.extend(cur_tool_outputs)

            time.sleep(self._run_retrieve_sleep_time)
        if run.status == "failed":
            raise ValueError(
                f"Run failed with status {run.status}.\n" f"Error: {run.last_error}"
            )
        return run, {"sources": sources}

    async def arun_assistant(
        self, instructions_prefix: Optional[str] = None
    ) -> Tuple[Any, Dict]:
        """Run assistant."""
        instructions_prefix = instructions_prefix or self._instructions_prefix
        run = self._client.beta.threads.runs.create(
            thread_id=self._thread_id,
            assistant_id=self._assistant.id,
            instructions=instructions_prefix,
        )
        from openai.types.beta.threads import Run

        run = cast(Run, run)

        sources = []

        while run.status in ["queued", "in_progress", "requires_action"]:
            run = self._client.beta.threads.runs.retrieve(
                thread_id=self._thread_id, run_id=run.id
            )
            if run.status == "requires_action":
                cur_tool_outputs = await self._arun_function_calling(run)
                sources.extend(cur_tool_outputs)

            await asyncio.sleep(self._run_retrieve_sleep_time)
        if run.status == "failed":
            raise ValueError(
                f"Run failed with status {run.status}.\n" f"Error: {run.last_error}"
            )
        return run, {"sources": sources}

    @property
    def latest_message(self) -> ChatMessage:
        """Get latest message."""
        raw_messages = self._client.beta.threads.messages.list(
            thread_id=self._thread_id, order="desc"
        )
        messages = from_openai_thread_messages(list(raw_messages))
        return messages[0]

    def _chat(
        self,
        message: str,
        chat_history: Optional[List[ChatMessage]] = None,
        function_call: Union[str, dict] = "auto",
        mode: ChatResponseMode = ChatResponseMode.WAIT,
    ) -> AGENT_CHAT_RESPONSE_TYPE:
        """Main chat interface."""
        # TODO: since chat interface doesn't expose additional kwargs
        # we can't pass in file_ids per message
        _added_message_obj = self.add_message(message)
        _run, metadata = self.run_assistant(
            instructions_prefix=self._instructions_prefix,
        )
        latest_message = self.latest_message
        # get most recent message content
        return AgentChatResponse(
            response=str(latest_message.content),
            sources=metadata["sources"],
        )

    async def _achat(
        self,
        message: str,
        chat_history: Optional[List[ChatMessage]] = None,
        function_call: Union[str, dict] = "auto",
        mode: ChatResponseMode = ChatResponseMode.WAIT,
    ) -> AGENT_CHAT_RESPONSE_TYPE:
        """Asynchronous main chat interface."""
        self.add_message(message)
        run, metadata = await self.arun_assistant(
            instructions_prefix=self._instructions_prefix,
        )
        latest_message = self.latest_message
        # get most recent message content
        return AgentChatResponse(
            response=str(latest_message.content),
            sources=metadata["sources"],
        )

    @trace_method("chat")
    def chat(
        self,
        message: str,
        chat_history: Optional[List[ChatMessage]] = None,
        function_call: Union[str, dict] = "auto",
    ) -> AgentChatResponse:
        with self.callback_manager.event(
            CBEventType.AGENT_STEP,
            payload={EventPayload.MESSAGES: [message]},
        ) as e:
            chat_response = self._chat(
                message, chat_history, function_call, mode=ChatResponseMode.WAIT
            )
            assert isinstance(chat_response, AgentChatResponse)
            e.on_end(payload={EventPayload.RESPONSE: chat_response})
        return chat_response

    @trace_method("chat")
    async def achat(
        self,
        message: str,
        chat_history: Optional[List[ChatMessage]] = None,
        function_call: Union[str, dict] = "auto",
    ) -> AgentChatResponse:
        with self.callback_manager.event(
            CBEventType.AGENT_STEP,
            payload={EventPayload.MESSAGES: [message]},
        ) as e:
            chat_response = await self._achat(
                message, chat_history, function_call, mode=ChatResponseMode.WAIT
            )
            assert isinstance(chat_response, AgentChatResponse)
            e.on_end(payload={EventPayload.RESPONSE: chat_response})
        return chat_response

    @trace_method("chat")
    def stream_chat(
        self,
        message: str,
        chat_history: Optional[List[ChatMessage]] = None,
        function_call: Union[str, dict] = "auto",
    ) -> StreamingAgentChatResponse:
        raise NotImplementedError("stream_chat not implemented")

    @trace_method("chat")
    async def astream_chat(
        self,
        message: str,
        chat_history: Optional[List[ChatMessage]] = None,
        function_call: Union[str, dict] = "auto",
    ) -> StreamingAgentChatResponse:
        raise NotImplementedError("astream_chat not implemented")

assistant property

assistant: Any

Get assistant.

client property

client: Any

Get client.

thread_id property

thread_id: str

Get thread id.

files_dict property

files_dict: Dict[str, str]

Get files dict.

latest_message property

latest_message: ChatMessage

Get latest message.

from_new classmethod

from_new(name: str, instructions: str, tools: Optional[List[BaseTool]] = None, openai_tools: Optional[List[Dict]] = None, thread_id: Optional[str] = None, model: str = 'gpt-4-1106-preview', instructions_prefix: Optional[str] = None, run_retrieve_sleep_time: float = 0.1, files: Optional[List[str]] = None, callback_manager: Optional[CallbackManager] = None, verbose: bool = False, file_ids: Optional[List[str]] = None, api_key: Optional[str] = None) -> OpenAIAssistantAgent

From new assistant.

Parameters:

Name	Type	Description	Default
name	str	name of assistant	required
instructions	str	instructions for assistant	required
tools	Optional[List[BaseTool]]	list of tools	None
openai_tools	Optional[List[Dict]]	list of openai tools	None
thread_id	Optional[str]	thread id	None
model	str	model	'gpt-4-1106-preview'
run_retrieve_sleep_time	float	run retrieve sleep time	0.1
files	Optional[List[str]]	files	None
instructions_prefix	Optional[str]	instructions prefix	None
callback_manager	Optional[CallbackManager]	callback manager	None
verbose	bool	verbose	False
file_ids	Optional[List[str]]	list of file ids	None
api_key	Optional[str]	OpenAI API key	None

Source code in llama-index-integrations/agent/llama-index-agent-openai/llama_index/agent/openai/openai_assistant_agent.py

@classmethod
def from_new(
    cls,
    name: str,
    instructions: str,
    tools: Optional[List[BaseTool]] = None,
    openai_tools: Optional[List[Dict]] = None,
    thread_id: Optional[str] = None,
    model: str = "gpt-4-1106-preview",
    instructions_prefix: Optional[str] = None,
    run_retrieve_sleep_time: float = 0.1,
    files: Optional[List[str]] = None,
    callback_manager: Optional[CallbackManager] = None,
    verbose: bool = False,
    file_ids: Optional[List[str]] = None,
    api_key: Optional[str] = None,
) -> "OpenAIAssistantAgent":
    """From new assistant.

    Args:
        name: name of assistant
        instructions: instructions for assistant
        tools: list of tools
        openai_tools: list of openai tools
        thread_id: thread id
        model: model
        run_retrieve_sleep_time: run retrieve sleep time
        files: files
        instructions_prefix: instructions prefix
        callback_manager: callback manager
        verbose: verbose
        file_ids: list of file ids
        api_key: OpenAI API key

    """
    from openai import OpenAI

    # this is the set of openai tools
    # not to be confused with the tools we pass in for function calling
    openai_tools = openai_tools or []
    tools = tools or []
    tool_fns = [t.metadata.to_openai_tool() for t in tools]
    all_openai_tools = openai_tools + tool_fns

    # initialize client
    client = OpenAI(api_key=api_key)

    # process files
    files = files or []
    file_ids = file_ids or []

    file_dict = _process_files(client, files)

    # TODO: openai's typing is a bit sus
    all_openai_tools = cast(List[Any], all_openai_tools)
    assistant = client.beta.assistants.create(
        name=name,
        instructions=instructions,
        tools=cast(List[Any], all_openai_tools),
        model=model,
    )
    return cls(
        client,
        assistant,
        tools,
        callback_manager=callback_manager,
        thread_id=thread_id,
        instructions_prefix=instructions_prefix,
        file_dict=file_dict,
        run_retrieve_sleep_time=run_retrieve_sleep_time,
        verbose=verbose,
    )

from_existing classmethod

from_existing(assistant_id: str, tools: Optional[List[BaseTool]] = None, thread_id: Optional[str] = None, instructions_prefix: Optional[str] = None, run_retrieve_sleep_time: float = 0.1, callback_manager: Optional[CallbackManager] = None, api_key: Optional[str] = None, verbose: bool = False) -> OpenAIAssistantAgent

From existing assistant id.

Parameters:

Name	Type	Description	Default
assistant_id	str	id of assistant	required
tools	Optional[List[BaseTool]]	list of BaseTools Assistant can use	None
thread_id	Optional[str]	thread id	None
run_retrieve_sleep_time	float	run retrieve sleep time	0.1
instructions_prefix	Optional[str]	instructions prefix	None
callback_manager	Optional[CallbackManager]	callback manager	None
api_key	Optional[str]	OpenAI API key	None
verbose	bool	verbose	False

Source code in llama-index-integrations/agent/llama-index-agent-openai/llama_index/agent/openai/openai_assistant_agent.py

@classmethod
def from_existing(
    cls,
    assistant_id: str,
    tools: Optional[List[BaseTool]] = None,
    thread_id: Optional[str] = None,
    instructions_prefix: Optional[str] = None,
    run_retrieve_sleep_time: float = 0.1,
    callback_manager: Optional[CallbackManager] = None,
    api_key: Optional[str] = None,
    verbose: bool = False,
) -> "OpenAIAssistantAgent":
    """From existing assistant id.

    Args:
        assistant_id: id of assistant
        tools: list of BaseTools Assistant can use
        thread_id: thread id
        run_retrieve_sleep_time: run retrieve sleep time
        instructions_prefix: instructions prefix
        callback_manager: callback manager
        api_key: OpenAI API key
        verbose: verbose

    """
    from openai import OpenAI

    # initialize client
    client = OpenAI(api_key=api_key)

    # get assistant
    assistant = client.beta.assistants.retrieve(assistant_id)
    # assistant.tools is incompatible with BaseTools so have to pass from params

    return cls(
        client,
        assistant,
        tools=tools,
        callback_manager=callback_manager,
        thread_id=thread_id,
        instructions_prefix=instructions_prefix,
        run_retrieve_sleep_time=run_retrieve_sleep_time,
        verbose=verbose,
    )

reset

reset() -> None

Delete and create a new thread.

Source code in llama-index-integrations/agent/llama-index-agent-openai/llama_index/agent/openai/openai_assistant_agent.py

def reset(self) -> None:
    """Delete and create a new thread."""
    self._client.beta.threads.delete(self._thread_id)
    thread = self._client.beta.threads.create()
    thread_id = thread.id
    self._thread_id = thread_id

get_tools

get_tools(message: str) -> List[BaseTool]

Get tools.

Source code in llama-index-integrations/agent/llama-index-agent-openai/llama_index/agent/openai/openai_assistant_agent.py

def get_tools(self, message: str) -> List[BaseTool]:
    """Get tools."""
    return self._tools

upload_files

upload_files(files: List[str]) -> Dict[str, Any]

Upload files.

Source code in llama-index-integrations/agent/llama-index-agent-openai/llama_index/agent/openai/openai_assistant_agent.py

def upload_files(self, files: List[str]) -> Dict[str, Any]:
    """Upload files."""
    return _process_files(self._client, files)

add_message

add_message(message: str, file_ids: Optional[List[str]] = None, tools: Optional[List[Dict[str, Any]]] = None) -> Any

Add message to assistant.

Source code in llama-index-integrations/agent/llama-index-agent-openai/llama_index/agent/openai/openai_assistant_agent.py

def add_message(
    self,
    message: str,
    file_ids: Optional[List[str]] = None,
    tools: Optional[List[Dict[str, Any]]] = None,
) -> Any:
    """Add message to assistant."""
    attachments = format_attachments(file_ids=file_ids, tools=tools)
    return self._client.beta.threads.messages.create(
        thread_id=self._thread_id,
        role="user",
        content=message,
        attachments=attachments,
    )

run_assistant

run_assistant(instructions_prefix: Optional[str] = None) -> Tuple[Any, Dict]

Run assistant.

Source code in llama-index-integrations/agent/llama-index-agent-openai/llama_index/agent/openai/openai_assistant_agent.py

def run_assistant(
    self, instructions_prefix: Optional[str] = None
) -> Tuple[Any, Dict]:
    """Run assistant."""
    instructions_prefix = instructions_prefix or self._instructions_prefix
    run = self._client.beta.threads.runs.create(
        thread_id=self._thread_id,
        assistant_id=self._assistant.id,
        instructions=instructions_prefix,
    )
    from openai.types.beta.threads import Run

    run = cast(Run, run)

    sources = []
    while run.status in ["queued", "in_progress", "requires_action"]:
        run = self._client.beta.threads.runs.retrieve(
            thread_id=self._thread_id, run_id=run.id
        )
        if run.status == "requires_action":
            cur_tool_outputs = self._run_function_calling(run)
            sources.extend(cur_tool_outputs)

        time.sleep(self._run_retrieve_sleep_time)
    if run.status == "failed":
        raise ValueError(
            f"Run failed with status {run.status}.\n" f"Error: {run.last_error}"
        )
    return run, {"sources": sources}

arun_assistant async

arun_assistant(instructions_prefix: Optional[str] = None) -> Tuple[Any, Dict]

Run assistant.

Source code in llama-index-integrations/agent/llama-index-agent-openai/llama_index/agent/openai/openai_assistant_agent.py

async def arun_assistant(
    self, instructions_prefix: Optional[str] = None
) -> Tuple[Any, Dict]:
    """Run assistant."""
    instructions_prefix = instructions_prefix or self._instructions_prefix
    run = self._client.beta.threads.runs.create(
        thread_id=self._thread_id,
        assistant_id=self._assistant.id,
        instructions=instructions_prefix,
    )
    from openai.types.beta.threads import Run

    run = cast(Run, run)

    sources = []

    while run.status in ["queued", "in_progress", "requires_action"]:
        run = self._client.beta.threads.runs.retrieve(
            thread_id=self._thread_id, run_id=run.id
        )
        if run.status == "requires_action":
            cur_tool_outputs = await self._arun_function_calling(run)
            sources.extend(cur_tool_outputs)

        await asyncio.sleep(self._run_retrieve_sleep_time)
    if run.status == "failed":
        raise ValueError(
            f"Run failed with status {run.status}.\n" f"Error: {run.last_error}"
        )
    return run, {"sources": sources}