如何在 LlamaIndex 代理中解决 reCAPTCHA v3

LlamaIndex 代理
    ↓ 选择一个类型化的工具
FunctionTool 包装器
    ↓ 验证可信引用
CapSolver 执行器
    ↓ 返回一个短时解决方案
浏览器服务
    ↓ 提交并验证
LlamaIndex 工作流继续

from dataclasses import dataclass
from urllib.parse import urlparse

@dataclass(frozen=True)
class CaptchaContext:
    context_id: str
    website_url: str
    website_key: str
    page_action: str
    enterprise: bool = False
    proxy_profile: str | None = None
    session_mode: bool = False

TRUSTED_CONTEXTS: dict[str, CaptchaContext] = {}
ALLOWED_HOSTS = {"staging.example.com", "portal.example.org"}


def get_trusted_context(context_id: str) -> CaptchaContext:
    context = TRUSTED_CONTEXTS.get(context_id)
    if context is None:
        raise ValueError("未知的 CAPTCHA 上下文")

    host = urlparse(context.website_url).hostname
    if host not in ALLOWED_HOSTS:
        raise PermissionError("目标超出批准的主机策略")

    if not context.website_key or not context.page_action:
        raise ValueError("受信任的上下文缺少必需的 v3 参数")

    return context

pip install git+https://github.com/capsolver-ai/capsolver-core.git
pip install git+https://github.com/capsolver-ai/capsolver-agent.git
pip install llama-index-core

export CAPSOLVER_API_KEY="your-capsolver-api-key"

import os
from capsolver_agent.schema import create_executor

executor = create_executor(
    api_key=os.environ["CAPSOLVER_API_KEY"],
    default_timeout=120,
)

from typing import Annotated

async def solve_recaptcha_v3(
    context_id: Annotated[
        str,
        "用于受信任的当前浏览器 CAPTCHA 上下文的不透明 ID"
    ],
) -> dict:
    """为批准的浏览器上下文解决 reCAPTCHA v3。

    仅在当前工作流报告支持的 reCAPTCHA v3 检查点时使用。永远不要猜测或修改目标 URL、站点密钥或操作。
    """
    context = get_trusted_context(context_id)

    captcha_type = (
        "reCaptchaV3Enterprise"
        if context.enterprise
        else "reCaptchaV3"
    )

    args = {
        "captcha_type": captcha_type,
        "website_url": context.website_url,
        "website_key": context.website_key,
        "page_action": context.page_action,
    }

    if context.proxy_profile:
        args["proxy"] = resolve_proxy(context.proxy_profile)

    result = await executor.execute("solve_captcha", args)
    if not result.get("success"):
        return {
            "success": False,
            "context_id": context_id,
            "error": normalize_error(result.get("error")),
        }

    solution = result.get("solution") or {}
    token = solution.get("token")
    if not token:
        return {
            "success": False,
            "context_id": context_id,
            "error": "solution did not contain a token",
        }

    receipt = await submit_solution_and_verify(
        context_id=context_id,
        token=token,
        session_cookie=extract_session_cookie(solution),
    )

    return {
        "success": receipt["verified"],
        "context_id": context_id,
        "verified": receipt["verified"],
        "next_state": receipt["next_state"],
    }

from llama_index.core.tools import FunctionTool

tool = FunctionTool.from_defaults(
    async_fn=solve_recaptcha_v3,
    name="solve_recaptcha_v3",
    description=(
        "为批准的当前浏览器上下文解决 reCAPTCHA v3。"
        "输入必须是浏览器服务提供的不透明 context_id。"
        "不要为不支持的页面或未批准的主机调用。"
    ),
)

schema = tool.metadata.get_parameters_dict()
print(schema)

from llama_index.core.agent.workflow import FunctionAgent

agent = FunctionAgent(
    llm=llm,
    tools=[tool],
    system_prompt=(
        "仅操作批准的浏览器工作流。当浏览器服务报告支持的 reCAPTCHA v3 检查点时，"
        "使用提供的 context_id 调用 solve_recaptcha_v3。仅调用一次。"
        "仅在 verified=true 时继续；否则请求审查。"
    ),
)

