browser_loop.py•2.83 kB
#!/usr/bin/env python3
import logging
from agents.action_generation import State
from agents.browser_agent import browser_agent, scoring_agent, feedback_agent
logger = logging.getLogger(__name__)
async def browser_loop(state: State) -> State:
updated_state = state.model_copy(deep=True)
updated_state.test_results = []
updated_state.winning_task = None
for test_input in state.test_inputs:
logger.info(f"Testing: {test_input}")
current_task = state.task_description
max_attempts = 3
winning_task_for_input = None
for attempt in range(1, max_attempts + 1):
logger.info(f" Attempt {attempt}/{max_attempts}")
logger.info(f" browser_agent input: website={state.website_url}, task={current_task}, inputs={test_input}")
browser_result = await browser_agent(state.website_url, current_task, test_input)
logger.info(f" browser_agent output: {browser_result}")
logger.info(f" scoring_agent input: task={current_task}, result={browser_result}")
evaluation = await scoring_agent(current_task, str(browser_result))
logger.info(f" scoring_agent output: score={evaluation.score}/10, reasoning={evaluation.reasoning}")
if evaluation.score >= 8:
winning_task_for_input = current_task
updated_state.test_results.append({
"test_input": test_input,
"attempts": attempt,
"final_score": evaluation.score,
"status": "pass"
})
break
if attempt < max_attempts:
logger.info(f" feedback_agent input: website={state.website_url}, task={current_task}, evaluation={evaluation}")
current_task = await feedback_agent(state.website_url, current_task, evaluation)
logger.info(f" feedback_agent output: {current_task}")
else:
updated_state.test_results.append({
"test_input": test_input,
"attempts": max_attempts,
"final_score": evaluation.score,
"status": "fail"
})
# Store the winning task if this input passed
if winning_task_for_input and not updated_state.winning_task:
updated_state.winning_task = winning_task_for_input
passed = sum(1 for r in updated_state.test_results if r["status"] == "pass")
total = len(updated_state.test_results)
updated_state.result = "pass" if passed == total else "fail"
logger.info(f"Summary: {passed}/{total} passed - Result: {updated_state.result}")
logger.info(f"Winning task: {updated_state.winning_task if updated_state.winning_task else 'None (all failed)'}")
return updated_state