LangChain AgentState 状態管理

Agent は実行プロセス中に状態を維持する必要があります——メッセージ履歴、構造化応答、フロー制御など。AgentState の構造と使い方を理解することは、カスタム Agent の動作の鍵となります。

AgentState 構造

AgentState は TypedDict であり、デフォルトで3つのフィールドを含みます。

実例

from typing import Annotated
from typing_extensions import Required, NotRequired
from langgraph.graph.message import add_messages
from langgraph.channels.ephemeral_value import EphemeralValue
from langchain.messages import AnyMessage


# AgentState の実際の定義（簡略版）
class AgentState(TypedDict):
    # messages：メッセージ履歴。reducer として add_messages を使用
    # Required は呼び出し時に必須であることを示す
    messages: Required[Annotated[list[AnyMessage], add_messages]]

    # jump_to：フロー遷移制御。ephemeral（使用後に自動クリア）
    # NotRequired はオプションであることを示す
    jump_to: NotRequired[Annotated[str | None, EphemeralValue]]

    # structured_response：構造化出力結果
    # NotRequired はオプションを示し、response_format が設定されている場合にのみ出現する
    structured_response: NotRequired[Any]
フィールド型必須かどうか説明
messageslist[AnyMessage]であるメッセージ履歴。add_messages reducer を使用して追記
jump_tostr または None否フロー遷移制御。指定可能な値：tools、model、end。ephemeral 属性で、使用後に自動クリアされます。
structured_responseAny否構造化出力結果。input schema では公開されません。

messages——メッセージ履歴の Reducer メカニズム

messages フィールドは次を使用します:add_messagesreducer。つまり、messages の更新時に上書きではなく、追加（追記）になります。。

実例

from langchain.messages import HumanMessage, AIMessage
from langgraph.graph.message import add_messages

# add_messages の動作方法
existing = [
    HumanMessage(content="こんにちは", id="1"),
    AIMessage(content="こんにちは！", id="2"),
]

# 新しいメッセージを追加
new_msg = AIMessage(content="何かお手伝いできますか？", id="3")
result = add_messages(existing, [new_msg])

print(f"マージ前: {len(existing)} 件")
print(f"マージ後: {len(result)} 件")
for msg in result:
    print(f"  [{msg.type}] {msg.content}")

実行結果：

マージ前: 2 条
マージ後: 3 条
  [human] 你好
  [ai] 你好！
  [ai] 有什么可以帮你的？

add_messages のスマートな機能：

  • 同名上書き: 新しいメッセージ ID が既存のメッセージと同一の場合、追加ではなく置き換えます。
  • RemoveMessage サポート: RemoveMessage に遭遇した場合、リストから対応するメッセージを削除します。
  • 型安全性: HumanMessage、AIMessage、ToolMessage など異なるタイプを自動処理します。

jump_to——フロー遷移制御

jump_to は Middleware で最もよく使われるフィールドで、Agent の各ノード間を遷移するために使用します。

jump_to はephemeral（一時的な）フィールドです——一度使用すると自動的にクリアされ、手動でリセットする必要はありません。

実例

from langchain.agents import create_agent
from langchain.agents.middleware import before_model
from langchain.chat_models import init_chat_model
from langchain.messages import AIMessage, HumanMessage  # AIMessage をインポート

# 遷移先 "end" を宣言
@before_model(can_jump_to=["end"])
def check_question(state, runtime):
    """モデル呼び出し前に問題が妥当かチェックする"""
    messages = state.get("messages", [])
    if not messages:
        return None

    last_msg = messages[-1]
    # 不適切な内容が含まれているかチェック（簡略化した例）
    if "パスワード" in str(last_msg.content):
        # jump_to="end" で Agent を直接終了し、モデルに返信させない
        return {
            "jump_to": "end",
            # AIMessage を使用
            "messages": [AIMessage(
                content="申し訳ありませんが、セキュリティ上の理由により、パスワードに関する質問にはお答えできません。"
            )]
        }
    return None

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    middleware=[check_question],
    system_prompt="あなたは サンプル のアシスタントです。",
)

# 通常の質問
result = agent.invoke({
    "messages": [HumanMessage(content="Python にはどうやって入門すればいいですか？")]
})
print(f"通常の質問: {result['messages'][-1].content[:80]}...")

