メッセージは LangChain におけるモデルのコンテキストの基本ユニットです。メッセージはモデルの入出力を表し、LLM とやり取りする際、会話の状態を表すのに必要なコンテンツとメタデータの両方をキャリーします。

LangChain 1.0 alpha : コアコンポーネント – メッセージ

作成 : クラスキャット・セールスインフォメーション
作成日時 : 09/23/2025
バージョン : 1.0.0a6

* 本記事は docs.langchain.com の以下のページを独自に翻訳した上で、補足説明を加えてまとめ直しています。スニペットはできる限り日本語を使用しています :

• LangChain OSS (v1-alpha) : Core Components – Messages

* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。

 

 

 

LangChain 1.0 alpha : コアコンポーネント – メッセージ

メッセージは LangChain におけるモデルのコンテキストの基本ユニットです。メッセージはモデルの入出力を表し、LLM とやり取りする際、会話の状態を表すのに必要なコンテンツとメタデータの両方をキャリーします。

メッセージは以下を含むオブジェクトです :

• ロール – メッセージ・タイプ (e.g., system、user) を識別します。

• コンテンツ – テキスト、画像、音声、ドキュメント 等のような、メッセージの実際の内容を表します。

• メタデータ – レスポンス情報、メッセージ ID、トークン使用量のようなオプションフィールド

LangChain は、すべてのモデルプロバイダーに渡り動作する標準的なメッセージタイプを提供し、呼び出されるモデルに関係なく一貫した動作を保証します。

 

基本的な使用方法

メッセージを使用する最も簡単な方法は、メッセージ・オブジェクトを作成して、モデルを呼び出すときにそれらを渡すことです。

from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

model = init_chat_model("openai:gpt-5-nano")

system_msg = SystemMessage("You are a helpful assistant.")
human_msg = HumanMessage("Hello, how are you?")

# Use with chat models
messages = [system_msg, human_msg]
response = model.invoke(messages)  # Returns AIMessage

出力例

Hi there! I’m here and ready to help with whatever you need. I don’t have feelings, but I’m ready to assist. What would you like help with today?

 

テキスト・プロンプト

テキストプロンプトは文字列です – 会話履歴を保持する必要のない、単純な生成タスクに最適です。

response = model.invoke("春について俳句を書いてください。")

出力例

もちろん。春をテーマにした俳句を3つ挙げます。

1) 春風や
　桜の下で
　鳥の声

2) 春雨や
　蛙が跳ねる
　春の風

3) 桜並木
　風に散る花
　春日かな

 
以下のような場合に、テキストプロンプトを使用します :

• 単一の、スタンドアローンなリクエストがある

• 会話履歴を必要としない

• 最小限のコードの複雑を望む

 

メッセージ・プロンプト

あるいは、メッセージ・オブジェクトのリストを提供することで、メッセージのリストをモデルに渡すこともできます。

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

messages = [
    SystemMessage("あなたは詩の専門家です。"),
    HumanMessage("春について俳句を作成してください。W"),
    AIMessage("桜🌸の花が咲きました ...")
]
response = model.invoke(messages)

出力例

以下、春をテーマにした俳句を3つご提案します。

1)
春風や
桜の花びらが
川を染める

2)
春の朝
霧の中に光る
蝶ひらり

3)
雪解けて
庭の小さな芽が
風に揺れる

ご希望の雰囲気や季節感があれば、別のスタイルでさらに作成します。

 
以下の場合、メッセージ・プロンプトを使用します :

• マルチターンの会話を管理

• マルチモーダル・コンテンツ (画像、音声、ファイル) を扱う

• システム指示を含む

 

辞書形式

OpenAI チャット completions 形式でメッセージを直接指定することもできます。

messages = [
    {"role": "system", "content": "あなたは詩の専門家です。"},
    {"role": "user", "content": "春について俳句を作成してください。"},
    {"role": "assistant", "content": "桜🌸の花が咲きました .."}
]
response = model.invoke(messages)

出力例

春を題材にした俳句をいくつかご用意しました。5-7-5の定型を意識しています。

- 春風や
  桜咲く頃
  川の音

- 春ひらく
  花びらゆらぐ
  朝露や

- 水辺へと
  鶯の声
  春来る日

