LangChain とは

インストール

pip install langchain

uv add langchain

動作確認環境

$ cat ./pyproject.toml 

[project]
name = "llm-sample"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
    "langchain>=1.3.14",
    "langchain-openai>=1.4.0",
]

LangChain 利用による実装の違い

シンプルな質問

共通コード（デフォルト値と引数処理関数）

# common.py

import argparse
import os


DEFAULT_BASE_URL = "https://api.openai.com"
DEFAULT_PROMPT = "短く自己紹介してください。"
DEFAULT_SYSTEM_PROMPT = "You are a helpful assistant."
DEFAULT_MODEL = "gpt-4o"


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="OpenAI-compatible Chat Completions API を HTTP で呼び出すサンプルです。"
    )
    parser.add_argument(
        "--prompt",
        default=DEFAULT_PROMPT,
        help="user メッセージとして送るプロンプト",
    )
    parser.add_argument(
        "--system-prompt",
        default=DEFAULT_SYSTEM_PROMPT,
        help="system メッセージとして送るプロンプト",
    )
    parser.add_argument(
        "--model",
        default=os.getenv("OPENAI_MODEL", DEFAULT_MODEL),
        help="利用するモデル名。未指定時は OPENAI_MODEL または gpt-4o-mini",
    )
    parser.add_argument(
        "--base-url",
        default=os.getenv("OPENAI_BASE_URL", DEFAULT_BASE_URL),
        help="API のベース URL。未指定時は OPENAI_BASE_URL または https://api.openai.com",
    )
    return parser.parse_args()

http でリクエスト

OpenAI の REST API を叩く方式。

• body: {"model": "gpt-4o", "message": [{"role": "system", "content": "xxxxxx"}, {"role": "user", "content": "yyyyyy"}]}
• header: {"Authorization": "Bearer ", "Content-Type": "application/json"}
• method: POST

import json
import os
import sys
from pathlib import Path
from urllib import error, request

sys.path.insert(0, str(Path(__file__).resolve().parent))
from common import parse_args


def build_url(base_url: str) -> str:
    return f"{base_url.rstrip('/')}/v1/chat/completions"


def call_chat_completions(
    *,
    api_key: str,
    model: str,
    prompt: str,
    system_prompt: str,
    base_url: str,
) -> dict:
    payload = {
        "model": model,
        "messages": [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": prompt},
        ],
    }
    body = json.dumps(payload).encode("utf-8")
    req = request.Request(
        build_url(base_url),
        data=body,
        headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        },
        method="POST",
    )
    with request.urlopen(req) as response:
        return json.loads(response.read().decode("utf-8"))


def extract_text(response_json: dict) -> str:
    choices = response_json.get("choices") or []
    if not choices:
        return ""
    message = choices[0].get("message") or {}
    content = message.get("content", "")
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        texts = []
        for item in content:
            if item.get("type") == "text" and item.get("text"):
                texts.append(item["text"])
        return "".join(texts)
    return ""


def main() -> int:
    args = parse_args()
    api_key = os.getenv("OPENAI_PAT")
    response_json = call_chat_completions(
        api_key=api_key,
        model=args.model,
        prompt=args.prompt,
        system_prompt=args.system_prompt,
        base_url=args.base_url,
    )
    text = extract_text(response_json)
    print("=== assistant ===")
    print(text or "(empty response)")
    print()
    print("=== raw response ===")
    print(json.dumps(response_json, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

uv run openai-http.py

=== assistant ===
こんにちは！私はAIアシスタントで、質問に答えたり、情報を提供したり、さまざまなタスクを手助けするために作られました。よろしくお願いします！何かお手伝いできることがあれば教えてください。

=== raw response ===
{
  "id": "chatcmpl-E4cal7aEsaQCV4Y1SaQf3FDnwHaWF",
  "object": "chat.completion",
  "created": 1784770367,
  "model": "gpt-4o-2024-08-06",
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": "こんにちは！私はAIアシスタントで、質問に答えたり、情報を提供したり、さまざまなタスクを手助けするために作られました。よろしくお願いします！何かお手伝いできることがあれば教えてください。",
        "refusal": null,
        "annotations": []
      },
      "logprobs": null,
      "finish_reason": "stop"
    }
  ],
  "usage": {
    "prompt_tokens": 23,
    "completion_tokens": 60,
    "total_tokens": 83,
    "prompt_tokens_details": {
      "cached_tokens": 0,
      "audio_tokens": 0
    },
    "completion_tokens_details": {
      "reasoning_tokens": 0,
      "audio_tokens": 0,
      "accepted_prediction_tokens": 0,
      "rejected_prediction_tokens": 0
    }
  },
  "service_tier": "default",
  "system_fingerprint": "fp_8d8752de36"
}