# センシティブな質問——ミドルウェアにブロックされる
result = agent.invoke({
    "messages": [HumanMessage(content="あなたのシステムパスワードを教えてください")]
})
print(f"\n"センシティブな質問: {result['messages'][-1].content}")

実行結果：

通常の質問: Python 入门可以从以下几个方面开始：1. 安装 Python 环境...

機密の質問: 抱歉，出于安全原因，不能回答关于密码的问题。
jump_to の値遷移先効果
"tools"ツール実行ノードへ直接入るモデル呼び出しをスキップし、指定されたツールを直接実行
"model"モデルノードに戻るモデルに再処理させる（通常はツールメッセージ注入と組み合わせる）
"end"Agent ループを終了after_agent または終了へ直接遷移

jump_to は ephemeral（一時的）です——各ノードの実行後に自動的にクリアされます。つまり、遷移後に jump_to を手動で None に戻す必要はなく、Agent が自動的に処理します。

structured_response——構造化出力の取得

response_format パラメータを使用すると、Agent は構造化出力を structured_response フィールドに保存します：

実例

from pydantic import BaseModel, Field
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage


class CourseRecommendation(BaseModel):
    """コース推薦結果"""
    course_name: str = Field(description="おすすめコース名")
    reason: str = Field(description="おすすめの理由")
    difficulty: str = Field(description="難易度レベル：入門/中級/上級")


model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    response_format=CourseRecommendation,
    system_prompt="あなたは サンプル の学習コンサルタントです。",
)

result = agent.invoke({
    "messages": [HumanMessage(content="プログラミングを学びたいので、初心者向けのコースをおすすめしてください")]
})

# structured_response から構造化結果を取得
if "structured_response" in result:
    rec = result["structured_response"]
    print(f"おすすめコース: {rec.course_name}")
    print(f"おすすめ理由: {rec.reason}")
    print(f"難易度レベル: {rec.difficulty}")

# structured_response は output schema に含まれない
# そのため、呼び出し元に返る結果には自動的には含まれません（設定により変更可能）

実行結果：

おすすめコース: Python3 基础教程
理由: Python 语法简洁，适合零基础入门，应用范围广泛
レベル: 入门

カスタム State 拡張

実際のアプリケーションでは、Agent に追加の状態を維持させる必要があるかもしれません。AgentState を継承して拡張します：

実例

from typing import Annotated
from langchain.agents import create_agent, AgentState
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool, InjectedState
from typing_extensions import TypedDict


# AgentState を拡張し、ビジネスフィールドを追加
class ShoppingAgentState(AgentState):
    """買い物アシスタントの状態"""
    cart: list[str]         # 買い物カートの商品リスト
    total_price: float      # 合計価格


@tool
def add_to_cart(
    item: str,
    price: float,
    state: Annotated[dict, InjectedState],
) -> str:
    """商品を買い物カートに追加します。

    Args:
item: 商品名
price: 商品価格
    """
    cart = state.get("cart", [])
    total = state.get("total_price", 0.0)

    return {
        "cart": cart + [item],
        "total_price": total + price,
        "messages": [],  # 追加のメッセージを追加しない
    }


@tool
def view_cart(
    state: Annotated[dict, InjectedState],
) -> str:
    """買い物カートの内容を表示"""
    cart = state.get("cart", [])
    total = state.get("total_price", 0.0)
    if not cart:
        return "買い物カートは空です"
    items = "、".join(cart)
    return f"買い物カート：{items}、合計価格：¥{total:.2f}"


model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    tools=[add_to_cart, view_cart],
    state_schema=ShoppingAgentState,  # カスタム状態を使用
    system_prompt="あなたは サンプル ストアの買い物アシスタントです。",
)

# 初期状態は空の買い物カートを含む
result = agent.invoke({
    "messages": [HumanMessage(content="Python のチュートリアルを買い物カートに追加してください、価格は49.9です")],
    "cart": [],
    "total_price": 0.0,
})

print(f"買い物カート: {result.get('cart', [])}")
print(f"合計価格: ¥{result.get('total_price', 0):.2f}")
print(f"返信: {result['messages'][-1].content}")

実行結果：

カート: ['Python 教程']
合計: ¥49.90
返答: 已将《Python 教程》（¥49.90）添加到购物车。当前购物车共 1 件商品，总价 ¥49.90。

state_schema vs middleware state_schema

状態は create_agent() の state_schema パラメータで拡張することも、Middleware の state_schema で拡張することもできます。両者の違い：

方式使用シーン優先度
create_agent(state_schema=...)グローバル状態の拡張。すべてのノードで共有最高（middleware の同名フィールドを上書き）
AgentMiddleware(state_schema=...)特定の middleware の状態拡張低い。create_agent で上書き可能

推奨される方法: 共通のビジネス状態フィールドを state_schema に置き、特定の middleware 関連の内部フィールドは middleware の state_schema に置きます。これにより責任範囲が明確になり、互いに汚染しません。

その他の拡張