integrate-arcjet-guard-langchain-py

Python LangChain に Arcjet Guard を統合する

arcjet.guard.langchain は、エージェントが既に使用している Arcjet クライアントをラップします。Arcjet API と直接通信することはありません。Guard の共通の基本機能（クライアント、ルール、ラベル、判定、キャプチャ、登録）は ../arcjet/references/guards_python.md にあります。LangChain 固有ではない内容については、このリファレンスを参照してください。

4 つの使い方に、1 つの判定ルールがあります。

• 任意の Python 呼び出し可能オブジェクト → コアの arcjet.guard にある guard_action / guard_action_sync。LangChain の追加パッケージは不要です。詳細は Python Guard 共通リファレンスを参照してください。
• 自分で呼び出す LangChain BaseTool → guard_tool。DENY の場合は ArcjetToolDeniedError が発生します（ツールの handle_tool_error によって変換される場合があります）。
• create_agent（モデルがツールを選択する場合） → ArcjetMiddleware + ToolPolicy。ポリシーのないツールはそのまま実行されます。
• チェーンまたはエージェントを監視する場合 → ArcjetCaptureHandler / ArcjetAsyncCaptureHandler。これらはリクエストを拒否できません。

これは Python の create_agent です。統合ページ https://docs.arcjet.com/guards/langchain/ では両方の言語を扱っています。JS ではこのモジュールではなく、 @arcjet/guard/langchain/v1（createAgent / wrapToolCall）を使用します。LangGraph JS の StateGraph / ToolNode ではありません（ドキュメント： https://docs.arcjet.com/guards/langgraph/）。 arcjet.guard.langchain をインポートしても、LangGraph は読み込まれません。 arcjet[langchain-agents] がインストールされていない状態で ArcjetMiddleware または ToolPolicy を参照すると、その追加パッケージ名を示すエラーが発生します。

agent.ainvoke の前に受信入力をスクリーニングする — 受信入力用のヘルパーはありません

受信入力用のヘルパーはありません。ainvoke / invoke の前にコアの aj.guard(...) を使用し、アプリケーション内でプロンプトインジェクション（およびその他の受信入力ルール）を適用してください。 コアの guard() はフェイルオープンです。ALLOW はルールが実行された証拠にはなりません。この呼び出し箇所でフェイルクローズにする必要がある場合は、 decision.has_failed_open() に基づいて判断してください。 guard_tool / ArcjetMiddleware は、すでにデフォルトでこの動作になっています。

キャプチャハンドラーは処理をブロックしません

LangChain はコールバックの戻り値を無視します。ArcjetCaptureHandler は記録のみを行います。ポリシーは guard_action / guard_tool / ArcjetMiddleware に設定します。

guard_tool() の前にツールを設定する

args_schema を絞り込み、手元にあるツールに handle_tool_error / callbacks / response_format を設定してからラップしてください。Guard を適用したハンドルへの変更は、呼び出しに反映されません。

まず人に確認する質問

コードから推測できないことだけを質問し、デフォルトを提案してください。

1. どのツールがリスクを伴いますか（外部への副作用、取り消し不能、金銭の消費、メッセージの送信など）？自分で呼び出すツールには guard_tool を使用します。モデルが選択するツールには、ArcjetMiddleware に ToolPolicy を設定します。
2. どのような制限が必要ですか？（例：「ユーザーごとに 1 分あたり 10 回の検索」→ TokenBucket。）
3. メタデータ上のユーザーは誰ですか — 個人を特定できないユーザー／テナント ID（PII は使用しない）ですか？ デフォルトは指定なしです。既存の会話 ID／セッション ID は相関 ID であり、ユーザー ID ではありません。 config["configurable"]["arcjet_correlation_id"] または with arcjet_sequence(correlation_id=...) に設定してください。
4. Arcjet の障害は許容できませんか？すべてのヘルパーで on_guard_error="deny" がデフォルトです。ainvoke の前に行う受信入力のスクリーニングについては、明示的に確認してください。ここでフェイルクローズにすると、障害が続く間はエージェントが実行されません。そのため、この呼び出し箇所に限って "allow" を選ぶのは一般的で妥当な選択です。

読者が誤解しやすい点

1. これは LangChain JS でも LangGraph JS でもありません。 @arcjet/guard/langchain/v1 や @arcjet/guard/langgraph/v1 をインポートしないでください。
2. インバウンド用ヘルパーはありません。 ainvoke の前に、コアの guard() でスクリーニングしてください。
3. ArcjetCaptureHandler では拒否できません。 ポリシーには guard_tool または ArcjetMiddleware を使用してください。
4. 相関 ID は呼び出し側が管理し、新たに生成されることはありません。 ターンごとに新しい ID を生成しないでください。LangChain の run_id は使用されません。設定キーは外側の arcjet_sequence より優先されます。configurable は metadata より先に確認されます。
5. tools= には、create_agent に渡したものと同じシーケンスを渡してください。 ポリシーキーに誤字がある場合、ツールが保護されないままになるのではなく、構築時に拒否されます。
6. レート制限のキーには、モデルが指定した注文 ID ではなく、認証済みの呼び出し元を使ってください。
7. ヘルパーによるキャプチャの success は、「アクションが実行された」という意味ではありません。 guard_action／guard_tool／ArcjetMiddleware は metadata.outcome に書き込みます。共通の Python Guard リファレンスを参照してください。
8. rules= は静的なシーケンスであり、ラムダではありません。呼び出し可能オブジェクトを渡すと TypeError が発生します。引数ごとのチェックはツール本体に記述するか、inputs= を使用してください（このアダプターは型付き入力を受け付けます。リゾルバーは (arguments, config) です）。
9. 判定がないことは、拒否を意味しません。 Console／CLI で確認してください。

