如何在 LangGraph 代理中解决 reCAPTCHA

用户引导的任务
      ↓
推理节点识别支持的挑战
      ↓
工具节点执行 CapSolver 工具
      ↓
结构化解决方案或标准化错误
      ↓
浏览器恢复、重试或请求人工审核

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

export CAPSOLVER_API_KEY="CAP-xxxxxxxxxxxxxxxx"
export OPENAI_API_KEY="sk-xxxxxxxxxxxxxxxx"

from capsolver_agent.langchain_tools import get_langchain_tools

tools = get_langchain_tools(api_key="YOUR_API_KEY")

import os
from typing import Literal

from capsolver_agent.langchain_tools import get_langchain_tools
from langchain_openai import ChatOpenAI
from langgraph.graph import START, StateGraph
from langgraph.graph.message import MessagesState
from langgraph.prebuilt import ToolNode, tools_condition

capsolver_tools = get_langchain_tools(
    api_key=os.environ["CAPSOLVER_API_KEY"]
)

model = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0,
).bind_tools(capsolver_tools)


def agent_node(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}


def safe_tool_error(error: Exception) -> str:
    return (
        "挑战工具失败。不要自动重试。"
        "将工作流返回给操作员审核。"
    )

builder = StateGraph(MessagesState)
builder.add_node("agent", agent_node)
builder.add_node(
    "tools",
    ToolNode(
        capsolver_tools,
        handle_tool_errors=safe_tool_error,
    ),
)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", tools_condition)
builder.add_edge("tools", "agent")

graph = builder.compile()

request = {
    "website_url": "https://staging.example.com/approved-form",
    "website_key": "6LcXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX",
}

messages = [
    (
        "system",
        "您仅在经批准的工作流中操作。如果支持的 reCAPTCHA 阻止下一步，请使用应用程序提供的精确 URL 和站点密钥调用 CapSolver solve_captcha 工具一次。"
        "永远不要发明目标或请求凭证。如果解决失败，请停止并请求操作员审核。",
    ),
    (
        "user",
        "继续经批准的预发布任务。浏览器报告在 {request['website_url']} 处有站点密钥 {request['website_key']} 的 reCAPTCHA v2。",
    ),
]

result = graph.invoke(
    {"messages": messages},
    config={"recursion_limit": 6},
)

from urllib.parse import urlparse

ALLOWED_HOSTS = {
    "staging.example.com",
    "qa.example.com",
}


def validate_target(url: str) -> str:
    parsed = urlparse(url)
    if parsed.scheme != "https":
        raise ValueError("仅允许 HTTPS 目标")
    if parsed.hostname not in ALLOWED_HOSTS:
        raise PermissionError("目标主机未获批准")
    return url

from typing import TypedDict

class RecoveryDecision(TypedDict):
    status: Literal["continue", "retry", "review"]
    reason: str


def classify_recovery(result: dict, attempt: int) -> RecoveryDecision:
    if result.get("success"):
        return {"status": "continue", "reason": "返回解决方案"}

    error = str(result.get("error", "未知错误"))
    if attempt == 0 and "timeout" in error.lower():
        return {"status": "retry", "reason": "允许一次有限重试"}

    return {"status": "review", "reason": error}

safe_event = {
    "workflow_id": "wf_01J...",
    "node": "tools",
    "tool": "solve_captcha",
    "target_host": "staging.example.com",
    "challenge_type": "recaptcha_v2",
    "attempt": 1,
    "duration_ms": 6420,
    "outcome": "success",
}