response = await agent.run(
    "批准的暂存工作流在 reCAPTCHA v3 检查点等待。使用 context_id ctx_7f19 并在 verified 时继续。"
)

async def collect_v3_context(page, context_id: str) -> CaptchaContext:
    website_url = page.url
    host = urlparse(website_url).hostname
    if host not in ALLOWED_HOSTS:
        raise PermissionError("未批准的目标")

    values = await page.evaluate("""
    () => {
      const scripts = Array.from(document.scripts)
        .map(s => s.textContent || '')
        .join('\n');

      const siteKey =
        document.querySelector('[data-sitekey]')?.getAttribute('data-sitekey')
        || null;

      const actionMatch = scripts.match(
        /grecaptcha(?:\.enterprise)?\.execute\([^,]+,\s*\{\s*action:\s*['\"]([^'\"]+)/
      );

      return {
        siteKey,
        pageAction: actionMatch ? actionMatch[1] : null,
        enterprise: scripts.includes('grecaptcha.enterprise')
      };
    }
    """)

    if not values["siteKey"] or not values["pageAction"]:
        raise RuntimeError("无法读取必需的 v3 参数")

    return CaptchaContext(
        context_id=context_id,
        website_url=website_url,
        website_key=values["siteKey"],
        page_action=values["pageAction"],
        enterprise=values["enterprise"],
    )

SESSION_KEYS = {
    "recaptcha-ca-t",
    "recaptcha_ca_t",
}


def extract_session_cookie(solution: dict) -> str | None:
    raw = solution.get("raw") or {}
    for key in SESSION_KEYS:
        value = solution.get(key) or raw.get(key)
        if value:
            return value
    return None

async def submit_solution_and_verify(
    context_id: str,
    token: str,
    session_cookie: str | None,
) -> dict:
    browser_state = BROWSER_CONTEXTS[context_id]
    page = browser_state.page

    if session_cookie:
        await browser_state.context.add_cookies([{
            "name": "recaptcha-ca-t",
            "value": session_cookie,
            "domain": urlparse(page.url).hostname,
            "path": "/",
            "secure": True,
        }])

    await page.evaluate(
        """({ token }) => {
          let input = document.querySelector(
            'textarea[name="g-recaptcha-response"]'
          );
          if (!input) {
            input = document.createElement('textarea');
            input.name = 'g-recaptcha-response';
            input.style.display = 'none';
            document.body.appendChild(input);
          }
          input.value = token;
          input.dispatchEvent(new Event('change', { bubbles: true }));
        }""",
        {"token": token},
    )

    await trigger_trusted_callback(page, browser_state.callback_name)

    try:
        await page.locator(browser_state.success_selector).wait_for(
            state="visible",
            timeout=15000,
        )
        return {"verified": True, "next_state": "continue"}
    except Exception:
        return {"verified": False, "next_state": "operator_review"}

from enum import Enum

class RecoveryState(str, Enum):
    DETECTED = "detected"
    SOLVING = "solving"
    VERIFIED = "verified"
    REVIEW_REQUIRED = "review_required"

ATTEMPTS: dict[str, int] = {}

async def guarded_solve(context_id: str) -> dict:
    attempts = ATTEMPTS.get(context_id, 0)
    if attempts >= 1:
        return {
            "success": False,
            "context_id": context_id,
            "next_state": RecoveryState.REVIEW_REQUIRED,
            "error": "recovery budget exhausted",
        }

    ATTEMPTS[context_id] = attempts + 1
    return await solve_recaptcha_v3(context_id)

from datetime import datetime, timezone


