Build a Real Agent This Weekend: From Zero to a Working Research Assistant

TOOLS = [
    {
        "name": "web_search",
        "description": (
            "Search the web for information on a topic. Returns a list of results, "
            "each with a title, URL, and snippet. Use this to find relevant sources "
            "for the research question. Prefer specific queries over broad ones."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "The search query. Be specific and targeted.",
                }
            },
            "required": ["query"],
        },
    },
    {
        "name": "read_page",
        "description": (
            "Read the content of a web page given its URL. Returns the extracted "
            "text content. Use this after web_search to read promising results in "
            "full. Do not read more than 3 pages per research question."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "url": {
                    "type": "string",
                    "description": "The full URL of the page to read.",
                }
            },
            "required": ["url"],
        },
    },
    {
        "name": "save_note",
        "description": (
            "Save a research note with source attribution. Use this to record key "
            "findings as you research. Each note should contain a single claim or "
            "finding, the source URL, and a brief explanation. Save notes as you go; "
            "do not wait until the end."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "claim": {
                    "type": "string",
                    "description": "The key finding or claim.",
                },
                "source_url": {
                    "type": "string",
                    "description": "The URL where this information was found.",
                },
                "explanation": {
                    "type": "string",
                    "description": "Brief context explaining the claim.",
                },
            },
            "required": ["claim", "source_url", "explanation"],
        },
    },
]

import json
import time

# Research notes accumulate here across iterations
research_notes: list[dict] = []


def web_search(query: str, timeout: int = 10) -> str:
    """Mock web search. Returns realistic results for any query.

    To use a real search API, replace this function body with:
        import httpx
        resp = httpx.get(
            "https://api.tavily.com/search",
            params={"query": query, "api_key": os.environ["TAVILY_API_KEY"]},
            timeout=timeout,
        )
        return resp.text
    """
    results = [
        {
            "title": f"Research findings on: {query}",
            "url": f"https://example.com/research/{query.replace(' ', '-')[:30]}",
            "snippet": (
                f"A comprehensive analysis of {query}. Key findings include "
                "measurable improvements in efficiency and documented trade-offs "
                "in implementation complexity."
            ),
        },
        {
            "title": f"Industry report: {query}",
            "url": f"https://example.com/report/{query.replace(' ', '-')[:30]}",
            "snippet": (
                f"2025 industry data on {query}. Survey of 500 practitioners "
                "reveals adoption rates of 34% in enterprise settings, with "
                "significant variation by sector."
            ),
        },
        {
            "title": f"Case study: {query} in practice",
            "url": f"https://example.com/case/{query.replace(' ', '-')[:30]}",
            "snippet": (
                f"Real-world implementation of {query} at a Fortune 500 company. "
                "Reduced manual effort by 60% over 18 months. Lessons learned "
                "and failure modes documented."
            ),
        },
    ]
    return json.dumps(results)


def read_page(url: str, timeout: int = 15) -> str:
    """Mock page reader. Returns realistic page content.

    To use a real page reader, replace this function body with:
        import httpx
        from bs4 import BeautifulSoup
        resp = httpx.get(url, timeout=timeout, follow_redirects=True)
        soup = BeautifulSoup(resp.text, "html.parser")
        return soup.get_text(separator="\\n", strip=True)[:5000]
    """
    return (
        f"Content from {url}\n\n"
        "Key findings from this source:\n"
        "1. Organizations that implemented structured approaches saw 40-60% "
        "improvement in target metrics over 12-18 months.\n"
        "2. The most common failure mode was insufficient stakeholder alignment, "
        "not technical complexity.\n"
        "3. Teams that measured outcomes from day one iterated 3x faster than "
        "teams that deferred measurement.\n"
        "4. Cost of implementation ranged from $50K-$500K depending on scope, "
        "with median ROI positive within 9 months.\n\n"
        "The study surveyed 200 organizations across financial services, "
        "healthcare, and technology sectors between 2023 and 2025."
    )


def save_note(claim: str, source_url: str, explanation: str) -> str:
    """Save a research note with source attribution."""
    note = {
        "claim": claim,
        "source_url": source_url,
        "explanation": explanation,
        "saved_at": time.strftime("%Y-%m-%dT%H:%M:%SZ"),
    }
    research_notes.append(note)
    return json.dumps({"status": "saved", "total_notes": len(research_notes)})

#!/usr/bin/env python3
"""Research assistant agent. Requires: pip install anthropic"""

import anthropic
import json
import sys
import time

# --- Configuration ---
MODEL = "claude-sonnet-4-20250514"
MAX_TOKENS = 4096
MAX_ITERATIONS = 10
TOKEN_WARNING_THRESHOLD = 80_000  # warn when input tokens approach this
TOOL_TIMEOUT = 15  # seconds

client = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY from env

