codekingpro/portable-devtools

1"""Schema definitions for representing agent actions, observations, and return values.2 3!!! warning4 5    The schema definitions are provided for backwards compatibility.6 7!!! warning8 9    New agents should be built using the10    [`langchain` library](https://pypi.org/project/langchain/), which provides a11    simpler and more flexible way to define agents.12 13    See docs on [building agents](https://docs.langchain.com/oss/python/langchain/agents).14 15Agents use language models to choose a sequence of actions to take.16 17A basic agent works in the following manner:18 191. Given a prompt an agent uses an LLM to request an action to take20    (e.g., a tool to run).212. The agent executes the action (e.g., runs the tool), and receives an observation.223. The agent returns the observation to the LLM, which can then be used to generate23    the next action.244. When the agent reaches a stopping condition, it returns a final return value.25 26The schemas for the agents themselves are defined in `langchain.agents.agent`.27"""28 29from __future__ import annotations30 31import json32from collections.abc import Sequence33from typing import Any, Literal34 35from langchain_core.load.serializable import Serializable36from langchain_core.messages import (37    AIMessage,38    BaseMessage,39    FunctionMessage,40    HumanMessage,41)42 43 44class AgentAction(Serializable):45    """Represents a request to execute an action by an agent.46 47    The action consists of the name of the tool to execute and the input to pass48    to the tool. The log is used to pass along extra information about the action.49    """50 51    tool: str52    """The name of the `Tool` to execute."""53 54    tool_input: str | dict55    """The input to pass in to the `Tool`."""56 57    log: str58    """Additional information to log about the action.59 60    This log can be used in a few ways. First, it can be used to audit what exactly the61    LLM predicted to lead to this `(tool, tool_input)`.62 63    Second, it can be used in future iterations to show the LLMs prior thoughts. This is64    useful when `(tool, tool_input)` does not contain full information about the LLM65    prediction (for example, any `thought` before the tool/tool_input).66    """67 68    type: Literal["AgentAction"] = "AgentAction"69 70    # Override init to support instantiation by position for backward compat.71    def __init__(self, tool: str, tool_input: str | dict, log: str, **kwargs: Any):72        """Create an `AgentAction`.73 74        Args:75            tool: The name of the tool to execute.76            tool_input: The input to pass in to the `Tool`.77            log: Additional information to log about the action.78        """79        super().__init__(tool=tool, tool_input=tool_input, log=log, **kwargs)80 81    @classmethod82    def is_lc_serializable(cls) -> bool:83        """`AgentAction` is serializable.84 85        Returns:86            `True`87        """88        return True89 90    @classmethod91    def get_lc_namespace(cls) -> list[str]:92        """Get the namespace of the LangChain object.93 94        Returns:95            `["langchain", "schema", "agent"]`96        """97        return ["langchain", "schema", "agent"]98 99    @property100    def messages(self) -> Sequence[BaseMessage]:101        """Return the messages that correspond to this action."""102        return _convert_agent_action_to_messages(self)103 104 105class AgentActionMessageLog(AgentAction):106    """Representation of an action to be executed by an agent.107 108    This is similar to `AgentAction`, but includes a message log consisting of109    chat messages.110 111    This is useful when working with `ChatModels`, and is used to reconstruct112    conversation history from the agent's perspective.113    """114 115    message_log: Sequence[BaseMessage]116    """Similar to log, this can be used to pass along extra information about what exact117    messages were predicted by the LLM before parsing out the `(tool, tool_input)`.118 119    This is again useful if `(tool, tool_input)` cannot be used to fully recreate the120    LLM prediction, and you need that LLM prediction (for future agent iteration).121 122    Compared to `log`, this is useful when the underlying LLM is a chat model (and123    therefore returns messages rather than a string).124    """125    # Ignoring type because we're overriding the type from AgentAction.126    # And this is the correct thing to do in this case.127    # The type literal is used for serialization purposes.128    type: Literal["AgentActionMessageLog"] = "AgentActionMessageLog"  # type: ignore[assignment]129 130 131class AgentStep(Serializable):132    """Result of running an `AgentAction`."""133 134    action: AgentAction135    """The `AgentAction` that was executed."""136 137    observation: Any138    """The result of the `AgentAction`."""139 140    @property141    def messages(self) -> Sequence[BaseMessage]:142        """Messages that correspond to this observation."""143        return _convert_agent_observation_to_messages(self.action, self.observation)144 145 146class AgentFinish(Serializable):147    """Final return value of an `ActionAgent`.148 149    Agents return an `AgentFinish` when they have reached a stopping condition.150    """151 152    return_values: dict153    """Dictionary of return values."""154 155    log: str156    """Additional information to log about the return value.157 158    This is used to pass along the full LLM prediction, not just the parsed out159    return value.160 161    For example, if the full LLM prediction was `Final Answer: 2` you may want to just162    return `2` as a return value, but pass along the full string as a `log` (for163    debugging or observability purposes).164    """165    type: Literal["AgentFinish"] = "AgentFinish"166 167    def __init__(self, return_values: dict, log: str, **kwargs: Any):168        """Override init to support instantiation by position for backward compat."""169        super().__init__(return_values=return_values, log=log, **kwargs)170 171    @classmethod172    def is_lc_serializable(cls) -> bool:173        """Return `True` as this class is serializable."""174        return True175 176    @classmethod177    def get_lc_namespace(cls) -> list[str]:178        """Get the namespace of the LangChain object.179 180        Returns:181            `["langchain", "schema", "agent"]`182        """183        return ["langchain", "schema", "agent"]184 185    @property186    def messages(self) -> Sequence[BaseMessage]:187        """Messages that correspond to this observation."""188        return [AIMessage(content=self.log)]189 190 191def _convert_agent_action_to_messages(192    agent_action: AgentAction,193) -> Sequence[BaseMessage]:194    """Convert an agent action to a message.195 196    This code is used to reconstruct the original AI message from the agent action.197 198    Args:199        agent_action: Agent action to convert.200 201    Returns:202        `AIMessage` that corresponds to the original tool invocation.203    """204    if isinstance(agent_action, AgentActionMessageLog):205        return agent_action.message_log206    return [AIMessage(content=agent_action.log)]207 208 209def _convert_agent_observation_to_messages(210    agent_action: AgentAction, observation: Any211) -> Sequence[BaseMessage]:212    """Convert an agent action to a message.213 214    This code is used to reconstruct the original AI message from the agent action.215 216    Args:217        agent_action: Agent action to convert.218        observation: Observation to convert to a message.219 220    Returns:221        `AIMessage` that corresponds to the original tool invocation.222    """223    if isinstance(agent_action, AgentActionMessageLog):224        return [_create_function_message(agent_action, observation)]225    content = observation226    if not isinstance(observation, str):227        try:228            content = json.dumps(observation, ensure_ascii=False)229        except Exception:230            content = str(observation)231    return [HumanMessage(content=content)]232 233 234def _create_function_message(235    agent_action: AgentAction, observation: Any236) -> FunctionMessage:237    """Convert agent action and observation into a function message.238 239    Args:240        agent_action: the tool invocation request from the agent.241        observation: the result of the tool invocation.242 243    Returns:244        `FunctionMessage` that corresponds to the original tool invocation.245    """246    if not isinstance(observation, str):247        try:248            content = json.dumps(observation, ensure_ascii=False)249        except Exception:250            content = str(observation)251    else:252        content = observation253    return FunctionMessage(254        name=agent_action.tool,255        content=content,256    )257