LoginSignup
1
2
@shimajiroxyz

【langchain】ChatOpenAIの入出力処理をざっくり理解する

Last updated at Posted at 2023-08-21

はじめに

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ライブラリとの関係についてざっくり理解できたのは、それなりに良かったと思います。いい勉強になりました。

1
2
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
1
2