Практическая реализация ReAct

Updated Jul 2026

От прототипа к продакшену

Основной цикл ReAct прост для понимания, но требует тщательной инженерной проработки для использования в продакшене. Этот файл охватывает практические вопросы: обработку ошибок, оптимизацию промптов, управление историей и тестирование самого тестового агента.

Надёжная обработка ошибок

Продакшен-агент ReAct должен обрабатывать три категории ошибок:

Категория 1: Ошибки агента (агент принимает плохое решение)

class AgentDecisionError(Exception):
    """Agent produced an unparseable or invalid action."""
    pass

def parse_decision(self, raw_response: str) -> Action:
    """Parse the LLM's response into a structured action."""
    raw = raw_response.strip()

    # Handle malformed responses
    if not raw:
        raise AgentDecisionError("Empty response from LLM")

    # Try to parse known action formats
    for prefix in ["NAVIGATE", "CLICK", "TYPE", "ASSERT", "DONE"]:
        if raw.upper().startswith(prefix):
            return Action(type=prefix, payload=raw[len(prefix):].strip())

    # If no known action matches, ask the LLM to retry
    retry_prompt = f"""
    Your previous response was not a valid action: "{raw}"
    Please respond with exactly one of:
    - NAVIGATE <url>
    - CLICK <selector>
    - TYPE <selector> <text>
    - ASSERT <condition>
    - DONE <pass|fail> <reason>
    """
    retry_response = self.llm.generate(retry_prompt)
    return self.parse_decision(retry_response)  # One retry only

Категория 2: Ошибки окружения (тестируемая система падает)

def execute_with_recovery(self, action: Action) -> ActionResult:
    """Execute an action with environment error recovery."""
    try:
        return self.execute(action)
    except ElementNotFoundError as e:
        # Try alternative selectors
        alternatives = self.browser.find_similar(action.selector)
        if alternatives:
            return ActionResult(
                status="recovered",
                message=f"Original selector '{action.selector}' not found. "
                        f"Found alternatives: {alternatives}",
                alternatives=alternatives
            )
        return ActionResult(status="failed", message=str(e))
    except TimeoutError:
        # Take a screenshot for debugging
        screenshot = self.browser.screenshot(f"timeout_step_{self.step_count}.png")
        return ActionResult(
            status="timeout",
            message="Action timed out",
            screenshot=screenshot
        )
    except NavigationError as e:
        # Page failed to load
        return ActionResult(
            status="nav_error",
            message=f"Navigation failed: {e}",
            url=self.browser.current_url()
        )

Категория 3: Инфраструктурные ошибки (инфраструктура агента падает)

def run_with_infrastructure_safety(self, objective: str) -> TestResult:
    """Run a test with infrastructure-level safety nets."""
    try:
        return self.run(objective)
    except LLMRateLimitError:
        return TestResult(status="ABORTED", reason="LLM rate limit exceeded")
    except LLMTimeoutError:
        return TestResult(status="ABORTED", reason="LLM response timeout")
    except BrowserCrashError:
        return TestResult(status="ABORTED", reason="Browser process crashed")
    except Exception as e:
        # Catch-all for unexpected errors
        return TestResult(
            status="ERROR",
            reason=f"Unexpected infrastructure error: {type(e).__name__}: {e}",
            screenshot=self.safe_screenshot()
        )

Оптимизация промптов для скорости и качества

Промпт фазы THINK -- самый критичный компонент. Каждый лишний токен в промпте стоит денег и времени.

Минимальный промпт (быстрый, дешёвый, менее точный)

minimal_prompt = f"""
Goal: {objective}
URL: {observation.url}
Text: {observation.text[:500]}
Last action: {self.history[-1] if self.history else 'none'}
Next action (NAVIGATE/CLICK/TYPE/ASSERT/DONE):"""

Когда использовать: Smoke-тесты, потоки с высокой уверенностью, среды с ограниченным бюджетом.

Стандартный промпт (сбалансированный)

standard_prompt = f"""
Objective: {objective}
Current URL: {observation.url}
Page content (truncated): {observation.text[:2000]}
Console errors: {observation.errors}
History: {self.history[-5:]}

What should I do next? Choose one:
- NAVIGATE <url>
- CLICK <selector>
- TYPE <selector> <text>
- ASSERT <condition>
- DONE <pass|fail> <reason>
"""

Когда использовать: Стандартное тестирование, staging-окружения.

Детализированный промпт (медленный, дорогой, наиболее точный)

detailed_prompt = f"""
You are a senior QA engineer testing: {objective}

CURRENT STATE:
  URL: {observation.url}
  Title: {observation.title}
  Visible text: {observation.text[:3000]}
  Visible buttons: {observation.buttons}
  Form fields: {observation.form_fields}
  Console errors: {observation.errors}
  Network failures: {observation.network_errors}

TEST HISTORY (last 10 steps):
{json.dumps(self.history[-10:], indent=2)}

CONSTRAINTS:
  - Steps remaining: {self.max_steps - self.step_count}
  - Allowed domains: {self.config.allowed_domains}
  - Do not click delete/remove buttons unless the objective requires it

REASONING REQUIRED:
1. Analyze the current state
2. Identify what has changed since the last step
3. Decide the most efficient next action toward the objective
4. Explain your reasoning in one sentence

ACTION (choose one):
  NAVIGATE <url>
  CLICK <selector>
  TYPE <selector> <text>
  ASSERT <condition>
  DONE <pass|fail> <reason>
"""

