integrate-arcjet-guard-langchain-py

在 Python LangChain 中集成 Arcjet Guard

arcjet.guard.langchain 会包装 Agent 现有的 Arcjet 客户端，不会自行与 Arcjet API 通信。共享的 Guard 基础内容（客户端、规则、标签、决策、捕获、注册）位于 ../arcjet/references/guards_python.md。对于非 LangChain 特有的内容，请加载该参考文档。

四种用法，同一条决策规则：

• 任何 Python 可调用对象 → 使用核心 arcjet.guard 中的 guard_action / guard_action_sync。无需安装 LangChain extra。详情请参阅共享的 Python Guard 参考文档。
• 由你自行调用的 LangChain BaseTool → 使用 guard_tool。DENY 会引发 ArcjetToolDeniedError（工具的 handle_tool_error 可能会转换该错误）。
• create_agent（由模型选择工具） → 使用 ArcjetMiddleware + ToolPolicy。没有策略的工具会直接放行。
• 观察链或 Agent → 使用 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 时会引发异常，并指出需要安装该 extra。

在 agent.ainvoke 前筛查入站内容 — 没有入站辅助工具

没有入站辅助工具。请在应用中调用核心 aj.guard(...)，并在 ainvoke / invoke 前执行，以应用提示词注入等入站规则。 核心 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，然后再包装。对受保护句柄所做的更改不会传递给调用。

首先向用户询问的问题

只询问无法从代码中推断的信息；同时建议默认值。

1. 哪些工具具有风险（会产生外部副作用、不可逆操作、花费金钱或发送消息）？你自行调用的工具应使用 guard_tool。由模型选择的工具应在 ArcjetMiddleware 上使用 ToolPolicy。
2. 有哪些限制？（例如，“每位用户每分钟查找 10 次” → TokenBucket。）
3. 元数据中的用户是谁 — 不透明的用户 / 租户 ID（绝不能是 PII）？默认值：无。已有的对话 / 会话 ID 是关联 ID，而不是用户 ID。请将其设为 config["configurable"]["arcjet_correlation_id"]，或使用 with arcjet_sequence(correlation_id=...)。
4. Arcjet 服务中断是否不可接受？所有辅助工具的默认值都是 on_guard_error="deny"。请明确询问是否要在 ainvoke 前进行入站筛查：在此处采用失败时拒绝策略，意味着服务中断期间 Agent 不会运行，因此在这个调用点选择 "allow" 是常见且合理的做法。

读者容易误解的事项

1. 这不是 LangChain JS，也不是 LangGraph JS。 不要导入 @arcjet/guard/langchain/v1 或 @arcjet/guard/langgraph/v1。
2. 没有入站辅助函数。 在调用 ainvoke 前，先使用核心 guard() 进行检查。
3. ArcjetCaptureHandler 无法拒绝请求。 策略由 guard_tool 或 ArcjetMiddleware 执行。
4. 关联标识由调用方负责，不会自动生成。 不要为每轮生成新 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= 是静态序列，不是 lambda——传入可调用对象会引发 TypeError。请在工具主体中执行逐参数检查，或使用 inputs=（此适配器接受带类型的输入；解析器为 (arguments, config)）。
9. 缺少决策不等于拒绝。 请在 Console/CLI 中验证。

步骤 1：安装并找到 Guard 客户端

这些辅助函数包含在 PyPI arcjet 1.0.0+ 中。请使用项目的包管理器安装。不要凭猜测的版本手动编辑 requirements.txt。

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

如果 Agent 尚无 Guard 客户端，请在模块作用域启动一次。异步代码使用 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= 为可选项。它们可以组合使用：受保护的 Agent 中包含受保护的工具时，每项策略只评估一次，并且两者都会落在同一个 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：检查入站请求并建立关联

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")

请让整个运行过程使用同一个 Sequence，可通过 with arcjet_sequence(correlation_id=session.id): 或 config={"configurable": {"arcjet_correlation_id": session.id}} 实现。不要为每轮生成新 ID。

验证集成

1. python -m py_compile（或项目的类型检查）通过。
2. 测试入站 PI（在 ainvoke 前）、guard_tool 拒绝、针对命名策略的中间件拒绝、不进行拦截的仅捕获处理器、速率限制，以及 fail-closed（Guard 无法访问）的情况。
3. 在 Arcjet Console / CLI（guards list）中确认各项决策共享调用方提供的关联 ID。
4. 使用真实 ARCJET_KEY 进行的手动端到端测试，需等你实际运行后才能验证。

完整示例： examples/fastapi-langchain-guard。 不要另造示例名称，也不要在此 skills 仓库中添加示例。