はじめに

langchainは言語モデルの扱いを簡単にするためのラッパーライブラリです。今回は、ChatOpenAIというクラスの内部でどのような処理が行われているのが、入力と出力に対する処理の観点から追ってみました。

ChatOpenAIにChatMessage形式の入力を与えて、ChatMessage形式の出力を得ることができます。

from langchain.chat_models import ChatOpenAI

chat = ChatOpenAI()
from langchain.schema import (
    AIMessage,
    HumanMessage,
    SystemMessage
)

input = [HumanMessage(content="Translate this sentence from English to French: I love programming.")]
output = chat(input)
print(output)

AIMessage(content="J'adore programmer.", additional_kwargs={}, example=False)

また、公式のopenaiライブラリの入出力は以下です。

import openai
import dotenv
import os
dotenv.load_dotenv()
openai.api_key = os.getenv("OPENAI_API_KEY")

input = [{"role":"user", "content": "Translate this sentence from English to French: I love programming."}]

output = openai.ChatCompletion.create(
  model="gpt-3.5-turbo",
  messages = input
)

print(output)

{
  "id": "...",
  "object": "chat.completion",
  "created": 1692505499,
  "model": "gpt-3.5-turbo-0613",
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": "J'adore la programmation."
      },
      "finish_reason": "stop"
    }
  ],
  "usage": {
    "prompt_tokens": 19,
    "completion_tokens": 8,
    "total_tokens": 27
  }
}

ChatOpenAIも内部でopenai.ChatCompletion.createを使っていますので、前者のinputが校舎にinputに加工され、後者のoutputが前者のoutputに加工されているはずです。

この処理の流れを追ってみます。

方針

langchainのソースコードを読みます。バージョンはv0.0.268です。

入力

ChatOpenAIはList[BaseMessage]型をうけとります。これが、どのように加工されてopenai.ChatCompletion.createに渡されるかを見てみます。

ChatOpenAI

ChatOpenAIのソースコードを読みます。

ChatOpenAIでは以下のようにChatOpenAIのインスタンスにmessageを入力できます。

chat = ChatOpenAI()

input = [HumanMessage(content="Translate this sentence from English to French: I love programming.")]
output = chat(input)

このとき内部ではChatOpenAI.__call__が呼ばれています。
しかし、ChatOpenAIでは__call__が定義されていませんので、継承元のBaseChatModelで定義されていると思われます。

BaseChatModel

BaseChatModelのソースコードを見ます。

BaseChatModel.__call__を見ます。

    def __call__(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        callbacks: Callbacks = None,
        **kwargs: Any,
    ) -> BaseMessage:
        generation = self.generate(
            [messages], stop=stop, callbacks=callbacks, **kwargs
        ).generations[0][0]
        if isinstance(generation, ChatGeneration):
            return generation.message
        else:
            raise ValueError("Unexpected generation type")

シンプルに考えるために、messagesにのみ注目します。第一引数でList[BaseMessage]型のmessagesを受け取り、それをリストにして、self.generateに渡しています。リストにしているのは、バッチ処理に対応するためと思います。

BaseChatModel.generateを見ます。

    def generate(
        self,
        messages: List[List[BaseMessage]],
        stop: Optional[List[str]] = None,
        callbacks: Callbacks = None,
        *,
        tags: Optional[List[str]] = None,
        metadata: Optional[Dict[str, Any]] = None,
        **kwargs: Any,
    ) -> LLMResult:
        """Top Level call"""
        params = self._get_invocation_params(stop=stop, **kwargs)
        options = {"stop": stop}

        callback_manager = CallbackManager.configure(
            callbacks,
            self.callbacks,
            self.verbose,
            tags,
            self.tags,
            metadata,
            self.metadata,
        )
        run_managers = callback_manager.on_chat_model_start(
            dumpd(self), messages, invocation_params=params, options=options
        )
        results = []
        for i, m in enumerate(messages):
            try:
                results.append(
                    self._generate_with_cache(
                        m,
                        stop=stop,
                        run_manager=run_managers[i] if run_managers else None,
                        **kwargs,
                    )
                )
            except (KeyboardInterrupt, Exception) as e:
                if run_managers:
                    run_managers[i].on_llm_error(e)
                raise e
        flattened_outputs = [
            LLMResult(generations=[res.generations], llm_output=res.llm_output)
            for res in results
        ]
        llm_output = self._combine_llm_outputs([res.llm_output for res in results])
        generations = [res.generations for res in results]
        output = LLMResult(generations=generations, llm_output=llm_output)
        if run_managers:
            run_infos = []
            for manager, flattened_output in zip(run_managers, flattened_outputs):
                manager.on_llm_end(flattened_output)
                run_infos.append(RunInfo(run_id=manager.run_id))
            output.run = run_infos
        return output

