Source code for langchain_core.utils.function_calling

"""在OpenAI Functions风格中创建函数规范的方法"""

from __future__ import annotations

import inspect
import uuid
from types import FunctionType, MethodType
from typing import (
    TYPE_CHECKING,
    Any,
    Callable,
    Dict,
    List,
    Literal,
    Optional,
    Tuple,
    Type,
    Union,
    cast,
)

from typing_extensions import TypedDict

from langchain_core._api import deprecated
from langchain_core.messages import (
    AIMessage,
    BaseMessage,
    HumanMessage,
    ToolMessage,
)
from langchain_core.pydantic_v1 import BaseModel
from langchain_core.utils.json_schema import dereference_refs

if TYPE_CHECKING:
    from langchain_core.tools import BaseTool

PYTHON_TO_JSON_TYPES = {
    "str": "string",
    "int": "integer",
    "float": "number",
    "bool": "boolean",
}


class FunctionDescription(TypedDict):
    """将可调用函数的表示发送到LLM。"""

    name: str
    """函数的名称。"""
    description: str
    """一个函数的描述。"""
    parameters: dict
    """函数的参数。"""


class ToolDescription(TypedDict):
    """代表一个可调用函数给OpenAI API。"""

    type: Literal["function"]
    function: FunctionDescription


def _rm_titles(kv: dict, prev_key: str = "") -> dict:
    new_kv = {}
    for k, v in kv.items():
        if k == "title":
            if isinstance(v, dict) and prev_key == "properties" and "title" in v.keys():
                new_kv[k] = _rm_titles(v, k)
            else:
                continue
        elif isinstance(v, dict):
            new_kv[k] = _rm_titles(v, k)
        else:
            new_kv[k] = v
    return new_kv


@deprecated(
    "0.1.16",
    alternative="langchain_core.utils.function_calling.convert_to_openai_function()",
    removal="0.3.0",
)
def convert_pydantic_to_openai_function(
    model: Type[BaseModel],
    *,
    name: Optional[str] = None,
    description: Optional[str] = None,
    rm_titles: bool = True,
) -> FunctionDescription:
    """将Pydantic模型转换为OpenAI API的函数描述。"""
    schema = dereference_refs(model.schema())
    schema.pop("definitions", None)
    title = schema.pop("title", "")
    default_description = schema.pop("description", "")
    return {
        "name": name or title,
        "description": description or default_description,
        "parameters": _rm_titles(schema) if rm_titles else schema,
    }


@deprecated(
    "0.1.16",
    alternative="langchain_core.utils.function_calling.convert_to_openai_tool()",
    removal="0.3.0",
)
def convert_pydantic_to_openai_tool(
    model: Type[BaseModel],
    *,
    name: Optional[str] = None,
    description: Optional[str] = None,
) -> ToolDescription:
    """将Pydantic模型转换为OpenAI API的函数描述。"""
    function = convert_pydantic_to_openai_function(
        model, name=name, description=description
    )
    return {"type": "function", "function": function}


def _get_python_function_name(function: Callable) -> str:
    """获取Python函数的名称。"""
    return function.__name__


def _parse_python_function_docstring(function: Callable) -> Tuple[str, dict]:
    """从函数的文档字符串中解析函数和参数描述。

假设函数文档字符串遵循Google Python风格指南。
"""
    docstring = inspect.getdoc(function)
    if docstring:
        docstring_blocks = docstring.split("\n\n")
        descriptors = []
        args_block = None
        past_descriptors = False
        for block in docstring_blocks:
            if block.startswith("Args:"):
                args_block = block
                break
            elif block.startswith("Returns:") or block.startswith("Example:"):
                # Don't break in case Args come after
                past_descriptors = True
            elif not past_descriptors:
                descriptors.append(block)
            else:
                continue
        description = " ".join(descriptors)
    else:
        description = ""
        args_block = None
    arg_descriptions = {}
    if args_block:
        arg = None
        for line in args_block.split("\n")[1:]:
            if ":" in line:
                arg, desc = line.split(":", maxsplit=1)
                arg_descriptions[arg.strip()] = desc.strip()
            elif arg:
                arg_descriptions[arg.strip()] += " " + line.strip()
    return description, arg_descriptions