LangChain でリクエスト

# openai-langchain.py

import argparse
import json
import os
import sys
from pathlib import Path
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI

sys.path.insert(0, str(Path(__file__).resolve().parent))
from common import parse_args


def build_url(base_url: str) -> str:
    return f"{base_url.rstrip('/')}/v1"


def main() -> int:
    args = parse_args()
    api_key = os.getenv("OPENAI_PAT")
    api_base_url = build_url(args.base_url)
    model = ChatOpenAI(
        api_key=api_key,
        base_url=api_base_url,
        model=args.model,
        temperature=0,
    )
    response = model.invoke(
        [
            SystemMessage(content=args.system_prompt),
            HumanMessage(content=args.prompt),
        ]
    )
    if isinstance(response.content, str):
        text = response.content
    else:
        text = "".join(
            item.get("text", "")
            for item in response.content
            if isinstance(item, dict) and item.get("type") == "text"
        )
    print("=== assistant ===")
    print(text or "(empty response)")
    print()
    print("=== vars(response) ===")
    print(json.dumps(vars(response), indent=2, ensure_ascii=False))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

uv run openai-langchain.py

=== assistant ===
こんにちは！私はAIアシスタントです。情報提供や質問への回答、日常のサポートなどを行います。何かお手伝いできることがあれば、お気軽にお知らせください！

=== vars(response) ===
{
  "content": "こんにちは！私はAIアシスタントです。情報提供や質問への回答、日常のサポートなどを行います。何かお手伝いできることがあれば、お気軽にお知らせください！",
  "additional_kwargs": {
    "refusal": null
  },
  "response_metadata": {
    "token_usage": {
      "completion_tokens": 48,
      "prompt_tokens": 23,
      "total_tokens": 71,
      "completion_tokens_details": {
        "accepted_prediction_tokens": 0,
        "audio_tokens": 0,
        "reasoning_tokens": 0,
        "rejected_prediction_tokens": 0
      },
      "prompt_tokens_details": {
        "audio_tokens": 0,
        "cached_tokens": 0
      }
    },
    "model_name": "gpt-4o-2024-08-06",
    "system_fingerprint": "fp_18164bb66d",
    "id": "chatcmpl-EOwfwIxyq6VFtfJh0TJoR5uq4zgoA",
    "service_tier": "default",
    "finish_reason": "stop",
    "logprobs": null
  },
  "type": "ai",
  "name": null,
  "id": "run--01a0ad50-7a98-7263-aa00-1d742602b217-0",
  "example": false,
  "tool_calls": [],
  "invalid_tool_calls": [],
  "usage_metadata": {
    "input_tokens": 23,
    "output_tokens": 48,
    "total_tokens": 71,
    "input_token_details": {
      "audio": 0,
      "cache_read": 0
    },
    "output_token_details": {
      "audio": 0,
      "reasoning": 0
    }
  }
}

Structured Output

回答を格納させるクラス：

# common_structured_output.py
from pydantic import BaseModel, Field

class AnswerWithJustification(BaseModel):
    """ユーザーの質問への回答と、その回答の正当性を含むデータモデル"""
    answer: str = Field(description="ユーザーの質問への回答")
    justification: str = Field(description="回答の正当性を示す根拠")

