Agent Series (23): Web Agent — Giving Your Agent Real Eyes on the Internet

User question
     │
     ▼
┌─────────────────────────────────────┐
│         agent_node                  │
│  SystemPrompt + messages → LLM      │
│  bound_llm.invoke(msgs)             │
└────────┬────────────────────────────┘
         │
    Has tool_calls?
         │
    ┌────┴─────┐
   Yes          No (or steps >= MAX_STEPS)
    │                │
    ▼                ▼
tools_node          END
web_search /
fetch_page
    │
    └──→ agent_node (loop)

class WState(TypedDict):
    messages: Annotated[list, add_messages]  # accumulated messages
    steps: int                                # steps consumed

@tool
def web_search(query: str) -> str:
    """
    Search the web with DuckDuckGo.
    Returns up to 5 results, each with title, snippet, and URL.
    Use the URLs from results to call fetch_page — never invent URLs.
    """
    try:
        resp = requests.get(
            "https://html.duckduckgo.com/html/",
            params={"q": query},
            headers=HEADERS,
            timeout=12,
        )
        soup = BeautifulSoup(resp.text, "html.parser")
        results = []
        for i, block in enumerate(soup.select(".result"), 1):
            if i > 5:
                break
            title   = (block.select_one(".result__title")   or soup.new_tag("x")).get_text(strip=True)
            snippet = (block.select_one(".result__snippet") or soup.new_tag("x")).get_text(strip=True)
            url_raw = (block.select_one(".result__url")     or soup.new_tag("x")).get_text(strip=True)
            url = f"https://{url_raw}" if url_raw and not url_raw.startswith("http") else url_raw
            results.append(f"{i}. {title}\n   {snippet}\n   URL: {url}")
        return "\n\n".join(results) if results else "No results found."
    except Exception as exc:
        return f"Search error: {exc}"

@tool
def fetch_page(url: str) -> str:
    """
    Fetch a web page and return its cleaned text (truncated to token budget).
    Only call with real URLs obtained from web_search results.
    """
    try:
        resp = requests.get(url, headers=HEADERS, timeout=12)
        resp.raise_for_status()
        full_text = clean_html(resp.text)
        orig_tokens = count_tokens(full_text)
        displayed = truncate_to_budget(full_text)
        shown_tokens = min(orig_tokens, PAGE_TOKEN_BUDGET)
        return (
            f"[URL: {url}]\n"
            f"[Size: {orig_tokens} tokens → showing {shown_tokens} tokens "
            f"(budget={PAGE_TOKEN_BUDGET})]\n\n"
            f"{displayed}"
        )
    except requests.HTTPError as exc:
        return f"HTTP {exc.response.status_code} — could not fetch {url}"
    except requests.ConnectionError:
        return f"Connection error — {url} may not exist or be unreachable"
    except Exception as exc:
        return f"Error fetching {url}: {type(exc).__name__}: {exc}"

fetch_page(https://totally-made-up-domain-xyz99999.org/docs/n...)
→ Connection error — https://totally-made-up-domain-xyz99999.org/docs/nonexistent may not exist or be unreachable

fetch_page(pypi.org/project/langgraph/)
→ [Size: 4576 tokens → showing 800 tokens (budget=800)]

PAGE_TOKEN_BUDGET = 800   # max tokens of page text sent to LLM per fetch

def count_tokens(text: str) -> int:
    """Rough estimate: ~3 chars per token for English/Chinese mix."""
    return max(1, len(text) // 3)

def truncate_to_budget(text: str, budget: int = PAGE_TOKEN_BUDGET) -> str:
    if count_tokens(text) <= budget:
        return text
    cutoff = budget * 3
    return text[:cutoff] + f"\n\n[... content truncated to ~{budget}-token budget ...]"

MAX_STEPS = 8

def router(state: WState) -> str:
    if state["steps"] >= MAX_STEPS:
        return END
    last = state["messages"][-1]
    if isinstance(last, AIMessage) and last.tool_calls:
        return "tools"
    return END

def agent_node(state: WState) -> dict:
    msgs = [SystemMessage(content=SYSTEM_PROMPT)] + state["messages"]
    response = bound_llm.invoke(msgs)
    return {"messages": [response], "steps": state["steps"] + 1}

state = graph.invoke(
    {"messages": [HumanMessage(content=query)], "steps": 0},
    config={"recursion_limit": MAX_STEPS * 3},
)

======================================================================
Web Agent Demo
Model: glm-4-flash  |  Token budget/page: 800  |  Max steps: 8
======================================================================

=== Part 3: Engineering Guards ===

──────────────────────────────────────────────────────────────────────
[Guard 1] URL error handling (bad / hallucinated URL)
  fetch_page(https://totally-made-up-domain-xyz99999.org/docs/n...)
  → Connection error — https://totally-made-up-domain-xyz99999.org/docs/nonexistent may not exist or be unreachable

──────────────────────────────────────────────────────────────────────
[Guard 2] Token budget enforcement (budget=800 tokens/page)
  fetch_page(pypi.org/project/langgraph/)
  → [Size: 4576 tokens → showing 800 tokens (budget=800)]

──────────────────────────────────────────────────────────────────────
[Guard 3] Step limit (MAX_STEPS=8) — agent cannot loop forever
  Graph router returns END when state['steps'] >= 8
  Even if tool_calls remain, execution stops.

TOOLS   = [web_search, fetch_page]
TOOL_MAP = {t.name: t for t in TOOLS}
bound_llm = llm.bind_tools(TOOLS)

SYSTEM_PROMPT = f"""You are a web research agent. Answer the user's question by browsing the web.

Workflow:
1. Call web_search to find relevant pages.
2. Call fetch_page on promising URLs to read content.
3. If you find the answer, give a clear, concise final response.
4. If a page doesn't help, try a different search query.

Strict rules:
- Only use URLs from web_search results — never invent or guess URLs.
- If fetch_page returns an error, try a different URL or search query.
- You have at most {MAX_STEPS} total steps. Be efficient.
- Once you have enough information, stop browsing and answer directly."""


class WState(TypedDict):
    messages: Annotated[list, add_messages]
    steps: int


def agent_node(state: WState) -> dict:
    msgs = [SystemMessage(content=SYSTEM_PROMPT)] + state["messages"]
    response = bound_llm.invoke(msgs)
    return {"messages": [response], "steps": state["steps"] + 1}


def tools_node(state: WState) -> dict:
    last = state["messages"][-1]
    results = []
    for tc in last.tool_calls:
        output = TOOL_MAP[tc["name"]].invoke(tc["args"])
        results.append(ToolMessage(content=str(output), tool_call_id=tc["id"]))
    return {"messages": results}


def router(state: WState) -> str:
    if state["steps"] >= MAX_STEPS:
        return END
    last = state["messages"][-1]
    if isinstance(last, AIMessage) and last.tool_calls:
        return "tools"
    return END


def build_graph():
    g = StateGraph(WState)
    g.add_node("agent", agent_node)
    g.add_node("tools", tools_node)
    g.set_entry_point("agent")
    g.add_conditional_edges("agent", router, {"tools": "tools", END: END})
    g.add_edge("tools", "agent")
    return g.compile()