どれも季語として「春」と自然の情景を取り入れています。お気に入りがあれば教えてください。別の雰囲気でも作成します。

 

メッセージ・タイプ

• System メッセージ – モデルに動作する方法を伝えて、インタラクションのコンテキストを提供します。

• Human メッセージ – ユーザ入力とモデルとのインタラクションを表します。

• AI メッセージ – テキストコンテンツ、ツール呼び出し、メタデータを含む、モデルにより生成されたレスポンス。

• Tool メッセージ – ツール呼び出しの出力を表します。

 

​System メッセージ

SystemMessage は、モデルの動作を準備する指示の初期セットを表します。雰囲気 (tone) を決め、モデルの役割りを定義し、レスポンスのガイドラインを確立するために system メッセージを使用できます。

Basic instructions

system_msg = SystemMessage("あなたは役立つコーディング・アシスタントです。")

messages = [
    system_msg,
    HumanMessage("REST API をどのように作成しますか?")
]
response = model.invoke(messages)

 
Detailed persona

from langchain_core.messages import SystemMessage, HumanMessage

system_msg = SystemMessage("""
あなたは Web フレームワークの専門知識を持つ上級 Python 開発者です。
必ずコード例を提示し、その理由を説明してください。
説明は簡潔でありながら完全に行ってください。
""")

messages = [
    system_msg,
    HumanMessage("REST API をどのように作成しますか?")
]
response = model.invoke(messages)

出力例

REST API を作成する基本的な道筋と、実装例を FastAPI を用いて示します。最新の Python で高性能・型安全・自動ドキュメントが得られる点が魅力です。

1) 要件定義と設計
- 何をリソースとして公開するかを決める（例: items, users, orders など）
- 各リソースに対して CRUD 操作を HTTP メソッドで設計する
  - Create: POST /items
  - Read: GET /items, GET /items/{id}
  - Update: PUT /items/{id}
  - Delete: DELETE /items/{id}
- バージョン管理（例: /api/v1/...）と適切なステータスコードを採用
- 入力検証・エラーハンドリング・認証・認可・データ検証を整備
- ドキュメント化（OpenAPI/Swaggerは自動生成が望ましい）

2) 技術選択と理由
- FastAPI を推奨：型ヒントを活用した検証・自動生成ドキュメント、非同期処理、パフォーマンス、初心者にも扱いやすい。
- ORM は SQLAlchemy 或いは Tortoise ORM など、データベースと連携してモデルを定義。
- テストには httpx（あるいは requests）と pytest。

3) プロジェクト構成の例
- app/
  - main.py            : FastAPI アプリの起動・ルーティング
  - api/
    - v1/
      - items.py       : アイテム関連のエンドポイント
  - models/
    - item.py          : SQLAlchemy の ORM モデル
  - schemas/
    - item.py          : Pydantic モデル（リクエスト/レスポンス用）
  - core/
    - config.py        : 設定（例: データベースURL）
  - db.py                : DB セッション管理
- tests/                 : テスト

4) 最小実装サンプル（FastAPI + in-memory store）
- ファイル名: main.py

```python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional

app = FastAPI(title="Sample API", version="1.0.0")

class Item(BaseModel):
    id: int
    name: str
    description: Optional[str] = None
    price: float

# インメモリストア（サンプル用）
_items: dict[int, Item] = {}

@app.post("/api/v1/items/", response_model=Item, status_code=201)
async def create_item(item: Item):
    if item.id in _items:
        raise HTTPException(status_code=400, detail="Item already exists")
    _items[item.id] = item
    return item

@app.get("/api/v1/items/", response_model=List[Item])
async def list_items():
    return list(_items.values())

@app.get("/api/v1/items/{item_id}", response_model=Item)
async def get_item(item_id: int):
    item = _items.get(item_id)
    if item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@app.put("/api/v1/items/{item_id}", response_model=Item)
async def update_item(item_id: int, item: Item):
    if item_id != item.id:
        raise HTTPException(status_code=400, detail="ID mismatch")
    if item_id not in _items:
        raise HTTPException(status_code=404, detail="Item not found")
    _items[item_id] = item
    return item

@app.delete("/api/v1/items/{item_id}", status_code=204)
async def delete_item(item_id: int):
    if item_id not in _items:
        raise HTTPException(status_code=404, detail="Item not found")
    del _items[item_id]
    return

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)
```