Когда использовать: Сложные потоки, исследовательское тестирование, отладочные сессии.

Стратегии управления историей

Буфер истории напрямую влияет на производительность агента. Слишком мало истории вызывает циклы; слишком много тратит токены.

Стратегия 1: Фиксированное окно

# Keep the last N steps
self.history = self.history[-5:]

Просто и предсказуемо. Хорошо работает для тестов менее 15 шагов.

Стратегия 2: Сжатая история

def summarize_history(self) -> str:
    """Compress history into a summary to save tokens."""
    if len(self.history) <= 3:
        return json.dumps(self.history)

    # Keep first step, summary of middle, and last 2 steps
    summary = f"Started at {self.history[0]['state_url']}. "
    summary += f"Took {len(self.history) - 3} intermediate steps. "
    if any(h.get('result') == 'error' for h in self.history[1:-2]):
        summary += "Encountered errors along the way. "
    summary += f"Recent actions: {json.dumps(self.history[-2:])}"
    return summary

Экономит токены, сохраняя контекст для долго работающих агентов.

Стратегия 3: История на основе вех

def milestone_history(self) -> list[dict]:
    """Keep only steps that represent significant state changes."""
    milestones = []
    last_url = None
    for step in self.history:
        # URL changed = navigation milestone
        if step.get("state_url") != last_url:
            milestones.append(step)
            last_url = step.get("state_url")
        # Error occurred = error milestone
        elif step.get("result") == "error":
            milestones.append(step)
        # Assertion made = assertion milestone
        elif "ASSERT" in step.get("action", ""):
            milestones.append(step)
    return milestones[-5:]  # Keep last 5 milestones

Лучший вариант для длинных многостраничных рабочих процессов, где важны только навигация и утверждения.

Тестирование тестового агента

Часто упускаемый из виду вопрос: как тестировать сам агент ReAct?

Модульное тестирование парсера решений

class TestDecisionParser:
    def test_parses_navigate_command(self):
        action = agent.parse_decision("NAVIGATE https://example.com")
        assert action.type == "NAVIGATE"
        assert action.payload == "https://example.com"

    def test_parses_click_with_complex_selector(self):
        action = agent.parse_decision('CLICK button[data-testid="submit"]')
        assert action.type == "CLICK"
        assert action.payload == 'button[data-testid="submit"]'

    def test_handles_malformed_response(self):
        with pytest.raises(AgentDecisionError):
            agent.parse_decision("I think we should click the button")

    def test_handles_empty_response(self):
        with pytest.raises(AgentDecisionError):
            agent.parse_decision("")

Интеграционное тестирование с мок-LLM

class TestReActAgentFlow:
    def test_completes_login_flow(self, mock_browser):
        """Agent should complete a login flow with a scripted LLM."""
        mock_llm = ScriptedLLM([
            'NAVIGATE https://app.example.com/login',
            'TYPE input[name=email] test@test.com',
            'TYPE input[name=password] secret',
            'CLICK button[type=submit]',
            'DONE pass "Successfully navigated to dashboard"',
        ])

        agent = ReActTestAgent(llm=mock_llm, browser=mock_browser)
        result = agent.run("Log in with valid credentials")

        assert result.status == "pass"
        assert result.steps_taken == 5

    def test_times_out_on_stuck_flow(self, mock_browser):
        """Agent should timeout if it cannot complete the objective."""
        mock_llm = ScriptedLLM([
            'CLICK #nonexistent-button',  # Repeated forever
        ] * 20)

        agent = ReActTestAgent(llm=mock_llm, browser=mock_browser, max_steps=5)
        result = agent.run("Click the submit button")

        assert result.status == "TIMEOUT"
        assert result.steps_taken == 5

Регрессионное тестирование поведения агента

Записывайте решения агента при успешном прогоне, затем воспроизводите для обнаружения регрессий:

class TestAgentRegression:
    def test_login_agent_matches_baseline(self, real_browser, real_llm):
        """Agent decisions should not drift from established baseline."""
        agent = ReActTestAgent(llm=real_llm, browser=real_browser)
        result = agent.run("Log in with test@test.com / password123")

        # Save the decision sequence
        decisions = [h["action"] for h in agent.history]

        # Compare against baseline (saved from a known-good run)
        baseline = load_baseline("login_test_baseline.json")

        # Allow some flexibility (exact decisions may vary)
        assert result.status == baseline["status"]
        assert len(decisions) <= baseline["max_steps"] * 1.5  # Allow 50% more steps

Бенчмарки производительности

Ключевой вывод

Продакшен-агент ReAct требует большего, чем просто основной цикл. Ему нужна надёжная обработка ошибок (ошибки агента, окружения, инфраструктуры), оптимизированные промпты для правильного соотношения скорости и качества, интеллектуальное управление историей для предотвращения лишних трат токенов и -- что критически важно -- тесты для самого агента. Инженерия вокруг цикла так же важна, как и сам цикл.