Source code for langchain.agents.openai_assistant.base
from __future__ import annotations
import json
from json import JSONDecodeError
from time import sleep
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Sequence, Tuple, Union
from langchain_core.agents import AgentAction, AgentFinish
from langchain_core.load import dumpd
from langchain_core.pydantic_v1 import Field
from langchain_core.runnables import RunnableConfig, RunnableSerializable
from langchain_core.tools import BaseTool
from langchain.callbacks.manager import CallbackManager
from langchain.tools.render import format_tool_to_openai_tool
if TYPE_CHECKING:
    import openai
    from openai.types.beta.threads import ThreadMessage
    from openai.types.beta.threads.required_action_function_tool_call import (
        RequiredActionFunctionToolCall,
    )
[docs]class OpenAIAssistantFinish(AgentFinish):
    """AgentFinish with run and thread metadata."""
    run_id: str
    thread_id: str
[docs]    @classmethod
    def is_lc_serializable(cls) -> bool:
        return False
[docs]class OpenAIAssistantAction(AgentAction):
    """AgentAction with info needed to submit custom tool output to existing run."""
    tool_call_id: str
    run_id: str
    thread_id: str
[docs]    @classmethod
    def is_lc_serializable(cls) -> bool:
        return False
def _get_openai_client() -> openai.OpenAI:
    try:
        import openai
        return openai.OpenAI()
    except ImportError as e:
        raise ImportError(
            "Unable to import openai, please install with `pip install openai`."
        ) from e
    except AttributeError as e:
        raise AttributeError(