List[List[BaseMessage]]型の第一引数messagesが、以下で一要素ずつ、つまりList[BaseMassage]型になってself._generate_with_cacheに渡されています。

        for i, m in enumerate(messages):
            try:
                results.append(
                    self._generate_with_cache(
                        m,
                        stop=stop,
                        run_manager=run_managers[i] if run_managers else None,
                        **kwargs,
                    )
                )

BaseChatModel._generate_with_cacheを見ます。

    def _generate_with_cache(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> ChatResult:
        new_arg_supported = inspect.signature(self._generate).parameters.get(
            "run_manager"
        )
        disregard_cache = self.cache is not None and not self.cache
        if langchain.llm_cache is None or disregard_cache:
            # This happens when langchain.cache is None, but self.cache is True
            if self.cache is not None and self.cache:
                raise ValueError(
                    "Asked to cache, but no cache found at `langchain.cache`."
                )
            if new_arg_supported:
                return self._generate(
                    messages, stop=stop, run_manager=run_manager, **kwargs
                )
            else:
                return self._generate(messages, stop=stop, **kwargs)
        else:
            llm_string = self._get_llm_string(stop=stop, **kwargs)
            prompt = dumps(messages)
            cache_val = langchain.llm_cache.lookup(prompt, llm_string)
            if isinstance(cache_val, list):
                return ChatResult(generations=cache_val)
            else:
                if new_arg_supported:
                    result = self._generate(
                        messages, stop=stop, run_manager=run_manager, **kwargs
                    )
                else:
                    result = self._generate(messages, stop=stop, **kwargs)
                langchain.llm_cache.update(prompt, llm_string, result.generations)
                return result

List[BaseMessage]型のまま、以下部分でself._generateに渡されています。

                if new_arg_supported:
                    result = self._generate(
                        messages, stop=stop, run_manager=run_manager, **kwargs
                    )

BaseChatModel._generateを見ます。

    @abstractmethod
    def _generate(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> ChatResult:
        """Top Level call"""

抽象メソッドです。

ChatOpenAI

継承先であるChatOpenAIを見ます。

ChatOpenAI._generateを見ます。

    def _generate(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        stream: Optional[bool] = None,
        **kwargs: Any,
    ) -> ChatResult:
        if stream if stream is not None else self.streaming:
            generation: Optional[ChatGenerationChunk] = None
            for chunk in self._stream(
                messages=messages, stop=stop, run_manager=run_manager, **kwargs
            ):
                if generation is None:
                    generation = chunk
                else:
                    generation += chunk
            assert generation is not None
            return ChatResult(generations=[generation])

        message_dicts, params = self._create_message_dicts(messages, stop)
        params = {**params, **kwargs}
        response = self.completion_with_retry(
            messages=message_dicts, run_manager=run_manager, **params
        )
        return self._create_chat_result(response)

List[BaseMessage]型の第一引数messagesがself._create_message_dictsでmessages_dictsに変換され、self.completion_with_retryに渡されています。
またself._create_message_dictsはparamsという値も返していますが、これはおそらく、BaseChatModelのopenaiに関するクラス変数（おそらくBaseChatModelのインスタンス化時に定義されたもの）を返していると思われます。

        message_dicts, params = self._create_message_dicts(messages, stop)
        params = {**params, **kwargs}
        response = self.completion_with_retry(
            messages=message_dicts, run_manager=run_manager, **params
        )

ChatOpenAI._create_message_dictsを見ます。

    def _create_message_dicts(
        self, messages: List[BaseMessage], stop: Optional[List[str]]
    ) -> Tuple[List[Dict[str, Any]], Dict[str, Any]]:
        params = self._client_params
        if stop is not None:
            if "stop" in params:
                raise ValueError("`stop` found in both the input and default params.")
            params["stop"] = stop
        message_dicts = [convert_message_to_dict(m) for m in messages]
        return message_dicts, params

入力がList[BaseMessage]、出力がTuple[List[Dict[str, Any]], Dict[str, Any]]です。出力の要素1つめはopenai.ChatCompletion.createに渡すmessages、要素2つめはその他のパラメータといったところでしょうか。messagesの変換には、convert_message_to_dictを使っています。

convert_message_to_dictを見ます。

def convert_message_to_dict(message: BaseMessage) -> dict:
    message_dict: Dict[str, Any]
    if isinstance(message, ChatMessage):
        message_dict = {"role": message.role, "content": message.content}
    elif isinstance(message, HumanMessage):
        message_dict = {"role": "user", "content": message.content}
    elif isinstance(message, AIMessage):
        message_dict = {"role": "assistant", "content": message.content}
        if "function_call" in message.additional_kwargs:
            message_dict["function_call"] = message.additional_kwargs["function_call"]
            # If function call only, content is None not empty string
            if message_dict["content"] == "":
                message_dict["content"] = None
    elif isinstance(message, SystemMessage):
        message_dict = {"role": "system", "content": message.content}
    elif isinstance(message, FunctionMessage):
        message_dict = {
            "role": "function",
            "content": message.content,
            "name": message.name,
        }
    else:
        raise TypeError(f"Got unknown type {message}")
    if "name" in message.additional_kwargs:
        message_dict["name"] = message.additional_kwargs["name"]
    return message_dict

isinstanceを使って、BaseMessageのインスタンスの型を判定し、対応するopenai.ChatCompletion.createに入力可能な辞書に変換しています。
なお、HumanMessageとSystemMessageはcontentだけが使われますが、AssistantMessageの場合は、additional_kwargsにfunction_callがあれば、それも正しく変換してくれるようです。

ということで、ChatOpenAI.completion_with_retryに入力する直前にBaseMessageをOpenAI API形式の辞書に変換していることがわかりました。

このまま行けるところまで処理を追います。
ChatOpenAI.completion_with_retryを見ます。

    def completion_with_retry(
        self, run_manager: Optional[CallbackManagerForLLMRun] = None, **kwargs: Any
    ) -> Any:
        """Use tenacity to retry the completion call."""
        retry_decorator = _create_retry_decorator(self, run_manager=run_manager)

        @retry_decorator
        def _completion_with_retry(**kwargs: Any) -> Any:
            return self.client.create(**kwargs)

        return _completion_with_retry(**kwargs)

引数はrun_managerとその他のキーワード引数です。
ChatOpenAI._generateでは以下の形式で、completion_with_retryに引数を渡していたので、messagesも合わせて**kwargsに含まれていることがわかります。

        message_dicts, params = self._create_message_dicts(messages, stop)
        params = {**params, **kwargs}
        response = self.completion_with_retry(
            messages=message_dicts, run_manager=run_manager, **params
        )

この**kwargsをself.client.createに渡しています。

return self.client.create(**kwargs)

self.clientはopenai.ChatCompletionなので（参考）、実質、openai.ChatCompletion.createに渡しており、その出力がそのままChatOpenAI.completion_with_retryの戻り値になります。

出力

入力を追うことで、ChatOpenAI.completion_with_retryの戻り値とopenai.ChatCompletion.createの戻り値が同じであることがわかりました。

つまり

{
  "id": "...",
  "object": "chat.completion",
  "created": 1692505499,
  "model": "gpt-3.5-turbo-0613",
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": "J'adore la programmation."
      },
      "finish_reason": "stop"
    }
  ],
  "usage": {
    "prompt_tokens": 19,
    "completion_tokens": 8,
    "total_tokens": 27
  }
}