実行方法:
- 環境構築
  - pip install fastapi uvicorn pydantic
- 起動
  - uvicorn main:app --reload
- ドキュメント
  - http://localhost:8000/api/v1/items/ で自動生成された Swagger UI/ ReDoc が利用可能

5) データベース連携の基本（SQLAlchemy 例）
- 追加ファイルを用意して ORM で永続化する基本例

ファイル: db.py
```python
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

DATABASE_URL = "sqlite:///./test.db"  # 環境に合わせて変更

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()
```

ファイル: models/item.py
```python
from sqlalchemy import Column, Integer, String, Float
from .base import Base  # ここでは Base が db.py の Base を指す想定

class ItemModel(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=False)
    description = Column(String, nullable=True)
    price = Column(Float, nullable=False)
```

ファイル: schemas/item.py
```python
from pydantic import BaseModel
from typing import Optional

class ItemCreate(BaseModel):
    id: int
    name: str
    description: Optional[str] = None
    price: float
```

ファイル: main.py の簡易統合例（要調整）
```python
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from db import SessionLocal, Base
from models.item import ItemModel
from schemas.item import ItemCreate

Base.metadata.create_all(bind=engine)

app = FastAPI()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/api/v1/items/", response_model=ItemCreate)
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
    db_item = ItemModel(id=item.id, name=item.name, description=item.description, price=item.price)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return item
```

補足:
- 実運用では、インメモリストアは避け、必ずデータベースへ永続化します。上の例は最小構成のイメージです。
- Pydantic のスキーマと SQLAlchemy のモデルを分けると、入力検証と永続化をきれいに分離できます。

6) 認証とエラーハンドリングの基本
- 認証: OAuth2 Password Flow や JWT を導入
- 実装例（簡易）: JWT トークンを発行するエンドポイントと、保護されたエンドポイントの例
```python
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def create_access_token(data: dict):
    return jwt.encode(data, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")

@app.post("/api/v1/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # 実際にはユーザー確認を行う
    access_token = create_access_token({"sub": form_data.username})
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/api/v1/protected")
async def protected_route(user: dict = Depends(get_current_user)):
    return {"hello": user.get("sub")}
```

7) ドキュメントとテスト
- FastAPI は OpenAPI に基づく自動ドキュメントを提供
  - http://localhost:8000/docs（Swagger UI）
  - http://localhost:8000/redoc（ReDoc）
- テスト例（httpx + pytest）
```python
# tests/test_items.py
import httpx
import pytest

BASE = "http://localhost:8000/api/v1"

@pytest.mark.asyncio
async def test_create_and_get_item():
    async with httpx.AsyncClient() as client:
        r = await client.post(f"{BASE}/items/", json={"id":1,"name":"Sample","price":9.99})
        assert r.status_code == 201
        r = await client.get(f"{BASE}/items/1")
        assert r.status_code == 200
```

8) デプロイと運用のヒント
- ASGI サーバーで運用
  - uvicornや gunicorn + uvicorn workers の組み合わせ
  - 例: gunicorn -k uvicorn.workers.UvicornWorker main:app --bind 0.0.0.0:8000
- Docker 化の基本
  - Dockerfile:
    - FROM python:3.11-slim
    - WORKDIR /app
    - COPY requirements.txt .
    - RUN pip install -r requirements.txt
    - COPY . .
    - CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
  - docker-compose で DB 連携と API を同時に起動する構成がよく用いられます。
- 監視・運用
  - ログ、エラーレスポンス、レイテンシの監視を組み込み
  - API バージョンを使った後方互換性の維持
  - セキュリティ対策として TLS、入力検証、 rate limit、認証・権限の適切な設定

要点
- FastAPI は最速で実用的な REST API を作成でき、型安全・自動ドキュメント・非同期処理を同時に得られます。
- まずは小さなエンドポイントから始め、徐々にデータベース連携・認証・エラーハンドリングを拡張するのが現実的です。
- 本番運用ではデータベース・認証・デプロイ手順を明確に分離・自動化し、テストとドキュメントを常に最新に保つことが重要です。

必要であれば、あなたの用途（データベース種別、認証要件、公開 API の規模）に合わせた具体的なコード構成とファイルテンプレを作成します。どのフレームワークがよいか（FastAPI/DRF/Flask など）も目的に合わせて提案します。

 

