Project Guide
Build a Browser Agent with Playwright and LangGraph
Source: mortalapps.comThis guide provides a comprehensive walkthrough for building a robust browser agent AI Python Playwright system capable of automating complex multi-step web tasks. You will learn to integrate a large language model (LLM) with Playwright, a powerful browser automation library, to create an intelligent agent that can navigate, interact with forms, extract data, and handle dynamic web content. This project is ideal for AI engineers, data scientists, and developers looking to build resilient web automation solutions beyond simple scraping.
The business problem this agent solves is the need for adaptable and intelligent web interaction. Traditional web scrapers are often brittle, breaking with minor UI changes, and struggle with dynamic content or complex workflows requiring decision-making. This agent overcomes these limitations by leveraging an LLM's reasoning capabilities to dynamically plan and execute actions, making it highly resilient and versatile for tasks like lead generation, competitive analysis, or automated testing.
By the end of this project, you will have a fully functional, deployable browser agent that can interpret natural language instructions, interact with web pages, and extract structured information. The system will feature robust error handling, visual verification through screenshots, and the ability to retry actions, providing a solid foundation for production-grade web automation.
What You Will Build
You will build an intelligent browser agent that takes a natural language goal, such as "find the price of product X on website Y and add it to cart," and executes it using a headless browser. The agent will leverage Playwright to perform actions like navigating to URLs, clicking buttons, typing into input fields, selecting dropdown options, and extracting text or attributes. A core feature will be its ability to capture screenshots at each critical step, enabling the LLM to visually verify the state of the page and self-correct if an action fails or the page state is unexpected. The agent will also include robust retry mechanisms for transient UI failures and Pydantic-driven structured output extraction for consistent data retrieval. The final system will be a Python application, orchestrated by LangGraph, exposed via a FastAPI endpoint, and packaged for containerized deployment.
The final architecture consists of a LangGraph agent at its core, managing the state and orchestrating the workflow. This agent communicates with a set of custom tools built on Playwright. The agent receives a task, consults its LLM for a plan, and then invokes the appropriate Playwright tools. These tools interact with a headless browser instance, capturing screenshots and HTML content. This visual and structural information is fed back to the agent's state, allowing the LLM to observe the outcome of its actions, reflect on any discrepancies, and decide on the next best step. The system iterates through this observe-plan-act loop until the task is completed or a predefined retry limit is reached.
Technology Stack
| Component | Technology | Why This Choice |
|---|---|---|
| Large Language Model (LLM) | OpenAI GPT-4o |
Selected for its strong reasoning capabilities, multimodal input support (crucial for screenshot analysis), and excellent tool-use performance, which is vital for complex agentic workflows. |
| Orchestration Framework | LangGraph |
Chosen for its ability to define stateful, cyclic graphs, allowing for complex decision-making, iterative refinement, and robust error handling within the agent's loop. It provides a clear way to manage the agent's state transitions. |
| Memory/State Management | In-memory (Python dictionary) |
For simplicity in this guide, an in-memory dictionary is used for the agent's state. This is easily extensible to more persistent solutions like Redis or PostgreSQL checkpointers in production environments. |
| Tool Execution / Browser Automation | Playwright |
Playwright is a modern, reliable, and fast library for browser automation. It supports multiple browsers, provides robust element selection, screenshot capabilities, and is well-suited for headless operation, making it ideal for agentic web interaction. |
| Observability | OpenTelemetry (via LangSmith) |
OpenTelemetry is the industry standard for distributed tracing. Integrating it allows for deep visibility into the agent's decision-making, tool calls, and LLM interactions, which is critical for debugging and performance analysis. |
| API & Web Server | FastAPI |
FastAPI is a high-performance, easy-to-use web framework for building APIs with Python. Its asynchronous support and Pydantic integration align well with modern agent development practices. |
| Deployment | Docker / Kubernetes |
Docker provides a consistent environment for the agent and its Playwright dependencies. Kubernetes offers scalable, resilient orchestration for production deployment, handling browser instances effectively. |
OpenAI GPT-4o is chosen as the Large Language Model due to its advanced reasoning, multimodal capabilities, and strong function-calling performance. The agent needs to interpret complex instructions, plan multi-step actions, and 'see' the current state of a web page through screenshots. GPT-4o excels in these areas, making it a reliable choice for the core intelligence of the browser agent. While other LLMs like Claude 3 Opus or Google Gemini Pro are also highly capable, GPT-4o's specific multimodal strengths for visual analysis are a significant advantage for screenshot-based verification.
LangGraph serves as the orchestration framework, providing a stateful, cyclic graph structure that perfectly matches the iterative nature of an intelligent agent. Unlike simpler sequential chains, LangGraph allows the agent to loop, reflect, and retry actions based on observed outcomes, which is essential for robust web automation. Alternatives like standard LangChain chains or AutoGen could be used, but LangGraph's explicit state management and conditional routing make it superior for complex, resilient agentic workflows where decisions depend on previous actions.
Playwright is selected for browser automation due to its reliability, speed, and comprehensive feature set. It offers excellent control over headless browsers, robust element selectors, and crucial screenshot capabilities that are integral to the agent's visual verification process. Selenium is a viable alternative, but Playwright generally offers a more modern API, better performance, and easier setup, especially within containerized environments. The ability to capture detailed screenshots and HTML content is paramount for the agent's perception loop, allowing the LLM to understand the web page's state.
OpenTelemetry, integrated via LangSmith for convenience, is the chosen observability solution. It provides standardized distributed tracing, which is indispensable for debugging complex agent workflows involving multiple LLM calls and tool invocations. Without detailed traces, understanding why an agent failed or made a particular decision becomes exceptionally difficult. Alternatives include proprietary logging solutions or simpler print statements, but these lack the depth and context provided by OpenTelemetry for production systems. FastAPI is used for the API layer due to its performance, ease of use, and first-class support for asynchronous operations and Pydantic, making it an efficient choice for exposing the agent's functionality.
Project Structure
.
├── Dockerfile
├── README.md
├── requirements.txt
├── app/
│ ├── __init__.py
│ ├── main.py
│ ├── agent.py
│ ├── tools.py
│ └── state.py
└── tests/
├── __init__.py
└── test_agent.pyThe root directory contains essential project files: Dockerfile for containerization, README.md for documentation, and requirements.txt listing Python dependencies. The app/ directory houses the core application logic. main.py defines the FastAPI application and exposes the agent's API endpoint. agent.py contains the LangGraph agent definition, including the graph structure and node logic. tools.py implements the custom browser interaction tools using Playwright. state.py defines the agent's state schema using TypedDict, ensuring type safety and clarity. The tests/ directory is for unit and integration tests, with test_agent.py specifically for verifying the agent's functionality. This modular structure promotes maintainability, scalability, and testability, making it easier to manage a complex agentic system.
Implementation
Phase 1: Environment Setup and Playwright Tooling
This phase focuses on setting up the development environment, installing necessary dependencies, and creating the foundational Playwright wrapper to interact with a headless browser. We will establish basic functions for browser initialization and page navigation.
import os
import logging
from playwright.sync_api import sync_playwright, Page, Browser
from typing import Optional, Dict, Any
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class BrowserAgentTools:
def __init__(self):
self.browser: Optional[Browser] = None
self.page: Optional[Page] = None
self.playwright_context = None
def _initialize_browser(self) -> None:
"""Initializes a new Playwright browser context."""
if self.playwright_context is None:
self.playwright_context = sync_playwright().start()
if self.browser is None or not self.browser.is_connected():
try:
self.browser = self.playwright_context.chromium.launch(headless=True)
self.page = self.browser.new_page()
logger.info("Playwright browser initialized successfully.")
except Exception as e:
logger.error(f"Failed to initialize Playwright browser: {e}")
raise
def _close_browser(self) -> None:
"""Closes the Playwright browser context."""
if self.browser:
try:
self.browser.close()
logger.info("Playwright browser closed.")
except Exception as e:
logger.warning(f"Error closing browser: {e}")
finally:
self.browser = None
self.page = None
if self.playwright_context:
try:
self.playwright_context.stop()
logger.info("Playwright context stopped.")
except Exception as e:
logger.warning(f"Error stopping playwright context: {e}")
finally:
self.playwright_context = None
def navigate(self, url: str) -> Dict[str, Any]:
"""Navigates the browser to the specified URL."""
self._initialize_browser()
if not self.page: # Should not happen after _initialize_browser
logger.error("Browser page not available for navigation.")
return {"status": "error", "message": "Browser page not initialized."}
try:
logger.info(f"Navigating to URL: {url}")
self.page.goto(url, timeout=int(os.environ.get("PLAYWRIGHT_TIMEOUT_MS", 30000)))
current_url = self.page.url
screenshot_path = self.screenshot("navigate_success")
html_content = self.get_html()
logger.info(f"Successfully navigated to {current_url}")
return {"status": "success", "url": current_url, "screenshot": screenshot_path, "html_preview": html_content[:500]}
except Exception as e:
logger.error(f"Failed to navigate to {url}: {e}")
screenshot_path = self.screenshot("navigate_failure")
return {"status": "error", "message": str(e), "screenshot": screenshot_path}
def screenshot(self, name: str) -> Optional[str]:
"""Captures a screenshot of the current page."""
if not self.page:
logger.warning("Cannot take screenshot: No active page.")
return None
try:
output_dir = "screenshots"
os.makedirs(output_dir, exist_ok=True)
file_path = os.path.join(output_dir, f"browser_agent_{name}.png")
self.page.screenshot(path=file_path)
logger.info(f"Screenshot saved to {file_path}")
return file_path
except Exception as e:
logger.error(f"Failed to take screenshot: {e}")
return None
def get_html(self) -> str:
"""Returns the full HTML content of the current page."""
if not self.page:
logger.warning("Cannot get HTML: No active page.")
return ""
try:
html = self.page.content()
logger.debug("Retrieved HTML content.")
return html
except Exception as e:
logger.error(f"Failed to get HTML content: {e}")
return ""
# Example Usage (for testing this phase)
if __name__ == "__main__":
os.environ["PLAYWRIGHT_TIMEOUT_MS"] = "60000" # Set a higher timeout for local testing
browser_tools = BrowserAgentTools()
try:
print("
--- Testing navigation to example.com ---")
result = browser_tools.navigate("https://www.example.com")
print(f"Navigation Result: {result}")
print("
--- Testing screenshot ---")
screenshot_path = browser_tools.screenshot("example_page")
print(f"Screenshot Path: {screenshot_path}")
print("
--- Testing get_html ---")
html_preview = browser_tools.get_html()[:200] # Print first 200 chars
print(f"HTML Preview: {html_preview}...")
except Exception as e:
logger.critical(f"An error occurred during example usage: {e}")
finally:
browser_tools._close_browser()
print("
Browser session closed.")
This initial phase sets up the
BrowserAgentTools class, which encapsulates all Playwright interactions. The _initialize_browser method ensures a headless Chromium browser instance is launched and a new page is created. We use sync_playwright() for simplicity in this guide, but for production, async_playwright() is generally preferred with async frameworks like FastAPI. The _close_browser method handles proper resource cleanup, preventing browser zombie processes.
The navigate tool is the first core functionality, allowing the agent to visit a specified URL. It includes error handling, logs key events, and captures a screenshot both on success and failure. This immediate visual feedback is crucial for the LLM's ability to 'see' the web page state and diagnose issues. The screenshot and get_html methods provide the perceptual input for the agent, allowing it to observe the current page. All sensitive configurations, like PLAYWRIGHT_TIMEOUT_MS, are retrieved from environment variables, adhering to best practices for production-ready code. The if __name__ == "__main__" block demonstrates how to run and test these basic tools independently.
Expected outcome
A new headless Chromium browser instance will launch, navigate to `https://www.example.com`, and a screenshot (`browser_agent_navigate_success.png` and `browser_agent_example_page.png`) will be saved in a `screenshots` directory. The HTML content of the page will be retrieved, and the browser will close cleanly. All actions will be logged to the console.
How to test
Run the app/tools.py script directly (python app/tools.py). Verify that a screenshots directory is created, containing the expected PNG files, and that the console output confirms successful navigation and HTML retrieval. Check for any error messages during browser initialization or navigation attempts.
Phase 2: LangGraph Agent State and Graph Definition
In this phase, we will define the agent's state using TypedDict for clarity and type safety. We will then set up the initial LangGraph graph, defining the nodes for planning and tool execution, and establish the basic flow of the agent.
import os
import logging
from typing import List, Tuple, Annotated, TypedDict, Union, Any, Optional
import operator
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from app.tools import BrowserAgentTools # Import the tools from Phase 1
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# Initialize LLM
# Ensure OPENAI_API_KEY is set in your environment
llm = ChatOpenAI(model="gpt-4o", temperature=0.1, api_key=os.environ.get("OPENAI_API_KEY"))
class AgentState(TypedDict):
"""Represents the state of the agent in the LangGraph."""
task: str # The overall task the agent needs to accomplish
chat_history: Annotated[List[BaseMessage], operator.add] # Conversation history
current_url: str # The current URL the browser is on
last_screenshot: Optional[str] # Path to the last screenshot taken
html_preview: Optional[str] # A preview of the last HTML content
tool_output: Optional[str] # Output from the last tool execution
iterations: int # Number of iterations the agent has run
max_iterations: int # Maximum iterations allowed
# Initialize the browser tools
browser_tools_instance = BrowserAgentTools()
# Define LangChain tools from the BrowserAgentTools methods
@tool
def navigate_to_url(url: str) -> Dict[str, Any]:
"""Navigates the browser to the specified URL.
Returns status, url, screenshot path, and HTML preview.
Example: navigate_to_url(url='https://www.google.com')
"""
try:
return browser_tools_instance.navigate(url)
except Exception as e:
logger.error(f"Tool 'navigate_to_url' failed: {e}")
return {"status": "error", "message": str(e)}
@tool
def get_current_page_screenshot(name: str = "current_page") -> Optional[str]:
"""Captures a screenshot of the current page and saves it.
Returns the path to the screenshot file.
Example: get_current_page_screenshot(name='search_results')
"""
try:
return browser_tools_instance.screenshot(name)
except Exception as e:
logger.error(f"Tool 'get_current_page_screenshot' failed: {e}")
return None
@tool
def get_current_page_html() -> str:
"""Returns the full HTML content of the current page.
Example: get_current_page_html()
"""
try:
return browser_tools_instance.get_html()
except Exception as e:
logger.error(f"Tool 'get_current_page_html' failed: {e}")
return ""
# Bind tools to the LLM
llm_with_tools = llm.bind_tools([navigate_to_url, get_current_page_screenshot, get_current_page_html])
def call_llm(state: AgentState) -> AgentState:
"""Invokes the LLM to decide the next action based on the current state."""
logger.info("Calling LLM for next action...")
current_iterations = state.get("iterations", 0) + 1
if current_iterations > state["max_iterations"]:
logger.warning(f"Max iterations ({state['max_iterations']}) reached. Ending task.")
return {**state, "iterations": current_iterations, "tool_output": "Max iterations reached. Cannot complete task."}
messages = list(state["chat_history"]) # copy to avoid mutating state
# Add current browser context to the prompt for the LLM
if state.get("last_screenshot"):
messages.append(AIMessage(content=f"Last screenshot: {state['last_screenshot']}"))
if state.get("html_preview"):
messages.append(AIMessage(content=f"Current page HTML preview: {state['html_preview'][:500]}..."))
if state.get("tool_output"):
messages.append(AIMessage(content=f"Previous tool output: {state['tool_output']}"))
try:
response = llm_with_tools.invoke(messages)
logger.info("LLM invoked successfully.")
return {
**state,
"chat_history": state["chat_history"] + [response],
"iterations": current_iterations
}
except Exception as e:
logger.error(f"LLM invocation failed: {e}")
return {**state, "chat_history": state["chat_history"] + [AIMessage(content=f"LLM error: {e}")], "tool_output": f"LLM error: {e}"}
def call_tool(state: AgentState) -> AgentState:
"""Executes the tool suggested by the LLM."""
logger.info("Calling tool based on LLM suggestion...")
messages = state["chat_history"]
last_message = messages[-1]
tool_output = ""
if not last_message.tool_calls:
logger.warning("No tool calls found in the last LLM message.")
return {**state, "tool_output": "No tool calls suggested by LLM."}
for tool_call in last_message.tool_calls:
tool_name = tool_call['name']
tool_args = tool_call['args']
logger.info(f"Attempting to execute tool: {tool_name} with args: {tool_args}")
try:
# Directly invoke the tool function by name
# This requires the tool functions to be accessible in the global scope or passed explicitly
if tool_name == "navigate_to_url":
output = navigate_to_url.invoke(tool_args)
elif tool_name == "get_current_page_screenshot":
output = get_current_page_screenshot.invoke(tool_args)
elif tool_name == "get_current_page_html":
output = get_current_page_html.invoke(tool_args)
else:
output = f"Unknown tool: {tool_name}"
logger.warning(output)
tool_output += output + "
"
continue
tool_output += str(output) + "
"
logger.info(f"Tool '{tool_name}' executed. Output: {output}")
# Update state based on tool output
if isinstance(output, dict):
if "url" in output:
state["current_url"] = output["url"]
if "screenshot" in output:
state["last_screenshot"] = output["screenshot"]
if "html_preview" in output:
state["html_preview"] = output["html_preview"]
except Exception as e:
logger.error(f"Error executing tool '{tool_name}': {e}")
tool_output += f"Error executing tool '{tool_name}': {e}
"
return {
**state,
"chat_history": state["chat_history"] + [AIMessage(content=tool_output)],
"tool_output": tool_output
}
def should_continue(state: AgentState) -> str:
"""Determines whether the agent should continue or end the task."""
if state["iterations"] >= state["max_iterations"]:
logger.info("Max iterations reached. Ending graph.")
return "end"
messages = state["chat_history"]
last_message = messages[-1]
if last_message.tool_calls: # If the LLM suggested a tool, continue to call it
logger.info("LLM suggested tool call. Continuing to tool execution.")
return "continue"
# If the LLM did not suggest a tool, it means it's done or stuck.
logger.info("LLM did not suggest a tool. Ending graph.")
return "end"
# Build the LangGraph graph
workflow = StateGraph(AgentState)
workflow.add_node("llm_node", call_llm)
workflow.add_node("tool_node", call_tool)
workflow.set_entry_point("llm_node")
workflow.add_conditional_edges(
"llm_node",
should_continue,
{
"continue": "tool_node",
"end": END
}
)
workflow.add_edge('tool_node', 'llm_node') # After tool execution, go back to LLM to decide next step
# Compile the graph
app = workflow.compile()
# Example usage (for testing this phase)
if __name__ == "__main__":
os.environ["OPENAI_API_KEY"] = os.environ.get("OPENAI_API_KEY", "YOUR_OPENAI_API_KEY") # Replace with your key or set env var
os.environ["PLAYWRIGHT_TIMEOUT_MS"] = "60000"
if os.environ["OPENAI_API_KEY"] == "YOUR_OPENAI_API_KEY":
logger.error("Please set your OPENAI_API_KEY environment variable.")
exit(1)
initial_state: AgentState = {
"task": "Navigate to example.com and tell me the main heading.",
"chat_history": [HumanMessage(content="Navigate to example.com and tell me the main heading.")],
"current_url": "",
"last_screenshot": None,
"html_preview": None,
"tool_output": None,
"iterations": 0,
"max_iterations": 5
}
try:
for s in app.stream(initial_state):
print(s)
print("------")
print("
Agent execution complete.")
except Exception as e:
logger.critical(f"An error occurred during agent execution: {e}")
finally:
# Ensure browser is closed even if agent errors out
browser_tools_instance._close_browser()
This phase introduces the core of our agent: the
AgentState and the LangGraph workflow. AgentState is defined as a TypedDict, providing a clear and type-safe schema for all information the agent needs to maintain, such as the task, chat_history, current_url, and critical last_screenshot and html_preview. The Annotated[List[BaseMessage], operator.add] for chat_history is a LangGraph specific syntax that tells the graph to append new messages to the list, rather than overwriting it.
We then define LangChain tool wrappers around the BrowserAgentTools methods from Phase 1. These @tool decorated functions allow the LLM to understand and invoke the browser actions. The llm_with_tools object binds these tools to the ChatOpenAI instance, enabling the LLM to generate tool calls in its responses.
The call_llm node is responsible for invoking the LLM, passing the current chat_history and relevant context like the last_screenshot and html_preview. This contextual information is critical for the LLM to make informed decisions. The call_tool node parses the LLM's tool call suggestions and executes the corresponding BrowserAgentTools method, updating the AgentState with the outcomes, including the new URL, screenshot path, and HTML.
The should_continue conditional edge determines the flow: if the LLM suggests a tool, the graph transitions to tool_node; otherwise, if the task is complete or the LLM is stuck, the graph ends. This establishes the fundamental observe-plan-act loop. An iteration counter max_iterations is added to prevent infinite loops, a common issue in agentic systems. The if __name__ == "__main__" block demonstrates how to run this basic agent, ensuring you set OPENAI_API_KEY.
Expected outcome
The agent will initialize, the LLM will be invoked, and it will likely suggest `navigate_to_url` for `https://www.example.com`. This tool will execute, updating the state with the new URL, a screenshot path, and HTML preview. The LLM will then be called again with this updated context. The agent will run for a few iterations (up to `max_iterations`) and then terminate, printing the final state and chat history. Screenshots will still be generated.
How to test
Set your OPENAI_API_KEY environment variable. Run python app/agent.py. Observe the console output for LLM calls and tool executions. Verify that the navigate_to_url tool is called and that new screenshots are generated in the screenshots directory. The final_state should reflect the navigation to example.com.
Phase 3: Advanced Browser Interaction Tools
This phase expands the agent's capabilities by adding more sophisticated Playwright tools for interacting with web elements, such as clicking, typing, and extracting text based on CSS selectors. These tools are crucial for automating complex forms and data retrieval.
import os
import logging
from playwright.sync_api import sync_playwright, Page, Browser, expect
from typing import Optional, Dict, Any
from tenacity import retry, stop_after_attempt, wait_exponential, before_log
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class BrowserAgentTools:
def __init__(self):
self.browser: Optional[Browser] = None
self.page: Optional[Page] = None
self.playwright_context = None
def _initialize_browser(self) -> None:
"""Initializes a new Playwright browser context."""
if self.playwright_context is None:
self.playwright_context = sync_playwright().start()
if self.browser is None or not self.browser.is_connected():
try:
self.browser = self.playwright_context.chromium.launch(headless=True)
self.page = self.browser.new_page()
logger.info("Playwright browser initialized successfully.")
except Exception as e:
logger.error(f"Failed to initialize Playwright browser: {e}")
raise
def _close_browser(self) -> None:
"""Closes the Playwright browser context."""
if self.browser:
try:
self.browser.close()
logger.info("Playwright browser closed.")
except Exception as e:
logger.warning(f"Error closing browser: {e}")
finally:
self.browser = None
self.page = None
if self.playwright_context:
try:
self.playwright_context.stop()
logger.info("Playwright context stopped.")
except Exception as e:
logger.warning(f"Error stopping playwright context: {e}")
finally:
self.playwright_context = None
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10), before=before_log(logger, logging.INFO))
def navigate(self, url: str) -> Dict[str, Any]:
"""Navigates the browser to the specified URL."""
self._initialize_browser()
if not self.page:
logger.error("Browser page not available for navigation.")
return {"status": "error", "message": "Browser page not initialized."}
try:
logger.info(f"Navigating to URL: {url}")
self.page.goto(url, timeout=int(os.environ.get("PLAYWRIGHT_TIMEOUT_MS", 30000)))
current_url = self.page.url
screenshot_path = self.screenshot("navigate_success")
html_content = self.get_html()
logger.info(f"Successfully navigated to {current_url}")
return {"status": "success", "url": current_url, "screenshot": screenshot_path, "html_preview": html_content[:500]}
except Exception as e:
logger.error(f"Failed to navigate to {url}: {e}")
screenshot_path = self.screenshot("navigate_failure")
raise # Re-raise to trigger retry logic
def screenshot(self, name: str = "current_page") -> Optional[str]:
"""Captures a screenshot of the current page."""
if not self.page:
logger.warning("Cannot take screenshot: No active page.")
return None
try:
output_dir = "screenshots"
os.makedirs(output_dir, exist_ok=True)
file_path = os.path.join(output_dir, f"browser_agent_{name}.png")
self.page.screenshot(path=file_path)
logger.info(f"Screenshot saved to {file_path}")
return file_path
except Exception as e:
logger.error(f"Failed to take screenshot: {e}")
return None
def get_html(self) -> str:
"""Returns the full HTML content of the current page."""
if not self.page:
logger.warning("Cannot get HTML: No active page.")
return ""
try:
html = self.page.content()
logger.debug("Retrieved HTML content.")
return html
except Exception as e:
logger.error(f"Failed to get HTML content: {e}")
return ""
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=5), before=before_log(logger, logging.INFO))
def click_element(self, selector: str) -> Dict[str, Any]:
"""Clicks an element identified by a CSS selector."""
self._initialize_browser()
if not self.page:
return {"status": "error", "message": "Browser page not initialized."}
try:
logger.info(f"Attempting to click element with selector: {selector}")
self.page.click(selector, timeout=int(os.environ.get("PLAYWRIGHT_ACTION_TIMEOUT_MS", 10000)))
screenshot_path = self.screenshot(f"click_success_{selector.replace(' ', '_')[:20]}")
html_content = self.get_html()
logger.info(f"Successfully clicked element: {selector}")
return {"status": "success", "selector": selector, "screenshot": screenshot_path, "html_preview": html_content[:500]}
except Exception as e:
logger.error(f"Failed to click element '{selector}': {e}")
screenshot_path = self.screenshot(f"click_failure_{selector.replace(' ', '_')[:20]}")
raise # Re-raise to trigger retry logic
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=5), before=before_log(logger, logging.INFO))
def type_text(self, selector: str, text: str) -> Dict[str, Any]:
"""Types text into an input field identified by a CSS selector."""
self._initialize_browser()
if not self.page:
return {"status": "error", "message": "Browser page not initialized."}
try:
logger.info(f"Attempting to type text '{text[:20]}...' into selector: {selector}")
self.page.fill(selector, text, timeout=int(os.environ.get("PLAYWRIGHT_ACTION_TIMEOUT_MS", 10000)))
screenshot_path = self.screenshot(f"type_success_{selector.replace(' ', '_')[:20]}")
html_content = self.get_html()
logger.info(f"Successfully typed into element: {selector}")
return {"status": "success", "selector": selector, "text": text, "screenshot": screenshot_path, "html_preview": html_content[:500]}
except Exception as e:
logger.error(f"Failed to type text into '{selector}': {e}")
screenshot_path = self.screenshot(f"type_failure_{selector.replace(' ', '_')[:20]}")
raise # Re-raise to trigger retry logic
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=5), before=before_log(logger, logging.INFO))
def extract_text(self, selector: str) -> Dict[str, Any]:
"""Extracts text content from an element identified by a CSS selector."""
self._initialize_browser()
if not self.page:
return {"status": "error", "message": "Browser page not initialized."}
try:
logger.info(f"Attempting to extract text from selector: {selector}")
element = self.page.locator(selector)
text_content = element.text_content(timeout=int(os.environ.get("PLAYWRIGHT_ACTION_TIMEOUT_MS", 10000)))
if text_content is None:
raise ValueError(f"No text content found for selector: {selector}")
screenshot_path = self.screenshot(f"extract_success_{selector.replace(' ', '_')[:20]}")
html_content = self.get_html()
logger.info(f"Successfully extracted text from {selector}: {text_content[:50]}...")
return {"status": "success", "selector": selector, "extracted_text": text_content, "screenshot": screenshot_path, "html_preview": html_content[:500]}
except Exception as e:
logger.error(f"Failed to extract text from '{selector}': {e}")
screenshot_path = self.screenshot(f"extract_failure_{selector.replace(' ', '_')[:20]}")
raise # Re-raise to trigger retry logic
# Example Usage (for testing this phase)
if __name__ == "__main__":
os.environ["PLAYWRIGHT_TIMEOUT_MS"] = "60000"
os.environ["PLAYWRIGHT_ACTION_TIMEOUT_MS"] = "15000"
browser_tools = BrowserAgentTools()
try:
print("
--- Testing navigation to a test page ---")
browser_tools.navigate("https://www.google.com") # Using Google as a test page
print("
--- Testing typing into search bar ---")
# Google search bar selector might be 'textarea[name="q"]' or 'input[name="q"]'
# We'll use a common one, but it might need adjustment based on Google's current UI.
# A more robust agent would inspect HTML for the correct selector.
search_selector = "textarea[name='q'], input[name='q']"
type_result = browser_tools.type_text(search_selector, "Playwright Python")
print(f"Type Result: {type_result}")
print("
--- Testing clicking search button ---")
# Google search button selector can be complex, often involves forms or specific data attributes.
# Attempting a common one, but may need adjustment.
# A more robust agent would inspect HTML for the correct selector.
click_result = browser_tools.click_element("input[name='btnK']") # This is often for 'Google Search' button
print(f"Click Result: {click_result}")
print("
--- Testing extracting text from results ---")
# Extracting title of the page after search
title_result = browser_tools.extract_text("title")
print(f"Title Extraction Result: {title_result}")
except Exception as e:
logger.critical(f"An error occurred during example usage: {e}")
finally:
browser_tools._close_browser()
print("
Browser session closed.")
This phase significantly enhances the
BrowserAgentTools by adding click_element, type_text, and extract_text methods. These are fundamental for any interactive web automation. Each method takes a CSS selector to identify the target element, allowing the agent to precisely interact with the page.
Crucially, we've integrated tenacity for robust retry logic with exponential backoff. Web elements might not be immediately available due to dynamic loading, network latency, or other transient issues. Applying @retry to navigate, click_element, type_text, and extract_text makes these tools much more resilient. If an operation fails, it will be retried a few times before giving up, significantly improving the agent's reliability. The before_log argument ensures that retry attempts are logged, providing clear visibility into transient failures.
Each new tool also captures a screenshot on both success and failure, and returns a html_preview. This ensures that the LLM always has the most up-to-date visual and structural context after any interaction, enabling it to detect if an action had the desired effect or if the page state changed unexpectedly. The example usage now demonstrates a more complex sequence: navigating, typing into a search bar, clicking a search button, and extracting the page title, simulating a common multi-step web task. Note that specific selectors for dynamic sites like Google might change and require adaptation.
Expected outcome
The agent will be able to perform navigation, type text into input fields, click elements, and extract text based on CSS selectors. The retry logic will handle transient failures gracefully. Multiple screenshots will be generated in the `screenshots` directory, reflecting each step of the interaction (navigation, typing, clicking). The console output will show successful execution of these new tools, including retry attempts if any occur.
How to test
Update app/tools.py with the new code. Run python app/tools.py. Observe the console logs for successful tool executions and any retry attempts. Check the screenshots directory to confirm screenshots are captured at each stage, indicating the browser's state after typing and clicking. You might need to adjust the search_selector and click_selector in the if __name__ == "__main__" block if Google's UI has changed.
Phase 4: Agent Logic: Planning, Execution, and Reflection
This phase refines the agent's core decision-making loop. We'll enhance the call_llm node to use a more sophisticated prompt that includes the available tools, current browser context (screenshot, HTML), and previous tool outputs, allowing the LLM to plan more effectively and self-correct. We'll also update the call_tool node to better handle tool outputs.
import os
import logging
from typing import List, Tuple, Annotated, TypedDict, Union, Any, Optional, Callable, Dict
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from tenacity import retry, stop_after_attempt, wait_exponential, before_log, RetryError
from app.tools import BrowserAgentTools # Import the updated tools from Phase 3
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# Initialize LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0.1, api_key=os.environ.get("OPENAI_API_KEY"))
class AgentState(TypedDict):
"""Represents the state of the agent in the LangGraph."""
task: str # The overall task the agent needs to accomplish
chat_history: Annotated[List[BaseMessage], "operator.add"] # Conversation history
current_url: str # The current URL the browser is on
last_screenshot: Optional[str] # Path to the last screenshot taken
html_preview: Optional[str] # A preview of the last HTML content
tool_output: Optional[str] # Output from the last tool execution
iterations: int # Number of iterations the agent has run
max_iterations: int # Maximum iterations allowed
error_message: Optional[str] # Last error message encountered
# Initialize the browser tools
browser_tools_instance = BrowserAgentTools()
# Define LangChain tools from the BrowserAgentTools methods
@tool
def navigate_to_url(url: str) -> Dict[str, Any]:
"""Navigates the browser to the specified URL.
Returns status, url, screenshot path, and HTML preview.
Example: navigate_to_url(url='https://www.google.com')
"""
try:
return browser_tools_instance.navigate(url)
except RetryError as e:
logger.error(f"Tool 'navigate_to_url' failed after retries: {e}")
return {"status": "error", "message": f"Failed to navigate to {url} after multiple retries: {e}"}
except Exception as e:
logger.error(f"Tool 'navigate_to_url' failed: {e}")
return {"status": "error", "message": str(e)}
@tool
def get_current_page_screenshot(name: str = "current_page") -> Optional[str]:
"""Captures a screenshot of the current page and saves it.
Returns the path to the screenshot file.
Example: get_current_page_screenshot(name='search_results')
"""
try:
return browser_tools_instance.screenshot(name)
except Exception as e:
logger.error(f"Tool 'get_current_page_screenshot' failed: {e}")
return None
@tool
def get_current_page_html() -> str:
"""Returns the full HTML content of the current page.
Example: get_current_page_html()
"""
try:
return browser_tools_instance.get_html()
except Exception as e:
logger.error(f"Tool 'get_current_page_html' failed: {e}")
return ""
@tool
def click_element(selector: str) -> Dict[str, Any]:
"""Clicks an element identified by a CSS selector.
Example: click_element(selector='button.submit-button')
"""
try:
return browser_tools_instance.click_element(selector)
except RetryError as e:
logger.error(f"Tool 'click_element' failed after retries: {e}")
return {"status": "error", "message": f"Failed to click element {selector} after multiple retries: {e}"}
except Exception as e:
logger.error(f"Tool 'click_element' failed: {e}")
return {"status": "error", "message": str(e)}
@tool
def type_text(selector: str, text: str) -> Dict[str, Any]:
"""Types text into an input field identified by a CSS selector.
Example: type_text(selector='input#username', text='myuser')
"""
try:
return browser_tools_instance.type_text(selector, text)
except RetryError as e:
logger.error(f"Tool 'type_text' failed after retries: {e}")
return {"status": "error", "message": f"Failed to type text into {selector} after multiple retries: {e}"}
except Exception as e:
logger.error(f"Tool 'type_text' failed: {e}")
return {"status": "error", "message": str(e)}
@tool
def extract_text(selector: str) -> Dict[str, Any]:
"""Extracts text content from an element identified by a CSS selector.
Returns the extracted text.
Example: extract_text(selector='h1.page-title')
"""
try:
return browser_tools_instance.extract_text(selector)
except RetryError as e:
logger.error(f"Tool 'extract_text' failed after retries: {e}")
return {"status": "error", "message": f"Failed to extract text from {selector} after multiple retries: {e}"}
except Exception as e:
logger.error(f"Tool 'extract_text' failed: {e}")
return {"status": "error", "message": str(e)}
# Map tool names to their functions for dynamic invocation
tools_map: Dict[str, Callable] = {
"navigate_to_url": navigate_to_url,
"get_current_page_screenshot": get_current_page_screenshot,
"get_current_page_html": get_current_page_html,
"click_element": click_element,
"type_text": type_text,
"extract_text": extract_text,
}
# Bind tools to the LLM
llm_with_tools = llm.bind_tools(list(tools_map.values()))
def call_llm(state: AgentState) -> AgentState:
"""Invokes the LLM to decide the next action based on the current state."""
logger.info("Calling LLM for next action...")
current_iterations = state.get("iterations", 0) + 1
if current_iterations > state["max_iterations"]:
logger.warning(f"Max iterations ({state['max_iterations']}) reached. Ending task.")
return {**state, "iterations": current_iterations, "tool_output": "Max iterations reached. Cannot complete task.", "error_message": "Max iterations reached."}
messages: List[Union[BaseMessage, Dict[str, Any]]] = []
# Add the initial task as a HumanMessage
if current_iterations == 1:
messages.append(HumanMessage(content=state["task"]))
else:
messages.extend(state["chat_history"])
# Prepare visual and HTML context for the LLM
llm_context = []
if state.get("last_screenshot"):
llm_context.append({"type": "image_url", "image_url": {"url": f"file://{state['last_screenshot']}"}})
if state.get("html_preview"):
llm_context.append({"type": "text", "text": f"Current page HTML preview: {state['html_preview']}"})
if state.get("tool_output"):
llm_context.append({"type": "text", "text": f"Previous tool output: {state['tool_output']}"})
if state.get("error_message"):
llm_context.append({"type": "text", "text": f"Previous error: {state['error_message']}. Please try to recover or adjust your approach."})
# Add current context as part of the last message if not empty
if llm_context and messages and isinstance(messages[-1], HumanMessage):
messages[-1].content = messages[-1].content if isinstance(messages[-1].content, str) else list(messages[-1].content)
messages[-1].content.extend(llm_context)
elif llm_context:
# If last message isn't HumanMessage or messages is empty, create a new AIMessage with context
messages.append(AIMessage(content=llm_context))
try:
# Ensure the final message content is correctly formatted for multimodal LLMs
if isinstance(messages[-1].content, str):
final_content = messages[-1].content
else:
final_content = []
for item in messages[-1].content:
if isinstance(item, str):
final_content.append({"type": "text", "text": item})
else:
final_content.append(item)
messages[-1].content = final_content
response = llm_with_tools.invoke(messages)
logger.info("LLM invoked successfully, response received.")
return {
**state,
"chat_history": state["chat_history"] + [response],
"iterations": current_iterations,
"error_message": None # Clear error after LLM runs
}
except Exception as e:
logger.error(f"LLM invocation failed: {e}")
return {**state, "chat_history": state["chat_history"] + [AIMessage(content=f"LLM error: {e}")], "tool_output": f"LLM error: {e}", "error_message": str(e)}
def call_tool(state: AgentState) -> AgentState:
"""Executes the tool suggested by the LLM."""
logger.info("Calling tool based on LLM suggestion...")
messages = state["chat_history"]
last_message = messages[-1]
tool_outputs: List[str] = []
error_occurred = False
if not last_message.tool_calls:
logger.warning("No tool calls found in the last LLM message.")
return {**state, "tool_output": "No tool calls suggested by LLM.", "error_message": "No tool calls suggested by LLM."}
for tool_call in last_message.tool_calls:
tool_name = tool_call['name']
tool_args = tool_call['args']
logger.info(f"Attempting to execute tool: {tool_name} with args: {tool_args}")
try:
if tool_name in tools_map:
output = tools_map[tool_name].invoke(tool_args)
tool_outputs.append(str(output))
logger.info(f"Tool '{tool_name}' executed. Output: {output}")
# Update state based on tool output
if isinstance(output, dict):
if "url" in output and output.get("status") == "success":
state["current_url"] = output["url"]
if "screenshot" in output:
state["last_screenshot"] = output["screenshot"]
if "html_preview" in output:
state["html_preview"] = output["html_preview"]
if output.get("status") == "error":
error_occurred = True
state["error_message"] = output.get("message", f"Tool {tool_name} failed.")
logger.error(f"Tool '{tool_name}' reported error: {state['error_message']}")
else:
output = f"Unknown tool: {tool_name}"
logger.warning(output)
tool_outputs.append(output)
error_occurred = True
state["error_message"] = output
except Exception as e:
logger.error(f"Error executing tool '{tool_name}': {e}")
tool_outputs.append(f"Error executing tool '{tool_name}': {e}
")
error_occurred = True
state["error_message"] = str(e)
# Add a ToolMessage to chat history for the LLM to see the result of its tool call
tool_message = ToolMessage(content="
".join(tool_outputs), tool_call_id=last_message.tool_calls[0]['id'] if last_message.tool_calls else "unknown")
return {
**state,
"chat_history": state["chat_history"] + [tool_message],
"tool_output": "
".join(tool_outputs),
"error_message": state["error_message"] if error_occurred else None # Keep error if it occurred
}
def should_continue(state: AgentState) -> str:
"""Determines whether the agent should continue or end the task."""
if state["iterations"] >= state["max_iterations"]:
logger.info("Max iterations reached. Ending graph.")
return "end"
if state.get("error_message") and state["iterations"] > 1: # If an error occurred and it's not the very first step
logger.warning(f"An error occurred: {state['error_message']}. Attempting LLM to recover.")
return "continue" # Allow LLM to attempt recovery
messages = state["chat_history"]
last_message = messages[-1]
# If the last message is a ToolMessage, the LLM needs to process its output
if isinstance(last_message, ToolMessage):
logger.info("Last message is ToolMessage. Continuing to LLM for processing.")
return "continue"
# If the LLM suggested a tool, continue to call it
if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
logger.info("LLM suggested tool call. Continuing to tool execution.")
return "continue"
# If the LLM did not suggest a tool and it's not a ToolMessage, it means it's done or stuck.
logger.info("LLM did not suggest a tool and is not processing a tool output. Ending graph.")
return "end"
# Build the LangGraph graph
workflow = StateGraph(AgentState)
workflow.add_node("llm_node", call_llm)
workflow.add_node("tool_node", call_tool)
workflow.set_entry_point("llm_node")
workflow.add_conditional_edges(
"llm_node",
should_continue,
{
"continue": "tool_node",
"end": END
}
)
workflow.add_conditional_edges(
"tool_node",
should_continue,
{
"continue": "llm_node",
"end": END
}
)
# Compile the graph
app = workflow.compile()
# Example usage (for testing this phase)
if __name__ == "__main__":
os.environ["OPENAI_API_KEY"] = os.environ.get("OPENAI_API_KEY", "YOUR_OPENAI_API_KEY")
os.environ["PLAYWRIGHT_TIMEOUT_MS"] = "60000"
os.environ["PLAYWRIGHT_ACTION_TIMEOUT_MS"] = "15000"
if os.environ["OPENAI_API_KEY"] == "YOUR_OPENAI_API_KEY":
logger.error("Please set your OPENAI_API_KEY environment variable.")
exit(1)
initial_task = "Navigate to mortalapps.com/agents/projects/ and find the title of the 'AI Research Agent' project. Then, navigate to the project page and extract the first sentence of its overview."
initial_state: AgentState = {
"task": initial_task,
"chat_history": [HumanMessage(content=initial_task)],
"current_url": "",
"last_screenshot": None,
"html_preview": None,
"tool_output": None,
"iterations": 0,
"max_iterations": 10,
"error_message": None
}
try:
print(f"
--- Starting agent for task: {initial_task} ---")
for s in app.stream(initial_state):
print(s)
print("------")
final_state = app.invoke(initial_state)
print("
Final State:")
print(f"Task: {final_state['task']}")
print(f"Final URL: {final_state['current_url']}")
print(f"Final Chat History: {[msg.content for msg in final_state['chat_history']]}")
print(f"Final Tool Output: {final_state['tool_output']}")
print(f"Final Error Message: {final_state['error_message']}")
except Exception as e:
logger.critical(f"An error occurred during agent execution: {e}")
finally:
browser_tools_instance._close_browser()
print("
Browser session closed.")
This phase significantly enhances the agent's intelligence by refining the
call_llm and call_tool nodes, and updating the LangGraph flow. We've added error_message to AgentState to explicitly track and communicate failures to the LLM, enabling it to attempt recovery. The tools_map dictionary allows for dynamic tool invocation, simplifying call_tool.
The call_llm node now constructs a more comprehensive prompt for the LLM. For multimodal models like GPT-4o, the last_screenshot is included as an image_url part of the message content, allowing the LLM to visually interpret the page. The html_preview and tool_output are also included as text, providing rich context. Crucially, if an error_message exists, it's explicitly passed to the LLM, prompting it to reflect and attempt to self-correct. The message formatting for multimodal input is carefully handled to ensure compatibility with ChatOpenAI.
The call_tool node is updated to properly process the output of each tool, especially handling status and error messages. If a tool reports an error, the error_message in the state is updated. The should_continue function is made more intelligent: it now checks for error_message in the state. If an error occurred, it directs the flow back to the llm_node to allow the LLM to decide on a recovery strategy, rather than immediately ending. This makes the agent significantly more resilient to unexpected web conditions. The example task demonstrates a more complex information retrieval scenario, requiring multiple steps of navigation and extraction.
Expected outcome
The agent will attempt to fulfill the multi-step task: navigate to a specific projects page, identify a project title, navigate to that project's detail page, and extract specific text. The LLM will use the visual (screenshots) and textual (HTML, tool output, errors) context to make decisions. If a tool fails, the LLM will receive an error message and attempt to devise a new plan or retry. The final `tool_output` or `chat_history` should contain the extracted information.
How to test
Update app/agent.py with the new code. Ensure app/tools.py is also updated from Phase 3. Run python app/agent.py. Monitor the console output closely to observe the agent's planning, tool calls, and state updates. Verify that the agent successfully navigates, identifies the project, and extracts the requested information. Pay attention to how it handles any transient errors (e.g., if a selector is initially not found) and if it recovers correctly.
Phase 5: Structured Output Extraction
This phase introduces the ability for the agent to extract structured data from web pages using Pydantic. We'll define a Pydantic model for the expected output and teach the LLM to generate responses conforming to this schema, making the agent's output reliable and machine-readable.
import os
import logging
from typing import List, Tuple, Annotated, TypedDict, Union, Any, Optional, Callable, Dict
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from tenacity import retry, stop_after_attempt, wait_exponential, before_log, RetryError
from pydantic import BaseModel, Field
import json
from app.tools import BrowserAgentTools # Import the updated tools from Phase 3
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# Initialize LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0.1, api_key=os.environ.get("OPENAI_API_KEY"))
class AgentState(TypedDict):
"""Represents the state of the agent in the LangGraph."""
task: str # The overall task the agent needs to accomplish
chat_history: Annotated[List[BaseMessage], "operator.add"] # Conversation history
current_url: str # The current URL the browser is on
last_screenshot: Optional[str] # Path to the last screenshot taken
html_preview: Optional[str] # A preview of the last HTML content
tool_output: Optional[str] # Output from the last tool execution
iterations: int # Number of iterations the agent has run
max_iterations: int # Maximum iterations allowed
error_message: Optional[str] # Last error message encountered
final_result: Optional[Dict[str, Any]] # Structured output from the agent
# Initialize the browser tools
browser_tools_instance = BrowserAgentTools()
# Define LangChain tools from the BrowserAgentTools methods
@tool
def navigate_to_url(url: str) -> Dict[str, Any]:
"""Navigates the browser to the specified URL.
Returns status, url, screenshot path, and HTML preview.
Example: navigate_to_url(url='https://www.google.com')
"""
try:
return browser_tools_instance.navigate(url)
except RetryError as e:
logger.error(f"Tool 'navigate_to_url' failed after retries: {e}")
return {"status": "error", "message": f"Failed to navigate to {url} after multiple retries: {e}"}
except Exception as e:
logger.error(f"Tool 'navigate_to_url' failed: {e}")
return {"status": "error", "message": str(e)}
@tool
def get_current_page_screenshot(name: str = "current_page") -> Optional[str]:
"""Captures a screenshot of the current page and saves it.
Returns the path to the screenshot file.
Example: get_current_page_screenshot(name='search_results')
"""
try:
return browser_tools_instance.screenshot(name)
except Exception as e:
logger.error(f"Tool 'get_current_page_screenshot' failed: {e}")
return None
@tool
def get_current_page_html() -> str:
"""Returns the full HTML content of the current page.
Example: get_current_page_html()
"""
try:
return browser_tools_instance.get_html()
except Exception as e:
logger.error(f"Tool 'get_current_page_html' failed: {e}")
return ""
@tool
def click_element(selector: str) -> Dict[str, Any]:
"""Clicks an element identified by a CSS selector.
Example: click_element(selector='button.submit-button')
"""
try:
return browser_tools_instance.click_element(selector)
except RetryError as e:
logger.error(f"Tool 'click_element' failed after retries: {e}")
return {"status": "error", "message": f"Failed to click element {selector} after multiple retries: {e}"}
except Exception as e:
logger.error(f"Tool 'click_element' failed: {e}")
return {"status": "error", "message": str(e)}
@tool
def type_text(selector: str, text: str) -> Dict[str, Any]:
"""Types text into an input field identified by a CSS selector.
Example: type_text(selector='input#username', text='myuser')
"""
try:
return browser_tools_instance.type_text(selector, text)
except RetryError as e:
logger.error(f"Tool 'type_text' failed after retries: {e}")
return {"status": "error", "message": f"Failed to type text into {selector} after multiple retries: {e}"}
except Exception as e:
logger.error(f"Tool 'type_text' failed: {e}")
return {"status": "error", "message": str(e)}
@tool
def extract_text(selector: str) -> Dict[str, Any]:
"""Extracts text content from an element identified by a CSS selector.
Returns the extracted text.
Example: extract_text(selector='h1.page-title')
"""
try:
return browser_tools_instance.extract_text(selector)
except RetryError as e:
logger.error(f"Tool 'extract_text' failed after retries: {e}")
return {"status": "error", "message": f"Failed to extract text from {selector} after multiple retries: {e}"}
except Exception as e:
logger.error(f"Tool 'extract_text' failed: {e}")
return {"status": "error", "message": str(e)}
# Define Pydantic model for structured output
class ProjectInfo(BaseModel):
title: str = Field(description="The title of the project.")
overview_first_sentence: str = Field(description="The first sentence of the project's overview.")
project_url: str = Field(description="The URL of the project page.")
@tool
def store_structured_data(data: ProjectInfo) -> str:
"""Stores the extracted structured data into the agent's final_result state.
Call this tool when all required information has been extracted and validated.
Example: store_structured_data(data={'title': 'AI Research Agent', 'overview_first_sentence': '...', 'project_url': '...'})"""
logger.info(f"Storing structured data: {data.json(indent=2)}")
return json.dumps(data.dict())
# Map tool names to their functions for dynamic invocation
tools_map: Dict[str, Callable] = {
"navigate_to_url": navigate_to_url,
"get_current_page_screenshot": get_current_page_screenshot,
"get_current_page_html": get_current_page_html,
"click_element": click_element,
"type_text": type_text,
"extract_text": extract_text,
"store_structured_data": store_structured_data,
}
# Bind tools to the LLM. Now includes the structured output tool.
llm_with_tools = llm.bind_tools(list(tools_map.values()))
def call_llm(state: AgentState) -> AgentState:
"""Invokes the LLM to decide the next action based on the current state."""
logger.info("Calling LLM for next action...")
current_iterations = state.get("iterations", 0) + 1
if current_iterations > state["max_iterations"]:
logger.warning(f"Max iterations ({state['max_iterations']}) reached. Ending task.")
return {**state, "iterations": current_iterations, "tool_output": "Max iterations reached. Cannot complete task.", "error_message": "Max iterations reached."}
messages: List[Union[BaseMessage, Dict[str, Any]]] = []
if current_iterations == 1:
messages.append(HumanMessage(content=state["task"]))
else:
messages.extend(state["chat_history"])
llm_context = []
if state.get("last_screenshot"):
llm_context.append({"type": "image_url", "image_url": {"url": f"file://{state['last_screenshot']}"}})
if state.get("html_preview"):
llm_context.append({"type": "text", "text": f"Current page HTML preview: {state['html_preview']}"})
if state.get("tool_output"):
llm_context.append({"type": "text", "text": f"Previous tool output: {state['tool_output']}"})
if state.get("error_message"):
llm_context.append({"type": "text", "text": f"Previous error: {state['error_message']}. Please try to recover or adjust your approach."})
if llm_context and messages and isinstance(messages[-1], HumanMessage):
messages[-1].content = messages[-1].content if isinstance(messages[-1].content, str) else list(messages[-1].content)
messages[-1].content.extend(llm_context)
elif llm_context:
messages.append(AIMessage(content=llm_context))
try:
if isinstance(messages[-1].content, str):
final_content = messages[-1].content
else:
final_content = []
for item in messages[-1].content:
if isinstance(item, str):
final_content.append({"type": "text", "text": item})
else:
final_content.append(item)
messages[-1].content = final_content
response = llm_with_tools.invoke(messages)
logger.info("LLM invoked successfully, response received.")
return {
**state,
"chat_history": state["chat_history"] + [response],
"iterations": current_iterations,
"error_message": None
}
except Exception as e:
logger.error(f"LLM invocation failed: {e}")
return {**state, "chat_history": state["chat_history"] + [AIMessage(content=f"LLM error: {e}")], "tool_output": f"LLM error: {e}", "error_message": str(e)}
def call_tool(state: AgentState) -> AgentState:
"""Executes the tool suggested by the LLM."""
logger.info("Calling tool based on LLM suggestion...")
messages = state["chat_history"]
last_message = messages[-1]
tool_outputs: List[str] = []
error_occurred = False
current_final_result = state.get("final_result")
if not last_message.tool_calls:
logger.warning("No tool calls found in the last LLM message.")
return {**state, "tool_output": "No tool calls suggested by LLM.", "error_message": "No tool calls suggested by LLM."}
for tool_call in last_message.tool_calls:
tool_name = tool_call['name']
tool_args = tool_call['args']
logger.info(f"Attempting to execute tool: {tool_name} with args: {tool_args}")
try:
if tool_name == "store_structured_data":
# Handle Pydantic validation for store_structured_data
try:
validated_data = ProjectInfo(**tool_args)
output = store_structured_data.invoke(validated_data)
current_final_result = validated_data.dict() # Store the validated dictionary
tool_outputs.append(output)
logger.info(f"Structured data stored successfully: {output}")
except Exception as e:
error_message = f"Failed to validate or store structured data: {e}. Raw args: {tool_args}"
logger.error(error_message)
tool_outputs.append(error_message)
error_occurred = True
state["error_message"] = error_message
continue # Continue to next tool call if any
elif tool_name in tools_map:
output = tools_map[tool_name].invoke(tool_args)
tool_outputs.append(str(output))
logger.info(f"Tool '{tool_name}' executed. Output: {output}")
if isinstance(output, dict):
if "url" in output and output.get("status") == "success":
state["current_url"] = output["url"]
if "screenshot" in output:
state["last_screenshot"] = output["screenshot"]
if "html_preview" in output:
state["html_preview"] = output["html_preview"]
if output.get("status") == "error":
error_occurred = True
state["error_message"] = output.get("message", f"Tool {tool_name} failed.")
logger.error(f"Tool '{tool_name}' reported error: {state['error_message']}")
else:
output = f"Unknown tool: {tool_name}"
logger.warning(output)
tool_outputs.append(output)
error_occurred = True
state["error_message"] = output
except Exception as e:
logger.error(f"Error executing tool '{tool_name}': {e}")
tool_outputs.append(f"Error executing tool '{tool_name}': {e}
")
error_occurred = True
state["error_message"] = str(e)
tool_message = ToolMessage(content="
".join(tool_outputs), tool_call_id=last_message.tool_calls[0]['id'] if last_message.tool_calls else "unknown")
return {
**state,
"chat_history": state["chat_history"] + [tool_message],
"tool_output": "
".join(tool_outputs),
"error_message": state["error_message"] if error_occurred else None,
"final_result": current_final_result
}
def should_continue(state: AgentState) -> str:
"""Determines whether the agent should continue or end the task."""
if state.get("final_result"):
logger.info("Final structured result found. Ending graph.")
return "end"
if state["iterations"] >= state["max_iterations"]:
logger.info("Max iterations reached. Ending graph.")
return "end"
if state.get("error_message") and state["iterations"] > 1:
logger.warning(f"An error occurred: {state['error_message']}. Attempting LLM to recover.")
return "continue"
messages = state["chat_history"]
last_message = messages[-1]
if isinstance(last_message, ToolMessage):
logger.info("Last message is ToolMessage. Continuing to LLM for processing.")
return "continue"
if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
logger.info("LLM suggested tool call. Continuing to tool execution.")
return "continue"
logger.info("LLM did not suggest a tool and is not processing a tool output. Ending graph.")
return "end"
# Build the LangGraph graph
workflow = StateGraph(AgentState)
workflow.add_node("llm_node", call_llm)
workflow.add_node("tool_node", call_tool)
workflow.set_entry_point("llm_node")
workflow.add_conditional_edges(
"llm_node",
should_continue,
{
"continue": "tool_node",
"end": END
}
)
workflow.add_conditional_edges(
"tool_node",
should_continue,
{
"continue": "llm_node",
"end": END
}
)
# Compile the graph
app = workflow.compile()
# Example usage (for testing this phase)
if __name__ == "__main__":
os.environ["OPENAI_API_KEY"] = os.environ.get("OPENAI_API_KEY", "YOUR_OPENAI_API_KEY")
os.environ["PLAYWRIGHT_TIMEOUT_MS"] = "60000"
os.environ["PLAYWRIGHT_ACTION_TIMEOUT_MS"] = "15000"
if os.environ["OPENAI_API_KEY"] == "YOUR_OPENAI_API_KEY":
logger.error("Please set your OPENAI_API_KEY environment variable.")
exit(1)
initial_task = "Navigate to mortalapps.com/agents/projects/ and find the 'AI Research Agent' project. Extract its title, the first sentence of its overview, and its project URL as structured JSON data. Then call the 'store_structured_data' tool with this information."
initial_state: AgentState = {
"task": initial_task,
"chat_history": [HumanMessage(content=initial_task)],
"current_url": "",
"last_screenshot": None,
"html_preview": None,
"tool_output": None,
"iterations": 0,
"max_iterations": 15,
"error_message": None,
"final_result": None
}
try:
print(f"
--- Starting agent for structured extraction task: {initial_task} ---")
for s in app.stream(initial_state):
print(s)
print("------")
final_state = app.invoke(initial_state)
print("
Final State:")
print(f"Task: {final_state['task']}")
print(f"Final URL: {final_state['current_url']}")
print(f"Final Chat History: {[msg.content for msg in final_state['chat_history']]}")
print(f"Final Tool Output: {final_state['tool_output']}")
print(f"Final Error Message: {final_state['error_message']}")
print(f"Final Structured Result: {json.dumps(final_state['final_result'], indent=2) if final_state['final_result'] else 'None'}")
except Exception as e:
logger.critical(f"An error occurred during agent execution: {e}")
finally:
browser_tools_instance._close_browser()
print("
Browser session closed.")
This phase significantly enhances the agent's ability to produce reliable, machine-readable output by integrating Pydantic for structured data extraction. We define a
ProjectInfo Pydantic model, specifying the exact schema for the data we want to extract (title, overview, URL). This model is then exposed to the LLM via a new tool, store_structured_data.
The store_structured_data tool is unique because it expects its data argument to conform to the ProjectInfo schema. When the LLM calls this tool, LangChain's tool-calling mechanism attempts to parse the LLM's generated arguments into a ProjectInfo object. This provides strong validation; if the LLM's output doesn't match the schema, a Pydantic validation error occurs, which is then captured and fed back to the LLM as an error_message. This allows the LLM to self-correct its output format.
The call_tool node is updated to specifically handle store_structured_data, performing the Pydantic validation and storing the validated data in the final_result field of the AgentState. The should_continue function is also updated to check for final_result. If this field is populated, it signals that the agent has successfully extracted the required structured data and can terminate, making the agent goal-oriented towards structured output.
Expected outcome
The agent will execute the task, navigating and extracting information. Once it has gathered all the required data (project title, overview sentence, and URL), it will call the `store_structured_data` tool, providing arguments that conform to the `ProjectInfo` Pydantic model. The `final_result` field in the `AgentState` will then contain a dictionary representing the extracted project information, and the agent will terminate successfully.
How to test
Update app/agent.py and app/tools.py (if you haven't already). Run python app/agent.py. Verify that the agent successfully extracts the project details and that the Final Structured Result in the console output contains valid JSON matching the ProjectInfo schema. If the LLM initially struggles with the schema, observe how it uses the error_message to self-correct its output in subsequent iterations.
Phase 6: API Endpoint and Deployment Readiness
This phase focuses on creating a FastAPI application to expose our agent via a REST API, making it callable by other services. We'll also prepare the project for containerization using a Dockerfile, ensuring all dependencies are managed for consistent deployment.
import os
import logging
from typing import List, Tuple, Annotated, TypedDict, Union, Any, Optional, Callable, Dict
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from tenacity import retry, stop_after_attempt, wait_exponential, before_log, RetryError
from pydantic import BaseModel, Field
import json
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# --- app/tools.py (updated from Phase 3) ---
class BrowserAgentTools:
def __init__(self):
self.browser: Optional[Browser] = None
self.page: Optional[Page] = None
self.playwright_context = None
def _initialize_browser(self) -> None:
if self.playwright_context is None:
self.playwright_context = sync_playwright().start()
if self.browser is None or not self.browser.is_connected():
try:
# Set browser launch options for robustness in containers
self.browser = self.playwright_context.chromium.launch(
headless=True,
args=[
'--no-sandbox',
'--disable-setuid-sandbox',
'--disable-dev-shm-usage',
'--disable-accelerated-2d-canvas',
'--no-first-run',
'--no-zygote',
'--single-process',
'--disable-gpu'
]
)
self.page = self.browser.new_page()
logger.info("Playwright browser initialized successfully.")
except Exception as e:
logger.error(f"Failed to initialize Playwright browser: {e}")
raise
def _close_browser(self) -> None:
if self.browser:
try:
self.browser.close()
logger.info("Playwright browser closed.")
except Exception as e:
logger.warning(f"Error closing browser: {e}")
finally:
self.browser = None
self.page = None
if self.playwright_context:
try:
self.playwright_context.stop()
logger.info("Playwright context stopped.")
except Exception as e:
logger.warning(f"Error stopping playwright context: {e}")
finally:
self.playwright_context = None
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10), before=before_log(logger, logging.INFO))
def navigate(self, url: str) -> Dict[str, Any]:
self._initialize_browser()
if not self.page:
logger.error("Browser page not available for navigation.")
return {"status": "error", "message": "Browser page not initialized."}
try:
logger.info(f"Navigating to URL: {url}")
self.page.goto(url, timeout=int(os.environ.get("PLAYWRIGHT_TIMEOUT_MS", 30000)))
current_url = self.page.url
screenshot_path = self.screenshot("navigate_success")
html_content = self.get_html()
logger.info(f"Successfully navigated to {current_url}")
return {"status": "success", "url": current_url, "screenshot": screenshot_path, "html_preview": html_content[:500]}
except Exception as e:
logger.error(f"Failed to navigate to {url}: {e}")
screenshot_path = self.screenshot("navigate_failure")
raise
def screenshot(self, name: str = "current_page") -> Optional[str]:
if not self.page:
logger.warning("Cannot take screenshot: No active page.")
return None
try:
output_dir = "screenshots"
os.makedirs(output_dir, exist_ok=True)
file_path = os.path.join(output_dir, f"browser_agent_{name}.png")
self.page.screenshot(path=file_path)
logger.info(f"Screenshot saved to {file_path}")
return file_path
except Exception as e:
logger.error(f"Failed to take screenshot: {e}")
return None
def get_html(self) -> str:
if not self.page:
logger.warning("Cannot get HTML: No active page.")
return ""
try:
html = self.page.content()
logger.debug("Retrieved HTML content.")
return html
except Exception as e:
logger.error(f"Failed to get HTML content: {e}")
return ""
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=5), before=before_log(logger, logging.INFO))
def click_element(self, selector: str) -> Dict[str, Any]:
self._initialize_browser()
if not self.page:
return {"status": "error", "message": "Browser page not initialized."}
try:
logger.info(f"Attempting to click element with selector: {selector}")
self.page.click(selector, timeout=int(os.environ.get("PLAYWRIGHT_ACTION_TIMEOUT_MS", 10000)))
screenshot_path = self.screenshot(f"click_success_{selector.replace(' ', '_')[:20]}")
html_content = self.get_html()
logger.info(f"Successfully clicked element: {selector}")
return {"status": "success", "selector": selector, "screenshot": screenshot_path, "html_preview": html_content[:500]}
except Exception as e:
logger.error(f"Failed to click element '{selector}': {e}")
screenshot_path = self.screenshot(f"click_failure_{selector.replace(' ', '_')[:20]}")
raise
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=5), before=before_log(logger, logging.INFO))
def type_text(self, selector: str, text: str) -> Dict[str, Any]:
self._initialize_browser()
if not self.page:
return {"status": "error", "message": "Browser page not initialized."}
try:
logger.info(f"Attempting to type text '{text[:20]}...' into selector: {selector}")
self.page.fill(selector, text, timeout=int(os.environ.get("PLAYWRIGHT_ACTION_TIMEOUT_MS", 10000)))
screenshot_path = self.screenshot(f"type_success_{selector.replace(' ', '_')[:20]}")
html_content = self.get_html()
logger.info(f"Successfully typed into element: {selector}")
return {"status": "success", "selector": selector, "text": text, "screenshot": screenshot_path, "html_preview": html_content[:500]}
except Exception as e:
logger.error(f"Failed to type text into '{selector}': {e}")
screenshot_path = self.screenshot(f"type_failure_{selector.replace(' ', '_')[:20]}")
raise
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=5), before=before_log(logger, logging.INFO))
def extract_text(self, selector: str) -> Dict[str, Any]:
self._initialize_browser()
if not self.page:
return {"status": "error", "message": "Browser page not initialized."}
try:
logger.info(f"Attempting to extract text from selector: {selector}")
element = self.page.locator(selector)
text_content = element.text_content(timeout=int(os.environ.get("PLAYWRIGHT_ACTION_TIMEOUT_MS", 10000)))
if text_content is None:
raise ValueError(f"No text content found for selector: {selector}")
screenshot_path = self.screenshot(f"extract_success_{selector.replace(' ', '_')[:20]}")
html_content = self.get_html()
logger.info(f"Successfully extracted text from {selector}: {text_content[:50]}...")
return {"status": "success", "selector": selector, "extracted_text": text_content, "screenshot": screenshot_path, "html_preview": html_content[:500]}
except Exception as e:
logger.error(f"Failed to extract text from '{selector}': {e}")
screenshot_path = self.screenshot(f"extract_failure_{selector.replace(' ', '_')[:20]}")
raise
# --- app/agent.py (updated from Phase 5, excluding the example usage block) ---
# Define Pydantic model for structured output
class ProjectInfo(BaseModel):
title: str = Field(description="The title of the project.")
overview_first_sentence: str = Field(description="The first sentence of the project's overview.")
project_url: str = Field(description="The URL of the project page.")
@tool
def store_structured_data(data: ProjectInfo) -> str:
logger.info(f"Storing structured data: {data.json(indent=2)}")
return json.dumps(data.dict())
# Map tool names to their functions for dynamic invocation
tools_map: Dict[str, Callable] = {
"navigate_to_url": navigate_to_url,
"get_current_page_screenshot": get_current_page_screenshot,
"get_current_page_html": get_current_page_html,
"click_element": click_element,
"type_text": type_text,
"extract_text": extract_text,
"store_structured_data": store_structured_data,
}
# Bind tools to the LLM. Now includes the structured output tool.
llm_with_tools = llm.bind_tools(list(tools_map.values()))
# AgentState definition (same as Phase 5)
class AgentState(TypedDict):
task: str
chat_history: Annotated[List[BaseMessage], "operator.add"]
current_url: str
last_screenshot: Optional[str]
html_preview: Optional[str]
tool_output: Optional[str]
iterations: int
max_iterations: int
error_message: Optional[str]
final_result: Optional[Dict[str, Any]]
# Node definitions (same as Phase 5)
def call_llm(state: AgentState) -> AgentState:
logger.info("Calling LLM for next action...")
current_iterations = state.get("iterations", 0) + 1
if current_iterations > state["max_iterations"]:
logger.warning(f"Max iterations ({state['max_iterations']}) reached. Ending task.")
return {**state, "iterations": current_iterations, "tool_output": "Max iterations reached. Cannot complete task.", "error_message": "Max iterations reached."}
messages: List[Union[BaseMessage, Dict[str, Any]]] = []
if current_iterations == 1:
messages.append(HumanMessage(content=state["task"]))
else:
messages.extend(state["chat_history"])
llm_context = []
if state.get("last_screenshot"):
llm_context.append({"type": "image_url", "image_url": {"url": f"file://{state['last_screenshot']}"}})
if state.get("html_preview"):
llm_context.append({"type": "text", "text": f"Current page HTML preview: {state['html_preview']}"})
if state.get("tool_output"):
llm_context.append({"type": "text", "text": f"Previous tool output: {state['tool_output']}"})
if state.get("error_message"):
llm_context.append({"type": "text", "text": f"Previous error: {state['error_message']}. Please try to recover or adjust your approach."})
if llm_context and messages and isinstance(messages[-1], HumanMessage):
messages[-1].content = messages[-1].content if isinstance(messages[-1].content, str) else list(messages[-1].content)
messages[-1].content.extend(llm_context)
elif llm_context:
messages.append(AIMessage(content=llm_context))
try:
if isinstance(messages[-1].content, str):
final_content = messages[-1].content
else:
final_content = []
for item in messages[-1].content:
if isinstance(item, str):
final_content.append({"type": "text", "text": item})
else:
final_content.append(item)
messages[-1].content = final_content
response = llm_with_tools.invoke(messages)
logger.info("LLM invoked successfully, response received.")
return {
**state,
"chat_history": state["chat_history"] + [response],
"iterations": current_iterations,
"error_message": None
}
except Exception as e:
logger.error(f"LLM invocation failed: {e}")
return {**state, "chat_history": state["chat_history"] + [AIMessage(content=f"LLM error: {e}")], "tool_output": f"LLM error: {e}", "error_message": str(e)}
def call_tool(state: AgentState) -> AgentState:
logger.info("Calling tool based on LLM suggestion...")
messages = state["chat_history"]
last_message = messages[-1]
tool_outputs: List[str] = []
error_occurred = False
current_final_result = state.get("final_result")
if not last_message.tool_calls:
logger.warning("No tool calls found in the last LLM message.")
return {**state, "tool_output": "No tool calls suggested by LLM.", "error_message": "No tool calls suggested by LLM."}
for tool_call in last_message.tool_calls:
tool_name = tool_call['name']
tool_args = tool_call['args']
logger.info(f"Attempting to execute tool: {tool_name} with args: {tool_args}")
try:
if tool_name == "store_structured_data":
try:
validated_data = ProjectInfo(**tool_args)
output = store_structured_data.invoke(validated_data)
current_final_result = validated_data.dict()
tool_outputs.append(output)
logger.info(f"Structured data stored successfully: {output}")
except Exception as e:
error_message = f"Failed to validate or store structured data: {e}. Raw args: {tool_args}"
logger.error(error_message)
tool_outputs.append(error_message)
error_occurred = True
state["error_message"] = error_message
continue
elif tool_name in tools_map:
output = tools_map[tool_name].invoke(tool_args)
tool_outputs.append(str(output))
logger.info(f"Tool '{tool_name}' executed. Output: {output}")
if isinstance(output, dict):
if "url" in output and output.get("status") == "success":
state["current_url"] = output["url"]
if "screenshot" in output:
state["last_screenshot"] = output["screenshot"]
if "html_preview" in output:
state["html_preview"] = output["html_preview"]
if output.get("status") == "error":
error_occurred = True
state["error_message"] = output.get("message", f"Tool {tool_name} failed.")
logger.error(f"Tool '{tool_name}' reported error: {state['error_message']}")
else:
output = f"Unknown tool: {tool_name}"
logger.warning(output)
tool_outputs.append(output)
error_occurred = True
state["error_message"] = output
except Exception as e:
logger.error(f"Error executing tool '{tool_name}': {e}")
tool_outputs.append(f"Error executing tool '{tool_name}': {e}
")
error_occurred = True
state["error_message"] = str(e)
tool_message = ToolMessage(content="
".join(tool_outputs), tool_call_id=last_message.tool_calls[0]['id'] if last_message.tool_calls else "unknown")
return {
**state,
"chat_history": state["chat_history"] + [tool_message],
"tool_output": "
".join(tool_outputs),
"error_message": state["error_message"] if error_occurred else None,
"final_result": current_final_result
}
def should_continue(state: AgentState) -> str:
if state.get("final_result"):
logger.info("Final structured result found. Ending graph.")
return "end"
if state["iterations"] >= state["max_iterations"]:
logger.info("Max iterations reached. Ending graph.")
return "end"
if state.get("error_message") and state["iterations"] > 1:
logger.warning(f"An error occurred: {state['error_message']}. Attempting LLM to recover.")
return "continue"
messages = state["chat_history"]
last_message = messages[-1]
if isinstance(last_message, ToolMessage):
logger.info("Last message is ToolMessage. Continuing to LLM for processing.")
return "continue"
if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
logger.info("LLM suggested tool call. Continuing to tool execution.")
return "continue"
logger.info("LLM did not suggest a tool and is not processing a tool output. Ending graph.")
return "end"
# Build the LangGraph graph
workflow = StateGraph(AgentState)
workflow.add_node("llm_node", call_llm)
workflow.add_node("tool_node", call_tool)
workflow.set_entry_point("llm_node")
workflow.add_conditional_edges(
"llm_node",
should_continue,
{
"continue": "tool_node",
"end": END
}
)
workflow.add_conditional_edges(
"tool_node",
should_continue,
{
"continue": "llm_node",
"end": END
}
)
# Compile the graph
agent_app = workflow.compile()
# --- app/main.py ---
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
# FastAPI app instance
app = FastAPI(title="Browser Agent API")
class AgentInput(BaseModel):
task: str
max_iterations: int = Field(default=15, ge=1, le=50)
class AgentOutput(BaseModel):
status: str
final_result: Optional[Dict[str, Any]]
logs: List[str]
final_url: Optional[str]
error_message: Optional[str]
@app.post("/run_agent", response_model=AgentOutput)
async def run_browser_agent(agent_input: AgentInput):
logger.info(f"Received agent request for task: {agent_input.task}")
initial_state: AgentState = {
"task": agent_input.task,
"chat_history": [HumanMessage(content=agent_input.task)],
"current_url": "",
"last_screenshot": None,
"html_preview": None,
"tool_output": None,
"iterations": 0,
"max_iterations": agent_input.max_iterations,
"error_message": None,
"final_result": None
}
# Capture logs during agent execution
log_stream = []
handler = logging.StreamHandler()
handler.setLevel(logging.INFO)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
try:
# LangGraph stream doesn't directly return logs, so we'll collect them.
# For production, integrate with OpenTelemetry for better observability.
final_state = None
for s in agent_app.stream(initial_state):
# The stream yields state updates; we'll log them and capture for output
log_stream.append(str(s))
final_state = s # Keep track of the last state
if not final_state:
raise RuntimeError("Agent did not produce any state updates.")
# Extract the relevant info from the final state for the API response
status = "success" if final_state.get("final_result") else "failed"
if final_state.get("error_message"): # Override status if there was an error
status = "failed"
return AgentOutput(
status=status,
final_result=final_state.get("final_result"),
logs=log_stream,
final_url=final_state.get("current_url"),
error_message=final_state.get("error_message")
)
except Exception as e:
logger.critical(f"Agent execution failed for task '{agent_input.task}': {e}")
raise HTTPException(status_code=500, detail=f"Agent execution failed: {e}")
finally:
# Ensure browser is closed regardless of success or failure
browser_tools_instance._close_browser()
logger.removeHandler(handler) # Clean up handler
logger.info(f"Agent task '{agent_input.task}' completed or failed. Browser closed.")
# --- requirements.txt ---
# langchain-openai
# langgraph
# playwright
# tenacity
# pydantic
# fastapi
# uvicorn
# --- Dockerfile ---
# ARG PYTHON_VERSION=3.10.13
# FROM mcr.microsoft.com/playwright/python:v1.44.0-jammy
# WORKDIR /app
# ENV PYTHONUNBUFFERED 1
# ENV PLAYWRIGHT_BROWSERS_PATH=/ms-playwright/chromium-1158
# COPY requirements.txt .
# RUN pip install --no-cache-dir -r requirements.txt
# COPY app/ ./app
# COPY screenshots/ ./screenshots # Create screenshots directory if not exists
# CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
This phase wraps our browser agent in a FastAPI application and provides the necessary
Dockerfile for deployment. The app/main.py file defines a FastAPI instance and an async endpoint /run_agent that accepts a task and max_iterations. This endpoint initializes the AgentState and invokes the LangGraph agent. It's designed to stream the agent's progress (though we capture full logs for the final response here) and return a structured AgentOutput containing the final result, logs, and status.
Crucially, the BrowserAgentTools initialization in app/tools.py is updated with args for playwright.chromium.launch. These arguments (--no-sandbox, --disable-setuid-sandbox, etc.) are essential for running Playwright reliably within a Docker container, especially in environments like Kubernetes where root privileges are restricted. The Dockerfile uses a Playwright-specific base image (mcr.microsoft.com/playwright/python) which pre-installs browser binaries, simplifying the setup. It copies the requirements.txt and the app/ directory, then uses uvicorn to serve the FastAPI application.
For production, the log_stream collection here would ideally be replaced by a more robust observability solution like OpenTelemetry, which can capture traces and metrics across the entire agent run. The finally block ensures _close_browser() is always called, releasing Playwright resources even if an error occurs, and cleans up the logging handler to prevent resource leaks in repeated calls.
Expected outcome
The project will be structured with a `main.py` exposing the agent via FastAPI. A `Dockerfile` will be ready to build a container image. You will be able to build the Docker image, run it, and send HTTP POST requests to `/run_agent` with a task, receiving a structured JSON response containing the agent's outcome, logs, and any extracted data. The Playwright browser will run headless inside the container.
How to test
- Save the updated
app/tools.pyandapp/agent.py(excluding theif __name__ == "__main__"block) into their respective files. Createapp/main.pywith the provided code. Createrequirements.txtandDockerfilein the root. - Build the Docker image:
docker build -t browser-agent . - Run the Docker container:
docker run -p 8000:8000 -e OPENAI_API_KEY="YOUR_OPENAI_API_KEY" browser-agent(replace with your key). - Once running, use
curlor Postman to send a request:curl -X POST -H "Content-Type: application/json" -d '{"task": "Navigate to example.com and extract the main heading text."}' http://localhost:8000/run_agent. Verify you get a JSON response with the extracted data and logs. Check container logs for agent activity.
Testing & Evaluation
Testing and evaluating an AI browser agent is critical due to the non-deterministic nature of LLMs and the dynamic, often unpredictable, environment of the web. A multi-faceted approach ensures robustness and reliability.
Functional Testing
Functional tests cover the agent's ability to perform specific, predefined tasks. These are typically end-to-end scenarios:
- Navigation and Form Filling: Test login flows, registration forms, or multi-page checkout processes on various websites (e.g., test a login on a known demo site, fill out a contact form). Assert that final states (e.g., 'logged in' message, successful submission confirmation) are reached.
- Data Extraction: Provide tasks like "Extract all product names and prices from this e-commerce category page." Compare the agent's structured output against a golden dataset for accuracy and completeness.
- Complex Workflows: Test scenarios involving conditional logic, such as "If product is out of stock, find an alternative; otherwise, add to cart." Mock or control the environment where possible to reliably trigger branches.
LLM-as-a-Judge Evaluation
For tasks where a precise, deterministic output is hard to define, or for evaluating the agent's reasoning, LLM-as-a-Judge can be invaluable. This involves using another, often more powerful, LLM to evaluate the performance of our browser agent.
Example Automated Evaluation Code (using langchain_openai and pydantic)
import os
import json
from typing import Dict, Any, List
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import HumanMessage
# Ensure OPENAI_API_KEY is set
if not os.environ.get("OPENAI_API_KEY"):
print("Please set OPENAI_API_KEY environment variable.")
exit(1)
llm_judge = ChatOpenAI(model="gpt-4o", temperature=0)
class EvaluationResult(BaseModel):
score: int = Field(description="Score from 0 (completely failed) to 5 (perfectly successful).")
reasoning: str = Field(description="Detailed explanation for the given score.")
issues: List[str] = Field(description="List of specific issues or errors found.")
extracted_data_correctness: Optional[Dict[str, Any]] = Field(default=None, description="Detailed correctness of extracted fields if applicable.")
def evaluate_agent_run(task: str, expected_output: Dict[str, Any], agent_output: Dict[str, Any]) -> EvaluationResult:
"""Evaluates an agent's run using an LLM-as-a-Judge.
Args:
task: The original task given to the agent.
expected_output: The ground truth structured output.
agent_output: The actual structured output from the agent.
"""
prompt_template = ChatPromptTemplate.from_messages([
("system", "You are an expert AI agent evaluator. Your task is to assess the performance of a browser agent based on its output and compare it against the expected outcome. Provide a score from 0-5, detailed reasoning, a list of issues, and specific correctness for each extracted data field. Focus on accuracy, completeness, and adherence to the task."),
("human", """
Agent Task: {task}
Expected Structured Output:
{expected_output}
Agent's Actual Structured Output:
{agent_output}
Evaluate the agent's performance. Consider if the agent successfully completed the task, if the extracted data is correct and complete, and if there were any errors or deviations. Provide your evaluation in the following JSON format:
{{ "score": int, "reasoning": str, "issues": list[str], "extracted_data_correctness": dict[str, bool | str] }}
""")
]).partial(expected_output=json.dumps(expected_output, indent=2), agent_output=json.dumps(agent_output, indent=2))
chain = prompt_template | llm_judge.with_structured_output(EvaluationResult)
try:
evaluation = chain.invoke({"task": task})
return evaluation
except Exception as e:
print(f"Error during LLM evaluation: {e}")
return EvaluationResult(score=0, reasoning=f"Evaluation failed: {e}", issues=["LLM evaluation error"], extracted_data_correctness={})
if __name__ == "__main__":
# Example usage (replace with actual agent run data)
sample_task = "Extract the title and first sentence of the overview for 'AI Research Agent' from mortalapps.com/agents/projects/."
sample_expected = {
"title": "Build a Production AI Research Agent with LangGraph",
"overview_first_sentence": "This guide provides a comprehensive walkthrough for building a production-ready AI research agent capable of autonomously searching, summarizing, and synthesizing information from the web.",
"project_url": "/agents/projects/ai-research-agent/"
}
sample_agent_output = {
"title": "Build a Production AI Research Agent with LangGraph",
"overview_first_sentence": "This guide provides a comprehensive walkthrough for building a production-ready AI research agent capable of autonomously searching, summarizing, and synthesizing information from the web.",
"project_url": "/agents/projects/ai-research-agent/" # Corrected URL
}
print("
--- Running LLM-as-a-Judge Evaluation ---")
eval_result = evaluate_agent_run(sample_task, sample_expected, sample_agent_output)
print(f"Score: {eval_result.score}")
print(f"Reasoning: {eval_result.reasoning}")
print(f"Issues: {eval_result.issues}")
print(f"Extracted Data Correctness: {eval_result.extracted_data_correctness}")Failure Scenario Testing
Anticipate and test how the agent handles adverse conditions:
- Missing Elements: A selector points to an element that's no longer on the page.
- Slow Loading: Pages that take a long time to load, or elements that appear with delay.
- CAPTCHAs/Bot Detection: How the agent recognizes and potentially escalates these.
- Dynamic Content: Pages that heavily rely on JavaScript to render content after initial load.
- Network Errors: Simulate connection drops or timeouts.
- Unexpected Pop-ups/Modals: Pages that present unexpected UI elements.
Edge Case Coverage
Test less common but important scenarios:
- Empty Pages: What if the target page has no content or the expected data is missing?
- Pagination/Infinite Scroll: Can the agent handle navigating through multiple pages of results or dynamically loading content?
- Different Layouts: If the target website has responsive design, test on pages with significantly different mobile/desktop layouts.
- Redirects: Ensure the agent follows redirects correctly and updates its
current_url.
Performance Validation
- Latency: Measure the average and P99 latency for common tasks. Browser automation can be slow, so identify bottlenecks.
- Resource Usage: Monitor CPU, memory, and network usage, especially for long-running tasks or multiple concurrent agent runs. Playwright instances are resource-intensive.
Regression Testing
As the agent evolves, maintain a suite of core functional tests. Run these tests automatically after every code change to ensure new features or bug fixes don't inadvertently break existing functionality. This is crucial for maintaining a stable and reliable browser agent over time.
Deployment
Deploying a browser agent involves managing several moving parts: the Python application, its dependencies (including Playwright and browser binaries), and ensuring robust operation in a production environment. We will focus on containerization with Docker and orchestration with Kubernetes.
Dockerfile
Our Dockerfile (located in the project root) is designed to create a self-contained image including Python, Playwright, and its browser dependencies. Using a Playwright-provided base image simplifies this significantly.
# Use a Playwright-specific Python image that includes browser binaries
ARG PYTHON_VERSION=3.10.13 # Specify a Python version for consistency
FROM mcr.microsoft.com/playwright/python:v1.44.0-jammy
# Set working directory inside the container
WORKDIR /app
# Ensure Python output is unbuffered, useful for logging in containers
ENV PYTHONUNBUFFERED 1
# Set the path to Playwright's browser binaries (already installed in base image)
ENV PLAYWRIGHT_BROWSERS_PATH=/ms-playwright/chromium-1158
# Install system dependencies if any are needed for specific tools (e.g., image manipulation)
# RUN apt-get update && apt-get install -y --no-install-recommends \
# libjpeg-dev libpng-dev && \
# rm -rf /var/lib/apt/lists/*
# Copy requirements file and install Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Copy the application code
COPY app/ ./app
# Create a directory for screenshots. Playwright might create it, but explicit is better.
RUN mkdir -p /app/screenshots
# Expose the port FastAPI listens on
EXPOSE 8000
# Command to run the FastAPI application with Uvicorn
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--log-level", "info"]Build the Docker image: docker build -t browser-agent:latest .
Kubernetes Manifest (example)
For production, Kubernetes provides orchestration, scaling, and self-healing capabilities. Here's a basic Deployment and Service manifest (k8s/browser-agent.yaml):
apiVersion: apps/v1
kind: Deployment
metadata:
name: browser-agent-deployment
labels:
app: browser-agent
spec:
replicas: 2 # Start with 2 replicas for high availability and basic scaling
selector:
matchLabels:
app: browser-agent
template:
metadata:
labels:
app: browser-agent
spec:
containers:
- name: browser-agent
image: browser-agent:latest # Use your built Docker image
ports:
- containerPort: 8000
env:
- name: OPENAI_API_KEY
valueFrom:
secretKeyRef:
name: browser-agent-secrets
key: openai_api_key
- name: PLAYWRIGHT_TIMEOUT_MS
value: "60000"
- name: PLAYWRIGHT_ACTION_TIMEOUT_MS
value: "15000"
# Resource limits are critical for Playwright, which can be memory-intensive
resources:
requests:
memory: "1Gi"
cpu: "500m"
limits:
memory: "2Gi"
cpu: "1000m"
readinessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 10
periodSeconds: 5
failureThreshold: 3
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 30
periodSeconds: 10
failureThreshold: 5
volumeMounts:
- name: screenshots-volume
mountPath: /app/screenshots
volumes:
- name: screenshots-volume
emptyDir: {}
---
apiVersion: v1
kind: Service
metadata:
name: browser-agent-service
spec:
selector:
app: browser-agent
ports:
- protocol: TCP
port: 80
targetPort: 8000
type: LoadBalancer # Expose externally via a LoadBalancerApply the Kubernetes manifest: kubectl apply -f k8s/browser-agent.yaml
Secrets Management
Never hardcode API keys or sensitive information. In Kubernetes, use Secrets. Create a secret for your OpenAI API key:
kubectl create secret generic browser-agent-secrets --from-literal=openai_api_key='sk-YOUR_OPENAI_API_KEY'This secret is then referenced in the Deployment manifest using valueFrom: secretKeyRef. For cloud deployments (AWS, Azure, GCP), use their respective secret management services (AWS Secrets Manager, Azure Key Vault, Google Secret Manager).
Health Check and Readiness Probe
The Kubernetes manifest includes readinessProbe and livenessProbe pointing to a /health endpoint. You'll need to add this to your FastAPI app/main.py:
# In app/main.py
@app.get("/health")
async def health_check():
return {"status": "ok"}- Readiness Probe: Determines if a pod is ready to serve traffic. If it fails, Kubernetes stops sending traffic to the pod. This is crucial for graceful startups and shutdowns.
- Liveness Probe: Checks if the container is running. If it fails, Kubernetes restarts the container. This handles cases where the application is hung but the process is still alive.
Scaling Considerations
Browser automation can be resource-intensive. Each Playwright browser instance consumes significant CPU and memory.
- Horizontal Pod Autoscaler (HPA): Configure HPA to scale your
browser-agent-deploymentbased on CPU or memory utilization. This ensures your agent can handle varying loads.
``yaml apiVersion: autoscaling/v2 kind: HorizontalPodAutoscaler metadata: name: browser-agent-hpa spec: scaleTargetRef: apiVersion: apps/v1 kind: Deployment name: browser-agent-deployment minReplicas: 1 maxReplicas: 5 metrics: - type: Resource resource: name: cpu target: type: Utilization averageUtilization: 70 # Target 70% CPU utilization ``
- Resource Limits: Set appropriate
requestsandlimitsin your Kubernetes deployment. This prevents a single agent instance from consuming all node resources and causing instability. - Dedicated Nodes: For very high loads, consider running Playwright-heavy agents on dedicated node pools with more memory and CPU.
Rollback Strategy
Kubernetes deployments inherently support rolling updates and rollbacks. If a new deployment introduces issues, you can revert to a previous stable version:
- Monitor: Closely monitor your deployment after an update using logs and metrics.
- Rollback: If issues are detected, use
kubectl rollout undo deployment/browser-agent-deploymentto revert to the previous revision. Kubernetes will gracefully terminate new pods and bring back old ones. Ensure your application is designed to be stateless or uses external persistent storage (e.g., for LangGraph checkpointers) to make rollbacks seamless.
Production Hardening
Moving a browser agent to production requires meticulous attention to reliability, security, observability, and cost. Here's a checklist for hardening your system:
Security
- Input Validation: Implement strict validation on all API inputs to prevent prompt injection, malformed data, or attempts to trigger unintended browser actions. For example, sanitize URLs before navigation.
- Dependency Scanning: Regularly scan
requirements.txtand the Docker image for known vulnerabilities using tools like Snyk, Trivy, or Grype. Update dependencies promptly. - Sandboxing Playwright (OWASP ASI05): Ensure Playwright runs with
--no-sandboxarguments in containers and ideally within a container runtime that provides strong isolation. Avoid running the browser as root. This prevents malicious web content from escaping the browser and compromising the host system. This is already partially addressed in the Dockerfile, but further OS-level sandboxing (e.g., gVisor) can be considered for extreme security. - Least Privilege: Configure your Kubernetes service accounts and cloud roles with the minimum necessary permissions. The agent should only have access to what it needs to perform its tasks.
Observability
- Structured Logging: Replace simple
print()statements with structured logging (e.g.,json_loggingorstructlog) that emits logs in JSON format. This makes logs easily parsable and searchable in centralized logging systems (ELK Stack, Grafana Loki). - Distributed Tracing (OpenTelemetry): Integrate OpenTelemetry throughout the agent's lifecycle. Instrument LLM calls, tool invocations, and internal graph transitions. This provides end-to-end visibility into complex multi-step processes, allowing you to debug latency and failures across the entire workflow. LangSmith or self-hosted Jaeger/Zipkin are excellent choices.
- Metrics & Alerts: Collect key metrics like LLM token usage, tool execution duration, success/failure rates, and queue depths. Set up alerts for anomalies (e.g., high error rates, increased latency, excessive token usage). Prometheus and Grafana are standard tools for this.
Reliability
- Circuit Breakers: Implement circuit breakers around external API calls (LLM, any other external services) and potentially around Playwright actions. If a dependency is consistently failing, the circuit breaker can temporarily stop calls, preventing cascading failures and allowing the dependency to recover.
- Fallback Mechanisms: If the primary LLM provider fails, have a fallback to a secondary provider or a simpler, cached response. For critical tasks, consider human-in-the-loop (HITL) escalation.
- Robust Retry Policies: The
tenacitylibrary provides excellent retry logic. Ensure timeouts are configured appropriately for Playwright actions and external API calls. Differentiate between transient and permanent errors. - Graceful Shutdown: Ensure the agent can gracefully shut down, closing all browser instances and releasing resources when the pod receives a termination signal from Kubernetes.
Rate Limiting
- API Rate Limiting: Implement rate limiting on your agent's API endpoint to protect against abuse and ensure fair usage, especially if exposed publicly. Tools like Nginx ingress or FastAPI's own rate limiters can be used.
- Upstream Rate Limiting: Be aware of rate limits imposed by external services (e.g., OpenAI API). Implement client-side rate limiting and exponential backoff when calling these services to avoid hitting limits and getting throttled.
Authentication
- API Key/OAuth: For internal services, a simple API key might suffice. For external or public-facing APIs, implement robust authentication and authorization using OAuth 2.0 or JWT. Ensure keys are rotated and managed securely.
- Internal Service Mesh: If deployed within a microservices architecture, leverage a service mesh (e.g., Istio, Linkerd) for mTLS and fine-grained access control between services.
Governance
- Auditing Agent Actions: Log every significant decision made by the LLM and every tool invoked, along with its parameters and outcome. This audit trail is crucial for understanding agent behavior, debugging, and compliance.
- Human-in-the-Loop (HITL): For sensitive or high-impact tasks, integrate a human review step. If the agent encounters uncertainty or a critical decision, it should escalate to a human operator for approval or intervention.
- Prompt Versioning: Store and version prompts in a system like LangSmith or a Git repository. This allows for tracking changes, A/B testing, and rolling back to previous prompt versions if performance degrades.
Cost Optimisation
- Token Usage Monitoring: Continuously monitor LLM token usage per task and per agent run. Identify inefficient prompts or excessive retries that lead to high costs.
- Dynamic Scaling: Use Kubernetes Horizontal Pod Autoscalers (HPA) to scale agent replicas up and down based on demand, minimizing idle resource costs. Consider scaling to zero if no tasks are pending.
- Model Selection: Evaluate if smaller, cheaper LLMs can perform specific sub-tasks effectively. Use larger models only when necessary for complex reasoning.
- Caching: Implement caching for frequently accessed web pages or LLM responses to reduce redundant calls and improve latency.
Learning Objectives
- By the end of this project, you will be able to design and implement a multi-step web automation agent using LangGraph.
- By the end of this project, you will be able to integrate Playwright for robust, dynamic browser control within an AI agent.
- By the end of this project, you will be able to implement visual verification and error recovery strategies using screenshots in web automation.
- By the end of this project, you will be able to define and extract structured data from web pages using Pydantic.
- By the end of this project, you will be able to apply prompt engineering techniques for effective tool use and planning in an agent.
- By the end of this project, you will be able to containerize and deploy a Python-based AI agent with browser dependencies.
Conceptual Prerequisites
What Is Agentic AI?
Understanding the core concepts of AI agents, their components (perception, deliberation, action), and their iterative nature is fundamental to building this browser agent.
ReAct Pattern: Reasoning and Acting in Agents
The ReAct pattern (Reasoning and Acting) is key to how our agent will plan its web interactions, execute them, and learn from the observed outcomes. This project implements a sophisticated ReAct-like loop.
State Machines for AI Agents
LangGraph is built upon state machines. A solid grasp of state management and transitions is crucial for understanding how the agent's workflow is orchestrated.
Agent Failure Recovery and Retry Limits
Web automation is inherently flaky. Understanding strategies for detecting, recovering from, and limiting failures is vital for building a robust browser agent.
Technical Prerequisites
Python 3.9+
Familiarity with modern Python syntax, async/await, and virtual environments is essential for following the code examples and implementing the agent.
Docker
Basic understanding of Docker containers is needed for setting up the development environment and deploying the Playwright-dependent agent.
Web Development Basics
A foundational understanding of HTML, CSS selectors, and how web pages load and interact will greatly assist in debugging and enhancing the agent's web interaction tools.