のような形式です。
これがどのように、ChatOpenAIの戻り値になっていくのか、処理を追います。

ChatOpenAI

ChatOpenAIを見ます。

ChatOpenAI.completion_with_retryを呼んでいるChatOpenAI._generateを見ます。

    def _generate(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        stream: Optional[bool] = None,
        **kwargs: Any,
    ) -> ChatResult:
        """略"""
        message_dicts, params = self._create_message_dicts(messages, stop)
        params = {**params, **kwargs}
        response = self.completion_with_retry(
            messages=message_dicts, run_manager=run_manager, **params
        )
        return self._create_chat_result(response)

responseをself._create_chat_resultに渡しています。

ChatOpenAI._create_chat_resultを見ます。

    def _create_chat_result(self, response: Mapping[str, Any]) -> ChatResult:
        generations = []
        for res in response["choices"]:
            message = convert_dict_to_message(res["message"])
            gen = ChatGeneration(
                message=message,
                generation_info=dict(finish_reason=res.get("finish_reason")),
            )
            generations.append(gen)
        token_usage = response.get("usage", {})
        llm_output = {"token_usage": token_usage, "model_name": self.model_name}
        return ChatResult(generations=generations, llm_output=llm_output)

response["choices"][i]["message"]（i=0,1,...)を取り出し、convert_dict_to_messageで変換しています。またその他の情報（token_usageなど）も合わせて取得しています。これをChatResult型として戻しています。