def recovery_event(context: CaptchaContext, result: dict) -> dict:
    return {
        "event": "recaptcha_v3_recovery",

如果您的策略将websiteKey视为配置，则不要记录它，并且永远不要记录解决方案令牌、会话cookie、API密钥、原始代理或完整的私有页面HTML。

[CapSolver错误常见问题](https://www.capsolver.com/faq/errors-and-troubleshooting) 可帮助规范错误类别。

> **优惠代码**：在 [CapSolver仪表板](https://dashboard.capsolver.com/dashboard/overview/?utm_source=offcial&utm_medium=blog&utm_campaign=how-to-solve-recaptcha-v3-in-llamaindex-agents) 使用代码 **WEBS** 可在每次充值时额外获得5%的奖励。

## 对比总结

| 集成模式 | 模型输入 | 密钥泄露风险 | 最佳用途 |
|---|---|---:|---|
| 模型提供所有任务字段 | URL、密钥、操作、代理 | 高 | 避免在生产环境中使用 |
| 使用验证字段的类型化FunctionTool | 明确字段 | 中 | 受控原型 |
| 不透明的上下文ID加服务器验证 | 仅上下文引用 | 低 | 生产环境LlamaIndex工作流 |
| 浏览器核心 `solve_on_page` | 无模型参数 | 最低 | 确定性Playwright恢复 |

不透明上下文模式让LlamaIndex代理有足够的控制权来请求恢复，而不会让它重写敏感或目标特定的参数。

## 生产检查清单

- 将API密钥和代理配置存储在密钥管理器中。
- 仅允许批准的主机和精确的工作流用途。
- 从当前实时页面中读取`websiteKey`和`pageAction`。
- 将企业模式和会话设置与目标集成匹配。
- 通过可信浏览器代码立即提交令牌。
- 在继续之前验证预期的应用程序状态。
- 仅允许一次解决尝试，然后转到操作员审核。
- 从追踪中删除令牌、cookie、代理和凭证。
- 每次SDK或提示更改时重新测试工具模式。

[CapSolver产品页面](https://www.capsolver.com/products) 列出了支持的解决方案类别，而 [CapSolver AI博客](https://www.capsolver.com/blog/ai) 讨论了相关的代理集成模式。

## 负责任使用

仅在您拥有、测试或明确获得自动化权限的应用程序上使用此工作流。技术能力不意味着访问权限。尊重目标条款、速率限制、隐私要求和认证边界。在未经授权的情况下，不要使用代理工具访问私人账户、受限记录或第三方工作流。在单独的策略和确认步骤后，将高影响操作（如提交、支付、预订和账户更改）进行隔离。

## 结论

生产环境中的LlamaIndex reCAPTCHA v3求解器应仅暴露一个狭窄、类型化的恢复功能。浏览器服务提供可信的上下文ID，服务器端代码保留精确的URL、站点密钥、操作、企业模式和代理策略，CapSolver返回一个短期解决方案，浏览器在代理继续之前验证预期状态。

通过 [CapSolver](https://www.capsolver.com/?utm_source=offcial&utm_medium=blog&utm_campaign=how-to-solve-recaptcha-v3-in-llamaindex-agents) 开始经过批准的LlamaIndex集成，在受控的测试工作流中进行测试，并在生产前添加参数定位和重试断言。

## 常见问题

### reCAPTCHA v3是否需要点击复选框？

不需要。reCAPTCHA v3基于评分，并且通常在后台运行。工作流必须保留目标的站点密钥、URL和操作。

### 为什么`pageAction`很重要？

该操作标识正在评估的操作，例如登录或提交。应使用从实时集成中读取的确切操作，而不是通用值。

### LlamaIndex代理应接收令牌吗？

优先选择服务器端提交，并仅返回已验证的状态。令牌是短期运行时数据，不应进入模型上下文或日志。

### 何时应启用会话模式？

仅在授权目标需要返回的会话值时才启用。将该值存储在短期加密运行时存储中。

### 失败尝试后应如何处理？

在配置的尝试预算后停止，记录已脱敏的诊断事件，如果适当的话刷新可信页面参数，并将工作流转到操作员审核。