如何在 LangGraph 代理中解决 Cloudflare Turnstile（验证码）

如何在LangGraph代理中解决Cloudflare Turnstile问题

Ethan Collins

Pattern Recognition Specialist

23-Jul-2026

TL;DR

一个LangGraph Cloudflare Turnstile求解器集成应该将CAPTCHA处理建模为受控恢复节点，而不是每个路由都可用的无限制工具。
CapSolver将capsolver-agent文档化为capsolver-core的工具适配层，具有LangChain工具用于代理框架和Playwright会话的浏览器方法。
保持导航决策在图中，确定性授权在应用代码中，识别在CapSolver服务中。
通过检测、求解、填充和最终页面断言保留一个浏览器会话；永远不要将原始令牌返回给模型。
使用有限重试、明确的终止状态、脱敏追踪和人工审核进行状态更改或不明确的操作。
下面的示例图包括状态定义、策略路由、CapSolver恢复节点、故障处理和验证。

一个LangGraph Cloudflare Turnstile集成应该做什么

一个LangGraph Cloudflare Turnstile集成允许代理在授权的浏览器工作流中从验证步骤中恢复，然后继续原始任务。图不应要求语言模型点击或通过小部件推理。相反，模型或浏览器控制器检测到工作流被阻塞，图评估策略，并由确定性适配器调用记录的CapSolver功能。

CapSolver在其代理工具指南中记录了这种分工：模型处理导航和决策，capsolver-agent暴露工具模式和执行器，capsolver-core执行检测、求解和浏览器填充。

这种架构使LangGraph具有有用的作用。它可以使得恢复可见，强制重试预算，将敏感操作路由到人工，确保浏览器在图继续之前验证成功。

先决条件和受支持的集成路径

使用隔离的Python环境。CapSolver当前的官方指南从GitHub安装核心和代理包：

bash 复制代码

python -m venv .venv
source .venv/bin/activate
pip install git+https://github.com/capsolver-ai/capsolver-core.git
pip install "capsolver-agent[langchain] @ git+https://github.com/capsolver-ai/capsolver-agent.git"
pip install langchain-openai langgraph playwright
playwright install chromium

将CAPSOLVER_API_KEY和任何模型凭证放在批准的密钥存储中。不要将真实值写入图状态、检查点、提示、追踪事件或源文件中。

您还需要：

一个拥有或明确授权的目标；
域名白名单；
明确的业务目的；
浏览器会话注册表；
最大重试次数；
超时预算；
最终页面断言；
对关键操作的人工审核规则。

设计LangGraph状态

在图状态中仅保留非秘密的操作数据：

python 复制代码

from typing import Literal, TypedDict

class AgentState(TypedDict, total=False):
    request_id: str
    purpose: str
    page_id: str
    current_url: str
    step: str
    challenge_detected: bool
    challenge_attempts: int
    challenge_status: Literal[
        "not-needed", "pending", "resolved", "review", "denied"
    ]
    error_code: str | None
    final_assertion_passed: bool

不要添加CapSolver凭证、解决方案令牌、cookies或原始页面内容。将浏览器对象存储在由page_id键控的应用程序拥有的注册表中；图检查点应仅包含不透明的标识符。

构建确定性授权节点

授权应在任何挑战工具之前运行：

python 复制代码

from urllib.parse import urlparse

ALLOWED_HOSTS = {"staging.example.com", "research.example.com"}
ALLOWED_PURPOSES = {"qa-validation", "public-data-research"}

def authorize_challenge(state: AgentState) -> AgentState:
    host = urlparse(state["current_url"]).hostname
    attempts = state.get("challenge_attempts", 0)

    if host not in ALLOWED_HOSTS:
        return {**state, "challenge_status": "denied", "error_code": "domain"}
    if state.get("purpose") not in ALLOWED_PURPOSES:
        return {**state, "challenge_status": "denied", "error_code": "purpose"}
    if attempts >= 2:
        return {**state, "challenge_status": "review", "error_code": "retry-limit"}
    return {**state, "challenge_status": "pending"}

此节点是语法验证的，并且独立于模型。在生产中，从版本化配置中加载策略并拒绝未知字段。

在图状态外注册浏览器会话

python 复制代码