Human メッセージ

HumanMessage はユーザの入力とインタラクションを表します。テキスト、画像、音声、ファイルと、その他のマルチモーダルなコンテンツを含むことができます。

 

テキストコンテンツ

​

Message オブジェクト

human_msg = HumanMessage("What is machine learning?")
response = model.invoke([human_msg])

文字列ショートカット

# Using a string is a shortcut for a single HumanMessage
response = model.invoke("What is machine learning?")

 

メッセージ・メタデータ

メタデータの追加

human_msg = HumanMessage(
    content="Hello!",
    name="alice",  # Optional: identify different users
    id="msg_123",  # Optional: unique identifier for tracing
)

 

AI メッセージ

AIMessage はモデル呼び出しの出力を表します。マルチモーダルデータ、ツール呼び出し、そして (後でアクセス可能な) プロバイダー固有のメタデータを含むことができます。

response = model.invoke("Explain AI")
print(type(response))  # <class 'langchain_core.messages.AIMessage'>

AIMessage オブジェクトはモデルを呼び出した際にモデルにより返され、これはレスポンス内の関連メタデータのすべてを含みます。けれども、これが AIMessage オブジェクトの作成/変更が可能な唯一の場所ではありません。

プロバイダーはメッセージのタイプにより異なって重み付けし、コンテキストを理解するので、新しい AIMessage オブジェクトの作成や (モデルから取得したかのように) メッセージ履歴への挿入は役立つ場合があります。

from langchain_core.messages import AIMessage, SystemMessage, HumanMessage

# Create an AI message manually (e.g., for conversation history)
ai_msg = AIMessage("その問題について喜んでお手伝いさせていただきます!")

# Add to conversation history
messages = [
    SystemMessage("あなたは役に立つアシスタントです"),
    HumanMessage("手伝ってもらえますか？"),
    ai_msg,  # Insert as if it came from the model
    HumanMessage("素晴らしい！2+2 はいくつですか？")
]

response = model.invoke(messages)

出力例

4です。ほかにも計算の問題があれば教えてください。

 

ツール呼び出しレスポンス

モデルがツールを呼び出すとき、ツール呼び出しレスポンスは AI メッセージに (tool_calls として) 含まれます :

Tool calling

model_with_tools = model.bind_tools([GetWeather])
response = model_with_tools.invoke("What's the weather in Paris?")

for tool_call in response.tool_calls:
    print(f"Tool: {tool_call['name']}")
    print(f"Args: {tool_call['args']}")
    print(f"ID: {tool_call['id']}")

 

ストリーミングとチャンク

ストリーミング中、完全なメッセージに組み合わせることができる AIMessageChunk オブジェクトを受け取ります :

chunks = []
full_message = None
for chunk in model.stream("Hi"):
    chunks.append(chunk)
    print(chunk.text)
    full_message = chunk if full_message is None else full_message + chunk

 

ツールメッセージ

ツール呼び出しをサポートするモデルについては、AI メッセージはツール呼び出しを含めることができます。ツールメッセージは、単一のツール実行の結果をモデルに返すために使用されます。

from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage

def get_weather(city: str) -> str:
    """Get weather for a given city."""
    return f"{city} では常に晴れています!"

model = init_chat_model("openai:gpt-5-nano")

# モデルがツール呼び出しを行った後
ai_message = AIMessage(
    content=[],
    tool_calls=[{
        "name": "get_weather",
        "args": {"location": "東京"},
        "id": "call_123"
    }]
)

# ツールを実行して結果メッセージを作成
weather_result = "晴れ, 22度"
tool_message = ToolMessage(
    content=weather_result,
    tool_call_id="call_123"  # Must match the call ID
)

# 会話の続行
messages = [
    HumanMessage("東京の天気はどうですか？"),
    ai_message,  # Model's tool call
    tool_message,  # Tool execution result
]
response = model.invoke(messages)  # Model processes the result

出力例

東京は晴れで、気温は22度です。日中は過ごしやすいですが日差しが強いので、外出時は帽子や日焼け止めをお忘れなく。
夜になると涼しくなる可能性があります。何か計画があれば、時間帯に合わせたアドバイスもします。

以上