http でリクエスト

# openai-so-http.py
import json
import os
import sys
import copy
from pathlib import Path
from urllib import error, request
from typing import Any
from pydantic import BaseModel

sys.path.insert(0, str(Path(__file__).resolve().parent))
from common import parse_args
from common_structured_output import AnswerWithJustification


def build_url(base_url: str) -> str:
    return f"{base_url.rstrip('/')}/v1/chat/completions"


def ensure_closed_object_schema(schema: Any) -> Any:
    if isinstance(schema, dict):
        normalized = copy.deepcopy(schema)
        if normalized.get("type") == "object":
            normalized.setdefault("additionalProperties", False)
            properties = normalized.get("properties", {})
            if isinstance(properties, dict):
                normalized["properties"] = {
                    key: ensure_closed_object_schema(value)
                    for key, value in properties.items()
                }
        if "$defs" in normalized and isinstance(normalized["$defs"], dict):
            normalized["$defs"] = {
                key: ensure_closed_object_schema(value)
                for key, value in normalized["$defs"].items()
            }
        if "items" in normalized:
            normalized["items"] = ensure_closed_object_schema(normalized["items"])
        for key in ("anyOf", "allOf", "oneOf"):
            if key in normalized and isinstance(normalized[key], list):
                normalized[key] = [
                    ensure_closed_object_schema(item) for item in normalized[key]
                ]
        return normalized
    if isinstance(schema, list):
        return [ensure_closed_object_schema(item) for item in schema]
    return schema


def build_response_format(
    model_class: type[BaseModel],
    *,
    schema_name: str,
    schema_description: str,
) -> dict[str, Any]:
    schema = ensure_closed_object_schema(model_class.model_json_schema())
    return {
        "type": "json_schema",
        "json_schema": {
            "name": schema_name,
            "description": schema_description,
            "schema": schema,
            "strict": True,
        },
    }


def call_chat_completions(
    *,
    api_key: str,
    model: str,
    prompt: str,
    system_prompt: str,
    base_url: str,
) -> dict:
    payload = {
        "model": model,
        "messages": [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": prompt},
        ],
        "response_format": build_response_format(
            AnswerWithJustification,
            schema_name="answer_with_justification",
            schema_description="質問への回答とその根拠",
        ),
    }
    body = json.dumps(payload).encode("utf-8")
    req = request.Request(
        build_url(base_url),
        data=body,
        headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        },
        method="POST",
    )
    with request.urlopen(req) as response:
        return json.loads(response.read().decode("utf-8"))


def extract_text(response_json: dict) -> str:
    choices = response_json.get("choices") or []
    if not choices:
        return ""
    message = choices[0].get("message") or {}
    content = message.get("content", "")
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        texts = []
        for item in content:
            if item.get("type") == "text" and item.get("text"):
                texts.append(item["text"])
        return "".join(texts)
    return ""