class BrowserRegistry:
    def __init__(self):
        self._pages = {}

    def register(self, page_id: str, page) -> None:
        self._pages[page_id] = page

    def get(self, page_id: str):
        if page_id not in self._pages:
            raise KeyError("browser page is not registered")
        return self._pages[page_id]

    async def remove(self, page_id: str) -> None:
        page = self._pages.pop(page_id, None)
        if page is not None:
            await page.close()

注册表防止序列化Playwright Page，并为宿主提供一个地方来强制清理。

领取您的CapSolver优惠码

立即提升您的自动化预算！
在充值CapSolver账户时使用优惠码 CAP26，每次充值可获得额外 5% 的奖励 —— 没有限制。
现在在您的 CapSolver仪表板中领取

实现CapSolver恢复节点

CapSolver的Core SDK文档记录了detect(page)和solve_on_page(page)用于浏览器模式。下面的适配器使用这些方法并仅返回图决策：

python 复制代码

import os
from capsolver_core import create_capsolver

async def solve_turnstile_node(
    state: AgentState,
    registry: BrowserRegistry,
) -> AgentState:
    page = registry.get(state["page_id"])
    attempts = state.get("challenge_attempts", 0) + 1

    async with create_capsolver(
        api_key=os.environ["CAPSOLVER_API_KEY"],
        default_timeout=120,
        polling_interval=5,
    ) as cap:
        detected = await cap.detect(page)
        if not detected:
            return {
                **state,
                "challenge_attempts": attempts,
                "challenge_status": "not-needed",
                "error_code": None,
            }

        results = await cap.solve_on_page(page)

    failures = [item for item in results if item.error or not item.filled]
    if failures:
        return {
            **state,
            "challenge_attempts": attempts,
            "challenge_status": "review" if attempts >= 2 else "pending",
            "error_code": "fill-back-failed",
        }

    return {
        **state,
        "challenge_attempts": attempts,
        "challenge_status": "resolved",
        "error_code": None,
    }

代码已通过语法检查，但未运行凭证。实时测试需要经过批准的页面和秘密。图永远不会收到solution.token。

验证页面结果

填充的令牌是中间结果。验证应用程序的预期状态：

python 复制代码

async def verify_page_node(
    state: AgentState,
    registry: BrowserRegistry,
) -> AgentState:
    page = registry.get(state["page_id"])
    try:
        await page.get_by_test_id("authorized-content").wait_for(timeout=15_000)
        return {
            **state,
            "final_assertion_passed": True,
            "step": "continue",
            "error_code": None,
        }
    except Exception:
        return {
            **state,
            "final_assertion_passed": False,
            "challenge_status": "review",
            "error_code": "page-assertion-failed",
        }

使用由您的应用程序拥有的断言。避免在日志中暴露个人或敏感页面内容的选择器。

组装LangGraph工作流

python 复制代码

from langgraph.graph import END, StateGraph

def route_after_authorization(state: AgentState) -> str:
    if state["challenge_status"] == "pending":
        return "solve"
    if state["challenge_status"] in {"denied", "review"}:
        return "human_review"
    return "verify"

def route_after_solve(state: AgentState) -> str:
    if state["challenge_status"] == "resolved":
        return "verify"
    if state["challenge_status"] == "pending":
        return "authorize"
    return "human_review"

def build_graph(authorize, solve, verify, human_review):
    graph = StateGraph(AgentState)
    graph.add_node("authorize", authorize)
    graph.add_node("solve", solve)
    graph.add_node("verify", verify)
    graph.add_node("human_review", human_review)
    graph.set_entry_point("authorize")
    graph.add_conditional_edges(
        "authorize",
        route_after_authorization,
        {"solve": "solve", "verify": "verify", "human_review": "human_review"},
    )
    graph.add_conditional_edges(
        "solve",
        route_after_solve,
        {"authorize": "authorize", "verify": "verify", "human_review": "human_review"},
    )
    graph.add_edge("verify", END)
    graph.add_edge("human_review", END)
    return graph.compile()

注入的函数可以封闭浏览器注册表。依赖注入使策略和故障路由可测试，而无需实时服务。

添加人工审核节点

审核员应收到：

请求ID；
批准的用途；
主机名；
尝试的操作；
重试次数；
脱敏错误类别；
如果允许，安全的截图引用；
建议的下一步。

审核员不应收到CapSolver凭证或解决方案令牌。提交、购买、账户更改或消息发送等状态更改操作即使在验证成功后也需要自己的授权。

