Skip to content

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
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
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
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
@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
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
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
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
@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
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
@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
356
357
358
359
360
361
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
363
364
365
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
367
368
369
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
371
372
373
374
375
376
377
378
379
380
381
382
383
384
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
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
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
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
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}