def _get_python_function_arguments(function: Callable, arg_descriptions: dict) -> dict:
    """获取描述Python函数参数的JsonSchema。

假设所有函数参数都是基本类型（int，float，str，bool）或是pydantic.BaseModel的子类。
"""
    properties = {}
    annotations = inspect.getfullargspec(function).annotations
    for arg, arg_type in annotations.items():
        if arg == "return":
            continue
        if isinstance(arg_type, type) and issubclass(arg_type, BaseModel):
            # Mypy error:
            # "type" has no attribute "schema"
            properties[arg] = arg_type.schema()  # type: ignore[attr-defined]
        elif (
            hasattr(arg_type, "__name__")
            and getattr(arg_type, "__name__") in PYTHON_TO_JSON_TYPES
        ):
            properties[arg] = {"type": PYTHON_TO_JSON_TYPES[arg_type.__name__]}
        elif (
            hasattr(arg_type, "__dict__")
            and getattr(arg_type, "__dict__").get("__origin__", None) == Literal
        ):
            properties[arg] = {
                "enum": list(arg_type.__args__),  # type: ignore
                "type": PYTHON_TO_JSON_TYPES[arg_type.__args__[0].__class__.__name__],  # type: ignore
            }
        if arg in arg_descriptions:
            if arg not in properties:
                properties[arg] = {}
            properties[arg]["description"] = arg_descriptions[arg]
    return properties


def _get_python_function_required_args(function: Callable) -> List[str]:
    """获取Python函数所需的参数。"""
    spec = inspect.getfullargspec(function)
    required = spec.args[: -len(spec.defaults)] if spec.defaults else spec.args
    required += [k for k in spec.kwonlyargs if k not in (spec.kwonlydefaults or {})]

    is_function_type = isinstance(function, FunctionType)
    is_method_type = isinstance(function, MethodType)
    if required and is_function_type and required[0] == "self":
        required = required[1:]
    elif required and is_method_type and required[0] == "cls":
        required = required[1:]
    return required


@deprecated(
    "0.1.16",
    alternative="langchain_core.utils.function_calling.convert_to_openai_function()",
    removal="0.3.0",
)
def convert_python_function_to_openai_function(
    function: Callable,
) -> Dict[str, Any]:
    """将一个Python函数转换为与OpenAI函数调用API兼容的字典。

假设Python函数具有类型提示和带有描述的文档字符串。如果文档字符串具有Google Python风格的参数描述，这些描述也将被包括在内。
"""
    description, arg_descriptions = _parse_python_function_docstring(function)
    return {
        "name": _get_python_function_name(function),
        "description": description,
        "parameters": {
            "type": "object",
            "properties": _get_python_function_arguments(function, arg_descriptions),
            "required": _get_python_function_required_args(function),
        },
    }


@deprecated(
    "0.1.16",
    alternative="langchain_core.utils.function_calling.convert_to_openai_function()",
    removal="0.3.0",
)
def format_tool_to_openai_function(tool: BaseTool) -> FunctionDescription:
    """将工具格式化为OpenAI函数API。"""
    if tool.args_schema:
        return convert_pydantic_to_openai_function(
            tool.args_schema, name=tool.name, description=tool.description
        )
    else:
        return {
            "name": tool.name,
            "description": tool.description,
            "parameters": {
                # This is a hack to get around the fact that some tools
                # do not expose an args_schema, and expect an argument
                # which is a string.
                # And Open AI does not support an array type for the
                # parameters.
                "properties": {
                    "__arg1": {"title": "__arg1", "type": "string"},
                },
                "required": ["__arg1"],
                "type": "object",
            },
        }


@deprecated(
    "0.1.16",
    alternative="langchain_core.utils.function_calling.convert_to_openai_tool()",
    removal="0.3.0",
)
def format_tool_to_openai_tool(tool: BaseTool) -> ToolDescription:
    """将工具格式化为OpenAI函数API。"""
    function = format_tool_to_openai_function(tool)
    return {"type": "function", "function": function}