在不调用外部服务的情况下测试图

单元测试可以将解决节点替换为确定性存根：

python 复制代码

async def solved_stub(state: AgentState) -> AgentState:
    return {
        **state,
        "challenge_attempts": state.get("challenge_attempts", 0) + 1,
        "challenge_status": "resolved",
        "error_code": None,
    }

async def failed_stub(state: AgentState) -> AgentState:
    return {
        **state,
        "challenge_attempts": state.get("challenge_attempts", 0) + 1,
        "challenge_status": "review",
        "error_code": "fixture-failure",
    }

测试批准和拒绝的域名、不支持的用途、重试耗尽、缺少浏览器页面、解决的挑战但页面断言失败，以及终端状态后的清理。

在页面参数已知时使用令牌模式

当应用程序知道Turnstile页面URL和公开站点密钥时，令牌模式可能更简单。CapSolver文档记录了AntiTurnstileTaskProxyLess任务，需要websiteURL和websiteKey，以及可选的metadata.action和metadata.cdata。

不要让模型发明这些字段。从批准的页面或应用程序配置中确定性地提取它们。

无秘密泄露的可观测性

追踪：

图节点；
请求ID；
策略决策；
主机名；
挑战类型；
尝试次数；
经过时间；
错误类别；
填充布尔值；
最终断言布尔值。

不要追踪包含凭证、浏览器cookies、原始令牌或未脱敏表单数据的提示。为截图和DOM证据定义保留和访问规则。

故障模式和修复

未检测到挑战

确认页面已完成加载，浏览器使用了预期的会话，并且SDK版本支持挑战类型。仅当页面断言仍能通过时，才将空检测结果视为not-needed。

任务在就绪前失败

记录错误类别，将参数与当前CapSolver文档进行比较，并在重试预算后停止。不要自动增加重试次数。

令牌填充失败

保持相同的浏览器页面，审查回调或小部件行为，并验证页面导航未替换上下文。

图无限循环

存储并强制challenge_attempts。在配置的限制后路由到人工审核。

验证成功但业务操作失败

保持CAPTCHA恢复与下游操作分离。图应显示操作本身的错误，而不是重新解决挑战。

生产检查清单

固定依赖版本或提交。
使用单独的预发布和生产策略。
将秘密存储在图状态之外。
限制域名和用途。
设置每调用和总截止时间。
保留一个浏览器会话。
脱敏令牌和cookies。
在填充后验证页面。
对关键操作添加人工审核。
测试清理和取消。
监控拒绝、错误、恢复和断言率。
审查政策变更如代码。

结论：将挑战处理作为图恢复状态

当LangGraph Cloudflare Turnstile集成像有限恢复工作流一样行为时，最可靠：检测、授权、解决、验证、继续或停止。图提供路由和可观测性；确定性代码提供策略；CapSolver提供记录的识别层。

仅在合法、授权的自动化中使用CapSolver。在固定实现前，查阅当前的代理工具文档、Core SDK指南和相关的CapSolver博客教程。

FAQ

Q: LangGraph是否自行解决Cloudflare Turnstile？

No. LangGraph控制工作流状态和路由；CapSolver适配器调用识别服务和浏览器方法。

Q: 是否应将解决方案令牌返回给模型？

No. 在受控浏览器适配器中应用它，并仅返回状态、错误类别和验证结果。

Q: 与Playwright一起使用哪个CapSolver方法？

当前Core SDK文档记录了detect(page)、get_captcha_info(page)和solve_on_page(page)用于浏览器模式。

Q: 图应允许多少次重试？

使用基于工作流的小的显式预算，并在路由到审核而不是允许无限制循环时进行路由。

Q: 代理可以对任何URL调用恢复节点吗？

No. 在节点调用CapSolver之前，强制执行确定性的域名和用途白名单。

Q: 什么证明挑战已成功处理？
页面级应用断言证明工作流恢复；仅提供者状态或填充的令牌不足以。

合规声明：本博客提供的信息仅供参考。CapSolver 致力于遵守所有适用的法律和法规。严禁以非法、欺诈或滥用活动使用 CapSolver 网络，任何此类行为将受到调查。我们的验证码解决方案在确保 100% 合规的同时，帮助解决公共数据爬取过程中的验证码难题。我们鼓励负责任地使用我们的服务。如需更多信息，请访问我们的服务条款和隐私政策。