まずconvert_dict_to_messageを見ます。

def convert_dict_to_message(_dict: Mapping[str, Any]) -> BaseMessage:
    role = _dict["role"]
    if role == "user":
        return HumanMessage(content=_dict["content"])
    elif role == "assistant":
        # Fix for azure
        # Also OpenAI returns None for tool invocations
        content = _dict.get("content", "") or ""
        if _dict.get("function_call"):
            additional_kwargs = {"function_call": dict(_dict["function_call"])}
        else:
            additional_kwargs = {}
        return AIMessage(content=content, additional_kwargs=additional_kwargs)
    elif role == "system":
        return SystemMessage(content=_dict["content"])
    elif role == "function":
        return FunctionMessage(content=_dict["content"], name=_dict["name"])
    else:
        return ChatMessage(content=_dict["content"], role=role)

各messageをroleに対応するBaseMessageに変換しています。assistantの場合は、function_callをAIMessageのadditional_kwargsに含めています。

ChatOpenAI._create_chat_resultに戻ります。次はmessageとfinish_reasonをChatGenerationに渡しているようです。

            gen = ChatGeneration(
                message=message,
                generation_info=dict(finish_reason=res.get("finish_reason")),
            )

ChatGnerationを見ます。

class ChatGeneration(Generation):
    """A single chat generation output."""

    text: str = ""
    """*SHOULD NOT BE SET DIRECTLY* The text contents of the output message."""
    message: BaseMessage
    """The message output by the chat model."""

    @root_validator
    def set_text(cls, values: Dict[str, Any]) -> Dict[str, Any]:
        """Set the text attribute to be the contents of the message."""
        values["text"] = values["message"].content
        return values

set_textでmessageのcontentをtext変数にコピーしているようです。
他の情報がありませんので、継承元のGenerationを見てみます。

class Generation(Serializable):
    """A single text generation output."""

    text: str
    """Generated text output."""

    generation_info: Optional[Dict[str, Any]] = None
    """Raw response from the provider. May include things like the 
        reason for finishing or token log probabilities.
    """
    # TODO: add log probs as separate attribute

    @property
    def lc_serializable(self) -> bool:
        """Whether this class is LangChain serializable."""
        return True

こちらもあまり情報がありませんが、とりあえずtextとgeneration_infoという変数を持つことが期待されているようです。

ChatOpenAI._create_chat_resultに戻ります。ChatGenerationがChatResultに渡されています。

return ChatResult(generations=generations, llm_output=llm_output)

ChatResultを見ます。

class ChatResult(BaseModel):
    """Class that contains all results for a single chat model call."""

    generations: List[ChatGeneration]
    """List of the chat generations. This is a List because an input can have multiple 
        candidate generations.
    """
    llm_output: Optional[dict] = None
    """For arbitrary LLM provider specific output."""

pydanticのBaseModelを継承したクラスで、シンプルにgenerationsとllm_outputをもつ、という定義だけされています。

ということで、ChatOpenAI.completion_with_retryの戻り値はChatOpenAI._generateでChatResultに変換されていることがわかりました。

ChatOpenAI._generateを呼び出しているChatOpenAI._generate_with_cacheは継承元のBaseChatModelで定義されているので、そちらを見ます。

BaseChatModel