# --- Tool definitions (TOOLS list from above) ---
TOOLS = [
    {
        "name": "web_search",
        "description": (
            "Search the web for information on a topic. Returns a list of results, "
            "each with a title, URL, and snippet. Use this to find relevant sources "
            "for the research question. Prefer specific queries over broad ones."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "The search query. Be specific and targeted.",
                }
            },
            "required": ["query"],
        },
    },
    {
        "name": "read_page",
        "description": (
            "Read the content of a web page given its URL. Returns the extracted "
            "text content. Use this after web_search to read promising results in "
            "full. Do not read more than 3 pages per research question."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "url": {
                    "type": "string",
                    "description": "The full URL of the page to read.",
                }
            },
            "required": ["url"],
        },
    },
    {
        "name": "save_note",
        "description": (
            "Save a research note with source attribution. Use this to record key "
            "findings as you research. Each note should contain a single claim or "
            "finding, the source URL, and a brief explanation. Save notes as you go; "
            "do not wait until the end."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "claim": {
                    "type": "string",
                    "description": "The key finding or claim.",
                },
                "source_url": {
                    "type": "string",
                    "description": "The URL where this information was found.",
                },
                "explanation": {
                    "type": "string",
                    "description": "Brief context explaining the claim.",
                },
            },
            "required": ["claim", "source_url", "explanation"],
        },
    },
]

# --- Tool implementations ---
research_notes: list[dict] = []


def web_search(query: str, timeout: int = TOOL_TIMEOUT) -> str:
    """Mock web search. Replace body with Tavily API call for production."""
    results = [
        {
            "title": f"Research findings on: {query}",
            "url": f"https://example.com/research/{query.replace(' ', '-')[:30]}",
            "snippet": (
                f"A comprehensive analysis of {query}. Key findings include "
                "measurable improvements in efficiency and documented trade-offs."
            ),
        },
        {
            "title": f"Industry report: {query}",
            "url": f"https://example.com/report/{query.replace(' ', '-')[:30]}",
            "snippet": (
                f"2025 industry data on {query}. Survey of 500 practitioners "
                "reveals adoption rates of 34% in enterprise settings."
            ),
        },
        {
            "title": f"Case study: {query} in practice",
            "url": f"https://example.com/case/{query.replace(' ', '-')[:30]}",
            "snippet": (
                f"Real-world implementation of {query} at a Fortune 500 company. "
                "Reduced manual effort by 60% over 18 months."
            ),
        },
    ]
    return json.dumps(results)


def read_page(url: str, timeout: int = TOOL_TIMEOUT) -> str:
    """Mock page reader. Replace body with httpx + BeautifulSoup for production."""
    return (
        f"Content from {url}\n\n"
        "Key findings:\n"
        "1. Structured approaches yielded 40-60% improvement over 12-18 months.\n"
        "2. Most common failure mode: insufficient stakeholder alignment.\n"
        "3. Teams measuring from day one iterated 3x faster.\n"
        "4. Implementation cost: $50K-$500K, median ROI positive within 9 months.\n\n"
        "Survey of 200 organizations across financial services, healthcare, "
        "and technology sectors (2023-2025)."
    )


def save_note(claim: str, source_url: str, explanation: str) -> str:
    """Save a research note with source attribution."""
    note = {
        "claim": claim,
        "source_url": source_url,
        "explanation": explanation,
        "saved_at": time.strftime("%Y-%m-%dT%H:%M:%SZ"),
    }
    research_notes.append(note)
    return json.dumps({"status": "saved", "total_notes": len(research_notes)})


# Map tool names to functions
TOOL_DISPATCH = {
    "web_search": lambda args: web_search(args["query"]),
    "read_page": lambda args: read_page(args["url"]),
    "save_note": lambda args: save_note(
        args["claim"], args["source_url"], args["explanation"]
    ),
}


def execute_tool(name: str, args: dict) -> str:
    """Execute a tool call with error handling and timeout tracking."""
    if name not in TOOL_DISPATCH:
        return json.dumps({"error": f"Unknown tool: {name}"})
    try:
        start = time.time()
        result = TOOL_DISPATCH[name](args)
        elapsed = time.time() - start
        if elapsed > TOOL_TIMEOUT:
            return json.dumps({
                "error": f"Tool {name} took {elapsed:.1f}s (limit: {TOOL_TIMEOUT}s)",
                "partial_result": result[:500] if result else None,
            })
        return result
    except Exception as e:
        return json.dumps({"error": f"Tool {name} failed: {str(e)}"})