def main() -> int:
    args = parse_args()
    api_key = os.getenv("OPENAI_PAT")
    response_json = call_chat_completions(
        api_key=api_key,
        model=args.model,
        prompt=args.prompt,
        system_prompt=args.system_prompt,
        base_url=args.base_url,
    )
    text = extract_text(response_json)
    print("=== assistant ===")
    print(text or "(empty response)")
    print()
    print("=== raw response ===")
    print(json.dumps(response_json, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

uv run openai-so-http.py --prompt "大阪から東京まで行くのに、鈍行電車と自動車だとどちらが早い？"

=== assistant ===
{"answer":"鈍行電車のほうが自動車より速く到着する可能性が高いです。","justification":"大阪から東京まで、鈍行電車を使って移動する場合、在来線を乗り継いで約9～10時間かかることが一般的です。自動車で移動する場合は、交通状況によりますが、通常の条件下では約7～9時間かかることがあります。しかし、交通渋滞が発生する可能性があるため、場合によっては自動車のほうが時間がかかることがあります。また、鈍行電車は定刻通りに運行することが多いため、時間の正確性においても鉄道の方が優位であることがあります。したがって、一般的には渋滞の影響を受けないことを考えると鈍行電車のほうが速いと考えられます。"}

=== raw response ===
{
  "id": "chatcmpl-EOxVlmrCo6QhzioIPpxS03dke831P",
  "object": "chat.completion",
  "created": 1789617341,
  "model": "gpt-4o-2024-08-06",
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": "{\"answer\":\"鈍行電車のほうが自動車より速く到着する可能性が高いです。\",\"justification\":\"大阪から東京まで、鈍行電車を使って移動する場合、在来線を乗り継いで約9～10時間かかることが一般的です。自動車で移動する場合は、交通状況によりますが、通常の条件下では約7～9時間かかることがあります。しかし、交通渋滞が発生する可能性があるため、場合によっては自動車のほうが時間がかかることがあります。また、鈍行電車は定刻通りに運行することが多いため、時間の正確性においても鉄道の方が優位であることがあります。したがって、一般的には渋滞の影響を受けないことを考えると鈍行電車のほうが速いと考えられます。\"}",
        "refusal": null,
        "annotations": []
      },
      "logprobs": null,
      "finish_reason": "stop"
    }
  ],
  "usage": {
    "prompt_tokens": 295,
    "completion_tokens": 226,
    "total_tokens": 521,
    "prompt_tokens_details": {
      "cached_tokens": 0,
      "audio_tokens": 0
    },
    "completion_tokens_details": {
      "reasoning_tokens": 0,
      "audio_tokens": 0,
      "accepted_prediction_tokens": 0,
      "rejected_prediction_tokens": 0
    }
  },
  "service_tier": "default",
  "system_fingerprint": "fp_29f19a4c7e"
}

LangChain でリクエスト

# openai-so-langchain.py
import argparse
import json
import os
import sys
from pathlib import Path
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI

sys.path.insert(0, str(Path(__file__).resolve().parent))
from common import parse_args
from common_structured_output import AnswerWithJustification


def build_url(base_url: str) -> str:
    return f"{base_url.rstrip('/')}/v1"


def main() -> int:
    args = parse_args()
    api_key = os.getenv("OPENAI_PAT")
    api_base_url = build_url(args.base_url)
    model = ChatOpenAI(
        api_key=api_key,
        base_url=api_base_url,
        model=args.model,
        temperature=0,
    )
    model_so = model.with_structured_output(AnswerWithJustification)
    response = model_so.invoke(
        [
            SystemMessage(content=args.system_prompt),
            HumanMessage(content=args.prompt),
        ]
    )
    if isinstance(response, AnswerWithJustification):
        print(response)
    print("=== vars(response) ===")
    print(json.dumps(vars(response), indent=2, ensure_ascii=False))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

uv run openai-so-langchain.py --prompt "大阪から東京まで行くのに、鈍行電車と自動車だとどちらが早い？"

answer='鈍行電車の方が自動車よりも早いです。' justification='大阪から東京までの距離は約500kmです。鈍行電車（普通列車）を利用する場合、乗り換えや待ち時間を含めても約9〜10時間程度で到着できます。一方、自動車で移動する場合、高速道路を利用しても渋滞や休憩を考慮すると約8〜10時間かかることが一般的です。したがって、条件によっては鈍行電車の方が早くなることがあります。'

=== vars(response) ===
{
  "answer": "鈍行電車の方が自動車よりも早いです。",
  "justification": "大阪から東京までの距離は約500kmです。鈍行電車（普通列車）を利用する場合、乗り換えや待ち時間を含めても約9〜10時間程度で到着できます。一方、自動車で移動する場合、高速道路を利用しても渋滞や休憩を考慮すると約8〜10時間かかることが一般的です。したがって、条件によっては鈍行電車の方が早くなることがあります。"
}

→ model.with_structured_output(MyClass) は、パース済みの構造化データのみを返し、トークン使用量などのメタデータを隠してしまう仕様 らしい。