ステップ 1：インストールしてガードクライアントを確認する

これらのヘルパーは、PyPI の arcjet 1.0.0+ に含まれています。プロジェクトのパッケージマネージャーを使ってインストールしてください。推測したバージョンを requirements.txt に手動で書き込まないでください。

# guard_tool + capture handlers
pip install "arcjet[langchain]"
# ArcjetMiddleware + ToolPolicy
pip install "arcjet[langchain-agents]"

エージェントにガードクライアントがまだない場合は、モジュールスコープで一度だけ起動してください。非同期コードでは launch_arcjet を、同期コードでは launch_arcjet_sync を使用します。

import os
from arcjet.guard import launch_arcjet

aj = launch_arcjet(key=os.environ["ARCJET_KEY"])

ステップ 2：呼び出す BaseTool を保護する — guard_tool

from arcjet.guard.langchain import guard_tool

send_email.args_schema = PublicEmailArgs  # narrow first, then wrap
guarded = guard_tool(
    guard=aj,
    tool=send_email,
    action="email.sent",
    rules=[email_limit(key=user_id, requested=1)],
    on_guard_error="deny",
)

arcjet[langchain] が必要です。結果も引き続き BaseTool です。DENY では ArcjetToolDeniedError が発生し、利用不可の場合は ArcjetToolUnavailableError が発生します。配線時にツール名を指定できる場合は、変更がより小さい guard_tool を使用してください。

ステップ 3：create_agent を保護する — ArcjetMiddleware

from langchain.agents import create_agent
from arcjet.guard.langchain import ArcjetMiddleware, ToolPolicy

tools = [send_email, search_orders]
agent = create_agent(
    model="openai:gpt-4o",
    tools=tools,
    middleware=[
        ArcjetMiddleware(
            guard=aj,
            policies={
                "send_email": ToolPolicy(
                    action="email.sent",
                    rules=[email_limit(key=user_id, requested=1)],
                ),
            },
            tools=tools,
            on_guard_error="deny",
        )
    ],
)

await agent.ainvoke(
    {"messages": [...]},
    config={"configurable": {"arcjet_correlation_id": session.id}},
)

arcjet[langchain-agents] が必要です。すでに register_arcjet() を呼び出している場合、guard= は省略可能です。両者は組み合わせて使用できます。保護されたエージェント内の保護されたツールでは、各ポリシーが一度ずつ評価され、どちらも同じ Sequence に記録されます。

ステップ 4：チェーンを監視する — ArcjetCaptureHandler

from arcjet.guard.langchain import ArcjetAsyncCaptureHandler, ArcjetCaptureHandler

# invoke → ArcjetCaptureHandler; ainvoke → ArcjetAsyncCaptureHandler
chain.invoke(inputs, config={"callbacks": [ArcjetCaptureHandler(guard=aj)]})
await chain.ainvoke(
    inputs, config={"callbacks": [ArcjetAsyncCaptureHandler(guard=aj)]}
)

guard_tool と同じ追加パッケージが必要です。ハンドラーは呼び出しと組み合わせてください。どちらも呼び出しを拒否することはできません。

ステップ 5：インバウンドをスクリーニングして相関 ID を設定する

from arcjet.guard import DetectPromptInjection

inbound = DetectPromptInjection()
decision = await aj.guard(
    label="message.received",
    rules=[inbound(user_text)],
)
if decision.conclusion == "DENY":
    raise RuntimeError("message blocked")
if decision.has_failed_open():
    raise RuntimeError("inbound guard unavailable")

with arcjet_sequence(correlation_id=session.id): または config={"configurable": {"arcjet_correlation_id": session.id}} を使い、実行全体で同じ Sequence を維持してください。ターンごとに新しい ID を生成しないでください。

インテグレーションを検証する

1. python -m py_compile（またはプロジェクトの型チェック）が成功すること。
2. インバウンド PI（ainvoke の前）、guard_tool による拒否、名前付きポリシーに対するミドルウェアの拒否、ブロックしないキャプチャ専用ハンドラー、レート制限、およびフェイルクローズ（ガードに到達できない場合）をテストすること。
3. Arcjet Console／CLI（guards list）で、判定に呼び出し側が管理する同じ相関 ID が使われていることを確認すること。
4. 実際の ARCJET_KEY を使った手動 E2E は、実行するまでは未検証です。

実装例： examples/fastapi-langchain-guard。 別の実装例名を作らないでください。このスキルリポジトリに実装例を追加しないでください。