Source code for langchain.agents.openai_functions_multi_agent.base
"""Module implements an agent that uses OpenAI's APIs function enabled API."""
import json
from json import JSONDecodeError
from typing import Any, List, Optional, Sequence, Tuple, Union
from langchain.agents import BaseMultiActionAgent
from langchain.agents.format_scratchpad.openai_functions import (
    format_to_openai_functions,
)
from langchain.callbacks.base import BaseCallbackManager
from langchain.callbacks.manager import Callbacks
from langchain.chat_models.openai import ChatOpenAI
from langchain.prompts.chat import (
    BaseMessagePromptTemplate,
    ChatPromptTemplate,
    HumanMessagePromptTemplate,
    MessagesPlaceholder,
)
from langchain.pydantic_v1 import root_validator
from langchain.schema import (
    AgentAction,
    AgentFinish,
    BasePromptTemplate,
    OutputParserException,
)
from langchain.schema.agent import AgentActionMessageLog
from langchain.schema.language_model import BaseLanguageModel
from langchain.schema.messages import (
    AIMessage,
    BaseMessage,
    SystemMessage,
)
from langchain.tools import BaseTool
# For backwards compatibility
_FunctionsAgentAction = AgentActionMessageLog
def _parse_ai_message(message: BaseMessage) -> Union[List[AgentAction], AgentFinish]:
    """Parse an AI message."""
    if not isinstance(message, AIMessage):
        raise TypeError(f"Expected an AI message got {type(message)}")
    function_call = message.additional_kwargs.get("function_call", {})
    if function_call:
        try:
            tools = json.loads(function_call["arguments"])["actions"]
        except JSONDecodeError:
            raise OutputParserException(
                f"Could not parse tool input: {function_call} because "