Source code for langchain.runnables.openai_functions
from operator import itemgetter
from typing import Any, Callable, List, Mapping, Optional, Union
from langchain_core.messages import BaseMessage
from langchain_core.output_parsers.openai_functions import JsonOutputFunctionsParser
from langchain_core.runnables import RouterRunnable, Runnable
from langchain_core.runnables.base import RunnableBindingBase
from typing_extensions import TypedDict
[docs]class OpenAIFunction(TypedDict):
"""ChatOpenAI的函数描述"""
name: str
"""函数的名称。"""
description: str
"""函数的描述。"""
parameters: dict
"""函数的参数。"""
[docs]class OpenAIFunctionsRouter(RunnableBindingBase[BaseMessage, Any]):
"""一个可运行的程序,用于路由到所选的函数。"""
functions: Optional[List[OpenAIFunction]]
def __init__(
self,
runnables: Mapping[
str,
Union[
Runnable[dict, Any],
Callable[[dict], Any],
],
],
functions: Optional[List[OpenAIFunction]] = None,
):
if functions is not None:
assert len(functions) == len(runnables)
assert all(func["name"] in runnables for func in functions)
router = (
JsonOutputFunctionsParser(args_only=False)
| {"key": itemgetter("name"), "input": itemgetter("arguments")}
| RouterRunnable(runnables)
)
super().__init__(bound=router, kwargs={}, functions=functions)