Source code for langchain.agents.output_parsers.openai_tools
from typing import List, Union
from langchain_core.agents import AgentAction, AgentFinish
from langchain_core.messages import BaseMessage
from langchain_core.outputs import ChatGeneration, Generation
from langchain.agents.agent import MultiActionAgentOutputParser
from langchain.agents.output_parsers.tools import (
ToolAgentAction,
parse_ai_message_to_tool_action,
)
OpenAIToolAgentAction = ToolAgentAction
[docs]def parse_ai_message_to_openai_tool_action(
message: BaseMessage,
) -> Union[List[AgentAction], AgentFinish]:
"""解析一个可能包含工具调用的AI消息。"""
tool_actions = parse_ai_message_to_tool_action(message)
if isinstance(tool_actions, AgentFinish):
return tool_actions
final_actions: List[AgentAction] = []
for action in tool_actions:
if isinstance(action, ToolAgentAction):
final_actions.append(
OpenAIToolAgentAction(
tool=action.tool,
tool_input=action.tool_input,
log=action.log,
message_log=action.message_log,
tool_call_id=action.tool_call_id,
)
)
else:
final_actions.append(action)
return final_actions
[docs]class OpenAIToolsAgentOutputParser(MultiActionAgentOutputParser):
"""将消息解析为代理动作/完成。
旨在与OpenAI模型一起使用,因为它依赖于OpenAI的特定tool_calls参数来传达要使用的工具。
如果传递了tool_calls参数,则将用它来获取工具名称和工具输入。
如果没有传递参数,则假定AIMessage是最终输出。"""
@property
def _type(self) -> str:
return "openai-tools-agent-output-parser"
[docs] def parse_result(
self, result: List[Generation], *, partial: bool = False
) -> Union[List[AgentAction], AgentFinish]:
if not isinstance(result[0], ChatGeneration):
raise ValueError("This output parser only works on ChatGeneration output")
message = result[0].message
return parse_ai_message_to_openai_tool_action(message)
[docs] def parse(self, text: str) -> Union[List[AgentAction], AgentFinish]:
raise ValueError("Can only parse messages")