"""
**Agent** 是一个使用LLM来选择一系列动作的类。
在Chains中，一系列动作是硬编码的。在Agents中，语言模型被用作推理引擎，确定要采取哪些动作以及顺序。
Agents选择并使用 **Tools** 和 **Toolkits** 进行动作。
**类层次结构:**
.. code-block::
BaseSingleActionAgent --> LLMSingleActionAgent
OpenAIFunctionsAgent
XMLAgent
Agent --> <name>Agent # 例如: ZeroShotAgent, ChatAgent
BaseMultiActionAgent --> OpenAIMultiFunctionsAgent
**主要辅助功能:**
.. code-block::
AgentType, AgentExecutor, AgentOutputParser, AgentExecutorIterator,
AgentAction, AgentFinish, AgentStep
""" # noqa: E501
from __future__ import annotations
import json
from typing import Any, List, Literal, Sequence, Union
from langchain_core.load.serializable import Serializable
from langchain_core.messages import (
AIMessage,
BaseMessage,
FunctionMessage,
HumanMessage,
)
[docs]class AgentAction(Serializable):
"""一个ActionAgent执行的动作的完整描述。"""
tool: str
"""要执行的工具的名称。"""
tool_input: Union[str, dict]
"""工具的输入。"""
log: str
"""关于该操作的额外日志信息。
这个日志可以用几种方式。首先，它可以用于审计LLM预测导致这个（工具，工具输入）的确切情况。其次，它可以在未来的迭代中显示LLM之前的想法。当（工具，工具输入）不包含关于LLM预测的完整信息时，这是有用的（例如，在工具/工具输入之前的任何“想法”）。"""
type: Literal["AgentAction"] = "AgentAction"
def __init__(
self, tool: str, tool_input: Union[str, dict], log: str, **kwargs: Any
):
"""覆盖init以支持通过位置进行实例化，以实现向后兼容。"""
super().__init__(tool=tool, tool_input=tool_input, log=log, **kwargs)
[docs] @classmethod
def is_lc_serializable(cls) -> bool:
"""返回类是否可序列化。"""
return True
[docs] @classmethod
def get_lc_namespace(cls) -> List[str]:
"""获取langchain对象的命名空间。"""
return ["langchain", "schema", "agent"]
@property
def messages(self) -> Sequence[BaseMessage]:
"""返回与此操作对应的消息。"""
return _convert_agent_action_to_messages(self)
[docs]class AgentActionMessageLog(AgentAction):
message_log: Sequence[BaseMessage]
"""类似于日志，这可以用来传递关于LLM预测的确切消息的额外信息，在解析出(tool, tool_input)之前。如果(tool, tool_input)不能完全重现LLM预测，并且您需要该LLM预测（用于未来的代理迭代），则这是非常有用的。与`log`相比，当基础LLM是ChatModel（因此返回消息而不是字符串）时，这是有用的。"""
# Ignoring type because we're overriding the type from AgentAction.
# And this is the correct thing to do in this case.
# The type literal is used for serialization purposes.
type: Literal["AgentActionMessageLog"] = "AgentActionMessageLog" # type: ignore
[docs]class AgentStep(Serializable):
"""运行AgentAction的结果。"""
action: AgentAction
"""执行的AgentAction。"""
observation: Any
"""AgentAction的结果。"""
@property
def messages(self) -> Sequence[BaseMessage]:
"""返回与此观察相对应的消息。"""
return _convert_agent_observation_to_messages(self.action, self.observation)
[docs]class AgentFinish(Serializable):
"""ActionAgent的最终返回值。"""
return_values: dict
"""返回值的字典。"""
log: str
"""关于返回值的额外信息记录。
这用于传递完整的LLM预测，而不仅仅是解析出来的返回值。例如，如果完整的LLM预测是`Final Answer: 2`，您可能只想返回`2`作为返回值，但是将完整字符串作为`log`传递（用于调试或可观察性目的）。"""
type: Literal["AgentFinish"] = "AgentFinish"
def __init__(self, return_values: dict, log: str, **kwargs: Any):
"""覆盖init以支持通过位置进行实例化，以实现向后兼容。"""
super().__init__(return_values=return_values, log=log, **kwargs)
[docs] @classmethod
def is_lc_serializable(cls) -> bool:
"""返回类是否可序列化。"""
return True
[docs] @classmethod
def get_lc_namespace(cls) -> List[str]:
"""获取langchain对象的命名空间。"""
return ["langchain", "schema", "agent"]
@property
def messages(self) -> Sequence[BaseMessage]:
"""返回与此观察相对应的消息。"""
return [AIMessage(content=self.log)]
def _convert_agent_action_to_messages(
agent_action: AgentAction,
) -> Sequence[BaseMessage]:
"""将代理动作转换为消息。
此代码用于从代理动作重构原始的AI消息。
参数：
agent_action：要转换的代理动作。
返回：
对应于原始工具调用的AIMessage。
"""
if isinstance(agent_action, AgentActionMessageLog):
return agent_action.message_log
else:
return [AIMessage(content=agent_action.log)]
def _convert_agent_observation_to_messages(
agent_action: AgentAction, observation: Any
) -> Sequence[BaseMessage]:
"""将代理动作转换为消息。
此代码用于从代理动作重构原始的AI消息。
参数：
agent_action：要转换的代理动作。
返回：
对应于原始工具调用的AIMessage。
"""
if isinstance(agent_action, AgentActionMessageLog):
return [_create_function_message(agent_action, observation)]
else:
return [HumanMessage(content=observation)]
def _create_function_message(
agent_action: AgentAction, observation: Any
) -> FunctionMessage:
"""将代理动作和观察转换为函数消息。
参数：
agent_action：代理发送的工具调用请求
observation：工具调用的结果
返回：
对应于原始工具调用的FunctionMessage。
"""
if not isinstance(observation, str):
try:
content = json.dumps(observation, ensure_ascii=False)
except Exception:
content = str(observation)
else:
content = observation
return FunctionMessage(
name=agent_action.tool,
content=content,
)