def run_agent(question: str) -> str:
    """Run the research assistant agent on a question."""
    system_prompt = (
        "You are a research assistant. Your job is to answer the user's research "
        "question by searching the web, reading relevant pages, and taking notes "
        "with source attribution.\n\n"
        "Process:\n"
        "1. Search for relevant sources using web_search.\n"
        "2. Read the most promising pages using read_page (max 3 pages).\n"
        "3. Save key findings using save_note as you discover them.\n"
        "4. Once you have enough information (at least 3 notes from 2+ sources), "
        "synthesize a brief with citations.\n\n"
        "Rules:\n"
        "- Every claim in your final brief must cite a source URL.\n"
        "- If search results are insufficient, try a different query.\n"
        "- If a page fails to load, skip it and try the next result.\n"
        "- Be concise. The brief should be 200-400 words."
    )

    messages = [{"role": "user", "content": question}]
    total_input_tokens = 0
    total_output_tokens = 0

    for iteration in range(1, MAX_ITERATIONS + 1):
        print(f"\n--- Iteration {iteration}/{MAX_ITERATIONS} ---")

        # Call the LLM
        try:
            response = client.messages.create(
                model=MODEL,
                max_tokens=MAX_TOKENS,
                system=system_prompt,
                tools=TOOLS,
                messages=messages,
            )
        except anthropic.APIError as e:
            print(f"  API error: {e}")
            break

        # Track token usage
        total_input_tokens += response.usage.input_tokens
        total_output_tokens += response.usage.output_tokens
        print(f"  Tokens: {response.usage.input_tokens} in, "
              f"{response.usage.output_tokens} out "
              f"(cumulative: {total_input_tokens} in, {total_output_tokens} out)")

        # Warn if approaching context limit
        if total_input_tokens > TOKEN_WARNING_THRESHOLD:
            print(f"  WARNING: Input tokens ({total_input_tokens}) approaching "
                  f"limit ({TOKEN_WARNING_THRESHOLD}). Agent should wrap up.")

        # Check stop reason
        if response.stop_reason == "end_turn":
            # Agent is done; extract final text
            final_text = ""
            for block in response.content:
                if hasattr(block, "text"):
                    final_text += block.text
            print(f"\n  Agent finished after {iteration} iterations.")
            print(f"  Total tokens: {total_input_tokens} in, "
                  f"{total_output_tokens} out")
            print(f"  Notes saved: {len(research_notes)}")
            return final_text

        if response.stop_reason != "tool_use":
            # Unexpected stop reason
            print(f"  Unexpected stop_reason: {response.stop_reason}")
            final_text = ""
            for block in response.content:
                if hasattr(block, "text"):
                    final_text += block.text
            return final_text or "Agent stopped unexpectedly."

        # Process tool calls
        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                print(f"  Tool call: {block.name}({json.dumps(block.input)[:80]}...)")
                result = execute_tool(block.name, block.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result,
                })

        # Append assistant message and tool results to conversation
        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": tool_results})

    # If we hit max iterations, ask for a final synthesis
    print(f"\n  Max iterations ({MAX_ITERATIONS}) reached. Requesting synthesis.")
    messages.append({
        "role": "user",
        "content": (
            "You have reached the maximum number of research iterations. "
            "Synthesize your findings into a brief now, using the notes you "
            "have saved. Cite sources for every claim."
        ),
    })
    try:
        response = client.messages.create(
            model=MODEL, max_tokens=MAX_TOKENS,
            system=system_prompt, tools=TOOLS, messages=messages,
        )
        for block in response.content:
            if hasattr(block, "text"):
                return block.text
    except anthropic.APIError as e:
        return f"Final synthesis failed: {e}"

    return "Agent completed without producing a final brief."


# --- Entry point ---
if __name__ == "__main__":
    question = (
        sys.argv[1] if len(sys.argv) > 1
        else "What are the most effective approaches to implementing "
             "data quality monitoring in enterprise data platforms?"
    )
    print(f"Research question: {question}\n")
    brief = run_agent(question)
    print("\n" + "=" * 60)
    print("RESEARCH BRIEF")
    print("=" * 60)
    print(brief)

import re


def eval_research_brief(brief: str, notes: list[dict]) -> dict:
    """Score a research brief on three dimensions.

    Returns a dict with scores and a pass/fail verdict.
    """
    scores = {}

    # 1. Source count: how many unique URLs are cited in the brief?
    urls_in_brief = set(re.findall(r"https?://[^\s\)\"]+", brief))
    scores["sources_cited"] = len(urls_in_brief)
    scores["sources_pass"] = len(urls_in_brief) >= 2

    # 2. Note coverage: what fraction of saved notes appear in the brief?
    notes_referenced = 0
    for note in notes:
        # Check if the claim (or a substantial substring) appears in the brief
        claim_words = note["claim"].lower().split()
        # Match if at least 60% of claim words appear in the brief
        brief_lower = brief.lower()
        matches = sum(1 for w in claim_words if w in brief_lower)
        if matches / max(len(claim_words), 1) >= 0.6:
            notes_referenced += 1
    scores["notes_referenced"] = notes_referenced
    scores["notes_total"] = len(notes)
    scores["coverage_pass"] = (
        notes_referenced / max(len(notes), 1) >= 0.5
    )

    # 3. Length check: is the brief within the target range?
    word_count = len(brief.split())
    scores["word_count"] = word_count
    scores["length_pass"] = 100 <= word_count <= 600

    # Overall verdict
    scores["passed"] = all([
        scores["sources_pass"],
        scores["coverage_pass"],
        scores["length_pass"],
    ])

    return scores


# Run the eval after the agent completes
if __name__ == "__main__":
    # ... (after run_agent returns)
    eval_result = eval_research_brief(brief, research_notes)
    print("\n" + "=" * 60)
    print("EVAL RESULTS")
    print("=" * 60)
    for key, value in eval_result.items():
        print(f"  {key}: {value}")
    print(f"\n  Verdict: {'PASS' if eval_result['passed'] else 'FAIL'}")