Build a Research AI Agent in Python

pip install "10xgraph[anthropic]" httpx
export ANTHROPIC_API_KEY=sk-ant-...
export TAVILY_API_KEY=tvly-...
python graph.py

[ User question ]
       |
       v
[ MAIN: Agent plans 2-3 subqueries ] <------+
       |                                    |
       | tool calls?                        |
       v                                    |
[ TOOL: web_search / fetch_url ] -----------+
       |
       | no tool calls
       v
[ END: cited answer ] --> validate_citations()

import os
import re
import httpx
from html.parser import HTMLParser
from tenxgraph.core.graph import Agent, StateGraph, ToolNode
from tenxgraph.core.state.agent_state import AgentState
from tenxgraph.core.state.message import Message
from tenxgraph.utils.constants import END

# Load the Tavily key from the environment (the Anthropic key is read by 10xGraph)
TAVILY_API_KEY = os.getenv("TAVILY_API_KEY")
if not TAVILY_API_KEY:
    raise ValueError("TAVILY_API_KEY not set in environment")

# Global counter so [source-N] tags stay unique across several searches
_source_counter = 0


def web_search(query: str, max_results: int = 5) -> str:
    """Search the web using Tavily. Returns sources with title, URL, and snippet."""
    global _source_counter
    if not query.strip():
        return "No query provided."
    
    try:
        response = httpx.post(
            "https://api.tavily.com/search",
            json={"api_key": TAVILY_API_KEY, "query": query, "max_results": max_results},
            timeout=10.0,
        )
        response.raise_for_status()
        data = response.json()
    except Exception as e:
        return f"Search error: {e}"
    
    results = data.get("results", [])
    if not results:
        return "No search results found."
    
    output = []
    for result in results:
        _source_counter += 1
        i = _source_counter
        title = result.get("title", "No title")
        url = result.get("url", "No URL")
        content = result.get("content", "No snippet")
        output.append(f"[source-{i}] {title}\n{url}\n{content}")
    
    return "\n\n".join(output)

def strip_html_tags(text: str) -> str:
    """Simple HTML tag stripper for fetched pages."""
    class MLStripper(HTMLParser):
        def __init__(self):
            super().__init__()
            self.reset()
            self.fed = []
        def handle_data(self, data):
            self.fed.append(data)
        def get_data(self):
            return ''.join(self.fed)
    
    stripper = MLStripper()
    try:
        stripper.feed(text)
        return stripper.get_data()
    except Exception:
        return text

def fetch_url(url: str) -> str:
    """Fetch the full text of a URL. Call after web_search to read promising sources."""
    if not url.strip():
        return "No URL provided."
    
    try:
        response = httpx.get(url, timeout=15.0, follow_redirects=True)
        response.raise_for_status()
        text = response.text
        text = strip_html_tags(text)
        text = re.sub(r'\s+', ' ', text).strip()
        return text[:20_000]  # Cap at 20k chars to keep context manageable
    except Exception as e:
        return f"Error fetching {url}: {e}"

tool_node = ToolNode([web_search, fetch_url])

graph = StateGraph()
graph.add_node(
    "MAIN",
    Agent(
        model="anthropic/claude-opus-5",
        system_prompt=[{
            "role": "system",
            "content": (
                "You are a research assistant. Given a question, break it into 2-3 subqueries "
                "to cover different angles. Use web_search to find sources, then fetch_url to read "
                "promising articles in full. Gather at least 3 distinct sources before synthesizing. "
                "\n"
                "When you have enough information, write a clear, concise answer. "
                "EVERY factual claim must include a citation marker like [source-1] from the search results. "
                "Inline the citations: 'According to [source-2], Python is X.' "
                "\n"
                "If you cannot find reliable sources, say so explicitly. Do not invent facts."
            )
        }],
        tool_node="TOOL",
    ),
)
graph.add_node("TOOL", tool_node)

def route_after_agent(state: AgentState) -> str:
    """Route to tools if the agent called them, otherwise end."""
    if not state.context or len(state.context) == 0:
        return END
    
    last_message = state.context[-1]
    
    # If last message is from assistant with tool calls, go to TOOL
    if hasattr(last_message, "tools_calls") and last_message.tools_calls:
        return "TOOL"
    
    # If last message is a tool result, go back to agent for synthesis
    if last_message.role == "tool":
        return "MAIN"
    
    # Otherwise end
    return END

graph.add_conditional_edges(
    "MAIN",
    route_after_agent,
    {"TOOL": "TOOL", END: END},
)
graph.add_edge("TOOL", "MAIN")
graph.set_entry_point("MAIN")

CITATION_RE = re.compile(r"\[source-\d+\]")

def validate_citations(answer: str, max_uncited_sentences: int = 1) -> tuple[bool, str]:
    """Validate that all factual sentences have citations."""
    sentences = re.split(r"(?<=[.!?])\s+", answer)
    uncited = [s for s in sentences if s.strip() and not CITATION_RE.search(s)]
    
    if len(uncited) > max_uncited_sentences:
        return False, f"{len(uncited)} sentences lack citations: {uncited[:2]}"
    return True, "OK"

def compile_and_run():
    """Compile the graph and run a query."""
    compiled_graph = graph.compile()
    
    question = "Compare Python and Rust on performance and ease of learning."
    
    print(f"\n{'='*60}")
    print(f"Question: {question}")
    print(f"{'='*60}\n")
    
    config = {"thread_id": "research-1", "recursion_limit": 10}
    result = compiled_graph.invoke(
        {"messages": [Message.text_message(question)]},
        config=config,
    )
    
    # Extract the final answer (the last assistant message with text)
    final_answer = None
    for msg in reversed(result.get("messages", [])):
        if msg.role == "assistant" and msg.text().strip():
            final_answer = msg.text()
            break

    if final_answer:
                break
    
    if final_answer:
        print("Answer:")
        print(final_answer)
        print()
        
        is_valid, msg = validate_citations(final_answer)
        print(f"Citation validation: {'PASS' if is_valid else 'FAIL'} ({msg})")
    else:
        print("No answer generated.")

if __name__ == "__main__":
    compile_and_run()

============================================================
Question: Compare Python and Rust on performance and ease of learning.
============================================================

Answer:
Python and Rust serve different purposes in software development. [source-1] Python excels in rapid development and has a gentler learning curve due to its readable syntax and extensive library ecosystem. [source-2] Rust prioritizes memory safety and performance, making it ideal for systems programming where efficiency is critical, though it has a steeper learning curve [source-3] due to its ownership model and borrow checker.

Citation validation: PASS (OK)