BaseChatModel._generate_with_cacheを見ます。

    def _generate_with_cache(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        run_manager: Optional[CallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> ChatResult:
            """略"""
            if isinstance(cache_val, list):
                return ChatResult(generations=cache_val)
            else:
                if new_arg_supported:
                    result = self._generate(
                        messages, stop=stop, run_manager=run_manager, **kwargs
                    )
                else:
                    result = self._generate(messages, stop=stop, **kwargs)
                langchain.llm_cache.update(prompt, llm_string, result.generations)
                return result

基本的に、self._generateの結果がそのまま戻されていることがわかります。
self._generateを呼び出しているself.generateを見ます。

    def generate(
        self,
        messages: List[List[BaseMessage]],
        stop: Optional[List[str]] = None,
        callbacks: Callbacks = None,
        *,
        tags: Optional[List[str]] = None,
        metadata: Optional[Dict[str, Any]] = None,
        **kwargs: Any,
    ) -> LLMResult:
        """略"""
        results = []
        for i, m in enumerate(messages):
            try:
                results.append(
                    self._generate_with_cache(
                        m,
                        stop=stop,
                        run_manager=run_managers[i] if run_managers else None,
                        **kwargs,
                    )
                )
            except (KeyboardInterrupt, Exception) as e:
                if run_managers:
                    run_managers[i].on_llm_error(e)
                raise e
        """略"""
        llm_output = self._combine_llm_outputs([res.llm_output for res in results])
        generations = [res.generations for res in results]
        output = LLMResult(generations=generations, llm_output=llm_output)
        """略"""
        return output

self._generate_with_cacheの戻り値（ChatResult型）はresults変数に追加されます。
そして、LLMResult型に変換されます。

LLMResultはChatResultと同様にgenerationsとllm_output、そして追加でllm実行時の情報をrunとして格納できるようです。以下、宣言冒頭だけ抜粋です。

class LLMResult(BaseModel):
    """Class that contains all results for a batched LLM call."""

    generations: List[List[Generation]]
    """List of generated outputs. This is a List[List[]] because
    each input could have multiple candidate generations."""
    llm_output: Optional[dict] = None
    """Arbitrary LLM provider-specific output."""
    run: Optional[List[RunInfo]] = None
    """List of metadata info for model call for each input."""

self.generateを呼び出しているBaseChatModel.__call__を見ます。

    def __call__(
        self,
        messages: List[BaseMessage],
        stop: Optional[List[str]] = None,
        callbacks: Callbacks = None,
        **kwargs: Any,
    ) -> BaseMessage:
        generation = self.generate(
            [messages], stop=stop, callbacks=callbacks, **kwargs
        ).generations[0][0]
        if isinstance(generation, ChatGeneration):
            return generation.message
        else:
            raise ValueError("Unexpected generation type")

self.generateの実行結果のgenerations要素の[0][0]を取得しています。
BaseChatModel.__call__ではList[BaseMessage]を1つだけ渡しているので、generateions[0]でmessagesに対応する生成結果であり、generateions[0][0]がその1つめのchoiceということになります。

そして以下でそのgenerationのmessageを返しています。openai.ChatCompletion.createの戻り値と対応付けるとresponse["choices"][0]["message"]相当のものです。

        if isinstance(generation, ChatGeneration):
            return generation.message

よってBaseMessage(AIMessage)がChatOpenAIにpromptを与えた結果として得られることがわかりました。

入出力のサマリ

処理の流れとinput、outputの型の変化をまとめます。

入力

ChatOpenAI.__call__に入力されたBaseMessage型のmessageは、ChatOpenAI._generateで、openai.ChatCompletion.createに入力可能なdict型に変換されます。
プロンプト以外のキーワード引数もそのままopenai.ChatCompletion.createに渡されます。

初期入力: BaseMessage (正確にはList[BaseMessage)
BaseMessage型
- ChatOpenAI.__call__ (BaseChatModel.__call__)
- BaseChatModel.generate
- BaseChatModel._generate_with_cache
- ChatOpenAI._generate
dict型
- ChatOpenAI.completion_with_retry

出力

ChatOpenAI.completion_with_retryで得られたdict型のmessageは、message以外の出力情報とともにChatResultに格納され、LLMResultに変換され、またmessage部分だけが取り出されて最終的にBaseMessageとして戻ります。

dict型
- ChatOpenAI.completion_with_retry
- ChatOpenAI._generate
- ChatOpenAI._create_chat_result
  - ChatOpenAI._convert_dict_to_messageでdictのうちmessage部分をBaseMessageに変換
  - BaseMessageと生成情報（finish_reasonのみ？)をChatGeneration型に格納
  - ChatGenerationとmessage以外の出力情報を合わせて、ChatResult型に格納
ChatResult型
- BaseChatModel._generate_with_cache
- BaseChatModel.generate
  - LLMResult型に変換
LLMResult型
- BaseChatModel.__call__
  - LLMResult.generations[0][0].message ( BaseMessage型 ) を取得して返す
最終出力: BaseMessage

実験

理解の確認のため、ChatOpenAIを実際に使ってみます。

パラメータの指定

message以外のキーワード引数は__call__時にopenaiに入力する形式で渡せばよいです。

from langchain.chat_models import ChatOpenAI
from langchain.schema import (
    AIMessage,
    HumanMessage,
    SystemMessage
)

chat = ChatOpenAI()
print(chat([HumanMessage(content="こんにちは")], max_tokens=5))

content='こんにちは！いつも' additional_kwargs={} example=False

文章が途中で切れておりmax_tokens=5が効いていることがわかります。

パラメータはインスタンス化時に指定することもできます。

chat = ChatOpenAI(max_tokens=5)
print(chat([HumanMessage(content="こんにちは")]))

content='こんにちは！ご用件' additional_kwargs={} example=False

インスタンス化と__call__の両方で指定した場合は、__call__の入力が優先されます。

chat = ChatOpenAI(max_tokens=100)
print(chat([HumanMessage(content="こんにちは")], max_tokens=5))

content='こんにちは！いつも' additional_kwargs={} example=False

chat = ChatOpenAI(max_tokens=5)
print(chat([HumanMessage(content="こんにちは")], max_tokens=100))

content='こんにちは！お元気ですか？何かお手伝いできることはありますか？' additional_kwargs={} example=False

function calling

ChatOpenAIでfunction callingしてみます。これもインスタンス化または__call__のときにキーワード引数を入れるだけです。結果はadditional_kwargsから取り出せます。

from langchain.chat_models import ChatOpenAI
from langchain.schema import (
    AIMessage,
    HumanMessage,
    SystemMessage
)
import json

functions = [
    # AIが、質問に対してこの関数を使うかどうか、
    # また使う時の引数は何にするかを判断するための情報を与える
    {
        "name": "Person",
        "description": "人物の情報を得る",
        "parameters": {
            "type": "object",
            "properties": {
                "name": {
                    "type": "string",
                    "description": "人物の名前",
                },
                "age": {
                    "type": "string",
                    "description": "人物の年齢",
                },
            },
            "required": ["name", "age"],
        },
    }
]
chat = ChatOpenAI()
response = chat([HumanMessage(content="太郎は３０歳です。")], functions=functions, function_call={"name": "Person"})
print(response)
arguments = json.loads(response.additional_kwargs["function_call"]["arguments"])
print(arguments)

content='' additional_kwargs={'function_call': {'name': 'Person', 'arguments': '{\n  "name": "太郎",\n  "age": "30"\n}'}} example=False
{'name': '太郎', 'age': '30'}

できました。

openai.ChatCompletion.createとほぼ同じ感覚で使えてよいですね。

サードパーティのopenai_function_callと組み合わせて使ってみます。

pip install instructor

from langchain.chat_models import ChatOpenAI
from langchain.schema import (
    AIMessage,
    HumanMessage,
    SystemMessage
)
import json

from instructor import OpenAISchema
from pydantic import Field

class UserDetails(OpenAISchema):
    "Correctly extracted user information"
    name: str = Field(..., description="User's full name")
    age: int

chat = ChatOpenAI()
response = chat([HumanMessage(content="太郎は３０歳です。")]
                , functions=[UserDetails.openai_schema]
                , function_call={"name": UserDetails.openai_schema["name"]})
print(response)
arguments = json.loads(response.additional_kwargs["function_call"]["arguments"])
print(arguments)

content='' additional_kwargs={'function_call': {'name': 'UserDetails', 'arguments': '{\n  "name": "太郎",\n  "age": 30\n}'}} example=False
{'name': '太郎', 'age': 30}

できました。
まあfunction callingに関しては、langchainにもラッパーがあるのでそっちを使ったほうが良いでしょうが、pydanticのバージョンを落とさないと現状使えなかったりするので、使いどきは一応あるかもしれません。

おわりに

ChatOpenAIを題材に、入出力がどのように加工されているのか、ソースコードを追ってみました。
langchainはchainの機能よりも、LLMのラッパー部分だけ取り出して使うと便利かなと思ったりしています。そのときに、langchainのドキュメントが意外と不十分なので、ソースコードを読んで、大本のopenaiライブラリとの関係についてざっくり理解できたのは、それなりに良かったと思います。いい勉強になりました。