def convert_to_openai_function(
    function: Union[Dict[str, Any], Type[BaseModel], Callable, BaseTool],
) -> Dict[str, Any]:
    """将原始函数/类转换为OpenAI函数。

参数：
    function：可以是字典、pydantic.BaseModel类或Python函数。
        如果传入的是字典，则假定它已经是一个有效的OpenAI函数或带有顶级'title'和'description'键的JSON模式。

返回：
    传入函数的字典版本，与OpenAI函数调用API兼容。
"""
    from langchain_core.tools import BaseTool

    # already in OpenAI function format
    if isinstance(function, dict) and all(
        k in function for k in ("name", "description", "parameters")
    ):
        return function
    # a JSON schema with title and description
    elif isinstance(function, dict) and all(
        k in function for k in ("title", "description", "properties")
    ):
        function = function.copy()
        return {
            "name": function.pop("title"),
            "description": function.pop("description"),
            "parameters": function,
        }
    elif isinstance(function, type) and issubclass(function, BaseModel):
        return cast(Dict, convert_pydantic_to_openai_function(function))
    elif isinstance(function, BaseTool):
        return cast(Dict, format_tool_to_openai_function(function))
    elif callable(function):
        return convert_python_function_to_openai_function(function)
    else:
        raise ValueError(
            f"Unsupported function\n\n{function}\n\nFunctions must be passed in"
            " as Dict, pydantic.BaseModel, or Callable. If they're a dict they must"
            " either be in OpenAI function format or valid JSON schema with top-level"
            " 'title' and 'description' keys."
        )


def convert_to_openai_tool(
    tool: Union[Dict[str, Any], Type[BaseModel], Callable, BaseTool],
) -> Dict[str, Any]:
    """将原始函数/类转换为OpenAI工具。

参数：
    tool：可以是字典、pydantic.BaseModel类、Python函数或BaseTool。如果传入字典，则假定它已经是有效的OpenAI工具、OpenAI函数或带有顶级'title'和'description'键的JSON模式。

返回：
    传入工具的字典版本，与OpenAI工具调用API兼容。
"""
    if isinstance(tool, dict) and tool.get("type") == "function" and "function" in tool:
        return tool
    function = convert_to_openai_function(tool)
    return {"type": "function", "function": function}


def tool_example_to_messages(
    input: str, tool_calls: List[BaseModel], tool_outputs: Optional[List[str]] = None
) -> List[BaseMessage]:
    """将一个示例转换为可以输入到LLM中的消息列表。

这段代码是一个适配器，将单个示例转换为可以输入到聊天模型中的消息列表。

每个示例的消息列表对应于：

1) HumanMessage: 包含应从中提取内容的内容。
2) AIMessage: 包含从模型中提取的信息
3) ToolMessage: 包含确认消息，告知模型已正确请求了一个工具。

ToolMessage是必需的，因为一些聊天模型是针对代理而不是用于提取的情况进行了超优化。

参数：
    input: 字符串，用户输入
    tool_calls: List[BaseModel]，表示为Pydantic BaseModel的工具调用列表
    tool_outputs: Optional[List[str]]，工具调用输出的列表。
        不需要提供。如果未提供，将插入一个占位符值。

返回：
    一个消息列表

示例：

    .. code-block:: python

        from typing import List, Optional
        from langchain_core.pydantic_v1 import BaseModel, Field
        from langchain_openai import ChatOpenAI

        class Person(BaseModel):
            '''关于一个人的信息。'''
            name: Optional[str] = Field(..., description="人的姓名")
            hair_color: Optional[str] = Field(
                ..., description="如果已知，人的头发颜色"
            )
            height_in_meters: Optional[str] = Field(
                ..., description="以米为单位的身高"
            )

        examples = [
            (
                "海洋辽阔而蔚蓝。它超过了20,000英尺的深度。",
                Person(name=None, height_in_meters=None, hair_color=None),
            ),
            (
                "Fiona从法国远行到了西班牙。",
                Person(name="Fiona", height_in_meters=None, hair_color=None),
            ),
        ]


        messages = []

        for txt, tool_call in examples:
            messages.extend(
                tool_example_to_messages(txt, [tool_call])
            )
"""
    messages: List[BaseMessage] = [HumanMessage(content=input)]
    openai_tool_calls = []
    for tool_call in tool_calls:
        openai_tool_calls.append(
            {
                "id": str(uuid.uuid4()),
                "type": "function",
                "function": {
                    # The name of the function right now corresponds to the name
                    # of the pydantic model. This is implicit in the API right now,
                    # and will be improved over time.
                    "name": tool_call.__class__.__name__,
                    "arguments": tool_call.json(),
                },
            }
        )
    messages.append(
        AIMessage(content="", additional_kwargs={"tool_calls": openai_tool_calls})
    )
    tool_outputs = tool_outputs or ["You have correctly called this tool."] * len(
        openai_tool_calls
    )
    for output, tool_call_dict in zip(tool_outputs, openai_tool_calls):
        messages.append(ToolMessage(content=output, tool_call_id=tool_call_dict["id"]))  # type: ignore
    return messages