Source code for langchain_core.output_parsers.openai_tools
import copy
import json
from json import JSONDecodeError
from typing import Any, Dict, List, Optional, Type
from langchain_core.exceptions import OutputParserException
from langchain_core.messages import AIMessage, InvalidToolCall
from langchain_core.output_parsers import BaseCumulativeTransformOutputParser
from langchain_core.outputs import ChatGeneration, Generation
from langchain_core.pydantic_v1 import BaseModel, ValidationError
from langchain_core.utils.json import parse_partial_json
[docs]def parse_tool_call(
raw_tool_call: Dict[str, Any],
*,
partial: bool = False,
strict: bool = False,
return_id: bool = True,
) -> Optional[Dict[str, Any]]:
"""解析单个工具调用。"""
if "function" not in raw_tool_call:
return None
if partial:
try:
function_args = parse_partial_json(
raw_tool_call["function"]["arguments"], strict=strict
)
except (JSONDecodeError, TypeError): # None args raise TypeError
return None
else:
try:
function_args = json.loads(
raw_tool_call["function"]["arguments"], strict=strict
)
except JSONDecodeError as e:
raise OutputParserException(
f"Function {raw_tool_call['function']['name']} arguments:\n\n"
f"{raw_tool_call['function']['arguments']}\n\nare not valid JSON. "
f"Received JSONDecodeError {e}"
)
parsed = {
"name": raw_tool_call["function"]["name"] or "",
"args": function_args or {},
}
if return_id:
parsed["id"] = raw_tool_call.get("id")
return parsed
[docs]def make_invalid_tool_call(
raw_tool_call: Dict[str, Any],
error_msg: Optional[str],
) -> InvalidToolCall:
"""从原始工具调用创建一个InvalidToolCall。"""
return InvalidToolCall(
name=raw_tool_call["function"]["name"],
args=raw_tool_call["function"]["arguments"],
id=raw_tool_call.get("id"),
error=error_msg,
)
[docs]def parse_tool_calls(
raw_tool_calls: List[dict],
*,
partial: bool = False,
strict: bool = False,
return_id: bool = True,
) -> List[Dict[str, Any]]:
"""解析工具调用列表。"""
final_tools: List[Dict[str, Any]] = []
exceptions = []
for tool_call in raw_tool_calls:
try:
parsed = parse_tool_call(
tool_call, partial=partial, strict=strict, return_id=return_id
)
if parsed:
final_tools.append(parsed)
except OutputParserException as e:
exceptions.append(str(e))
continue
if exceptions:
raise OutputParserException("\n\n".join(exceptions))
return final_tools
[docs]class JsonOutputToolsParser(BaseCumulativeTransformOutputParser[Any]):
"""从OpenAI响应中解析工具。"""
strict: bool = False
"""是否允许非符合JSON规范的字符串。
参见：https://docs.python.org/3/library/json.html#encoders-and-decoders
在解析输出可能包含Unicode字符或换行符时很有用。"""
return_id: bool = False
"""是否返回工具调用ID。"""
first_tool_only: bool = False
"""是否仅返回第一个工具调用。
如果为False，则结果将是工具调用的列表，如果未找到工具调用，则为空列表。
如果为True，并且找到多个工具调用，则仅返回第一个工具调用，其他工具调用将被忽略。
如果未找到工具调用，则返回None。"""
[docs] def parse_result(self, result: List[Generation], *, partial: bool = False) -> Any:
generation = result[0]
if not isinstance(generation, ChatGeneration):
raise OutputParserException(
"This output parser can only be used with a chat generation."
)
message = generation.message
if isinstance(message, AIMessage) and message.tool_calls:
tool_calls = [dict(tc) for tc in message.tool_calls]
for tool_call in tool_calls:
if not self.return_id:
_ = tool_call.pop("id")
else:
try:
raw_tool_calls = copy.deepcopy(message.additional_kwargs["tool_calls"])
except KeyError:
return []
tool_calls = parse_tool_calls(
raw_tool_calls,
partial=partial,
strict=self.strict,
return_id=self.return_id,
)
# for backwards compatibility
for tc in tool_calls:
tc["type"] = tc.pop("name")
if self.first_tool_only:
return tool_calls[0] if tool_calls else None
return tool_calls
[docs]class JsonOutputKeyToolsParser(JsonOutputToolsParser):
"""从OpenAI响应中解析工具。"""
key_name: str
"""返回工具的类型。"""
[docs] def parse_result(self, result: List[Generation], *, partial: bool = False) -> Any:
parsed_result = super().parse_result(result, partial=partial)
if self.first_tool_only:
single_result = (
parsed_result
if parsed_result and parsed_result["type"] == self.key_name
else None
)
if self.return_id:
return single_result
elif single_result:
return single_result["args"]
else:
return None
parsed_result = [res for res in parsed_result if res["type"] == self.key_name]
if not self.return_id:
parsed_result = [res["args"] for res in parsed_result]
return parsed_result
[docs]class PydanticToolsParser(JsonOutputToolsParser):
"""从OpenAI响应中解析工具。"""
tools: List[Type[BaseModel]]
# TODO: Support more granular streaming of objects. Currently only streams once all
# Pydantic object fields are present.
[docs] def parse_result(self, result: List[Generation], *, partial: bool = False) -> Any:
json_results = super().parse_result(result, partial=partial)
if not json_results:
return None if self.first_tool_only else []
json_results = [json_results] if self.first_tool_only else json_results
name_dict = {tool.__name__: tool for tool in self.tools}
pydantic_objects = []
for res in json_results:
try:
if not isinstance(res["args"], dict):
raise ValueError(
f"Tool arguments must be specified as a dict, received: "
f"{res['args']}"
)
pydantic_objects.append(name_dict[res["type"]](**res["args"]))
except (ValidationError, ValueError) as e:
if partial:
continue
else:
raise e
if self.first_tool_only:
return pydantic_objects[0] if pydantic_objects else None
else:
return pydantic_objects