If you’re building agents with LlamaIndex and want to use Claude 3.5 Sonnet without managing multiple provider accounts, n4n.ai gives you a single OpenAI-compatible endpoint that routes to 240+ models including Anthropic’s latest. This tutorial walks through a complete, runnable example: a research agent that searches the web, fetches pages, and synthesizes answers with citations. You’ll see tool definitions, streaming responses, and how to handle provider fallbacks when something degrades.
Prerequisites
- Python 3.10+
- An n4n.ai API key (get one at n4n.ai)
- Basic familiarity with LlamaIndex concepts (agents, tools, query engines)
Install the dependencies:
pip install llama-index llama-index-llms-openai llama-index-tools-tavily-research httpxWe use the OpenAI-compatible client because n4n.ai speaks that protocol natively. The llama-index-llms-openai package works out of the box — just point it at the n4n.ai base URL.
Configure the LLM client
Create a small configuration module so you can swap models or endpoints without hunting through code.
# config.py
import os
from llama_index.llms.openai import OpenAI
N4N_API_KEY = os.getenv("N4N_API_KEY")
if not N4N_API_KEY:
raise RuntimeError("Set N4N_API_KEY in your environment")
# n4n.ai OpenAI-compatible endpoint
N4N_BASE_URL = "https://api.n4n.ai/v1"
def get_llm(model: str = "anthropic/claude-3.5-sonnet", **kwargs) -> OpenAI:
"""
Return an OpenAI-compatible client pointed at n4n.ai.
The model string follows the format: provider/model-name
n4n.ai normalizes this across all 240+ models.
"""
return OpenAI(
model=model,
api_key=N4N_API_KEY,
base_url=N4N_BASE_URL,
temperature=0.1,
max_tokens=4096,
**kwargs,
)Set your key and test it:
export N4N_API_KEY="your-key-here"
python -c "from config import get_llm; print(get_llm().complete('Say hello in one sentence').text)"Expected output:
Hello! How can I help you today?Define the research tools
We’ll give the agent two tools: a web search (via Tavily) and a page fetcher for deeper reads. LlamaIndex’s FunctionTool wraps any Python callable.
# tools.py
import os
import httpx
from typing import List, Dict, Any
from llama_index.core.tools import FunctionTool
TAVILY_API_KEY = os.getenv("TAVILY_API_KEY")
if not TAVILY_API_KEY:
raise RuntimeError("Set TAVILY_API_KEY for web search")
async def web_search(query: str, max_results: int = 5) -> List[Dict[str, Any]]:
"""
Search the web via Tavily and return structured results.
"""
url = "https://api.tavily.com/search"
payload = {
"api_key": TAVILY_API_KEY,
"query": query,
"max_results": max_results,
"search_depth": "advanced",
"include_answer": True,
"include_raw_content": False,
}
async with httpx.AsyncClient(timeout=30.0) as client:
resp = await client.post(url, json=payload)
resp.raise_for_status()
data = resp.json()
results = []
for item in data.get("results", []):
results.append({
"title": item.get("title"),
"url": item.get("url"),
"content": item.get("content"),
"score": item.get("score"),
})
return results
async def fetch_page(url: str, max_chars: int = 8000) -> Dict[str, Any]:
"""
Fetch a single page and return cleaned text.
"""
async with httpx.AsyncClient(timeout=30.0, follow_redirects=True) as client:
resp = await client.get(url, headers={"User-Agent": "Mozilla/5.0"})
resp.raise_for_status()
text = resp.text
# Naive cleanup — replace with readability/trafilatura for production
from html import unescape
import re
text = re.sub(r"<script.*?</script>", "", text, flags=re.DOTALL | re.IGNORECASE)
text = re.sub(r"<style.*?</style>", "", text, flags=re.DOTALL | re.IGNORECASE)
text = re.sub(r"<[^>]+>", " ", text)
text = unescape(text)
text = re.sub(r"\s+", " ", text).strip()
return {
"url": url,
"text": text[:max_chars],
"truncated": len(text) > max_chars,
}
search_tool = FunctionTool.from_defaults(
fn=web_search,
name="web_search",
description="Search the web for current information. Returns title, URL, snippet, and relevance score.",
)
fetch_tool = FunctionTool.from_defaults(
fn=fetch_page,
name="fetch_page",
description="Fetch and extract readable text from a specific URL. Use after web_search to get full content.",
)Build the agent
LlamaIndex’s ReActAgent implements the Reasoning + Acting loop. We’ll configure it with our tools, a system prompt that enforces citation discipline, and streaming for responsive UX.
# agent.py
from llama_index.core.agent import ReActAgent
from llama_index.core.memory import ChatMemoryBuffer
from llama_index.core.callbacks import CallbackManager
from llama_index.core.tools import BaseTool
from config import get_llm
from tools import search_tool, fetch_tool
SYSTEM_PROMPT = """\
You are a research agent. Answer questions by searching the web, fetching relevant pages, and synthesizing findings.
Rules:
1. Always cite sources inline using [title](url) format.
2. Prefer primary sources and recent content (last 12 months).
3. If sources conflict, present both perspectives.
4. Never hallucinate — if you don't know, say so.
5. Keep answers concise but complete.
"""
def build_agent(tools: list[BaseTool] | None = None) -> ReActAgent:
if tools is None:
tools = [search_tool, fetch_tool]
llm = get_llm()
memory = ChatMemoryBuffer.from_defaults(token_limit=8000)
agent = ReActAgent.from_tools(
tools=tools,
llm=llm,
memory=memory,
system_prompt=SYSTEM_PROMPT,
verbose=True,
max_iterations=10,
)
return agentRun a streaming query
Streaming lets you show partial output as the agent reasons and calls tools — critical for perceived latency.
# run_agent.py
import asyncio
from agent import build_agent
async def main():
agent = build_agent()
query = "What are the key differences between Claude 3.5 Sonnet and GPT-4o for coding tasks? Cite benchmarks."
print(f"Query: {query}\n")
print("--- Streaming response ---\n")
# astream_chat yields tokens as they arrive
async for chunk in agent.astream_chat(query):
print(chunk.delta, end="", flush=True)
print("\n\n--- Done ---")
if __name__ == "__main__":
asyncio.run(main())Run it:
python run_agent.pyExpected output (truncated for brevity):
Query: What are the key differences between Claude 3.5 Sonnet and GPT-4o for coding tasks? Cite benchmarks.
--- Streaming response ---
Thought: The user wants a comparison of Claude 3.5 Sonnet and GPT-4o for coding, with benchmarks. I need to search for recent benchmarks and analysis.
Action: web_search
Action Input: {"query": "Claude 3.5 Sonnet vs GPT-4o coding benchmarks 2024", "max_results": 5}
Observation: [{"title": "Claude 3.5 Sonnet vs GPT-4o: Coding Benchmark Comparison", "url": "https://example.com/benchmark", "content": "In HumanEval, Claude 3.5 Sonnet scores 92.0% vs GPT-4o 90.2%...", "score": 0.94}, ...]
Thought: Good, I have benchmark data. Let me fetch the primary source for more detail.
Action: fetch_page
Action Input: {"url": "https://example.com/benchmark"}
Observation: {"url": "https://example.com/benchmark", "text": "Full article text with detailed breakdown...", "truncated": false}
Based on recent benchmarks, here are the key differences for coding tasks:
**Benchmark Performance**
- HumanEval: Claude 3.5 Sonnet 92.0% vs GPT-4o 90.2% [Claude 3.5 Sonnet vs GPT-4o: Coding Benchmark Comparison](https://example.com/benchmark)
- MBPP: Claude 3.5 Sonnet 88.5% vs GPT-4o 87.1% [same source]
- LiveCodeBench: GPT-4o edges out slightly on newer problems
**Practical Differences**
- Claude 3.5 Sonnet shows stronger reasoning on multi-step refactoring tasks
- GPT-4o has better instruction following for exact output formats
- Both handle large contexts well (200k+ tokens)
**Verdict**: For pure coding benchmarks, Claude 3.5 Sonnet holds a slight edge. For production workflows requiring strict format adherence, GPT-4o may be preferable.
--- Done ---Handle provider fallbacks gracefully
n4n.ai automatically routes around degraded providers, but you should still handle transient errors in your code. Wrap the agent call with a retry policy that respects Retry-After headers.
# resilience.py
import asyncio
import httpx
from tenacity import (
retry,
stop_after_attempt,
wait_exponential_jitter,
retry_if_exception_type,
)
# Exception types that warrant a retry
RETRYABLE_EXCEPTIONS = (
httpx.TimeoutException,
httpx.NetworkError,
httpx.HTTPStatusError,
)
def is_retryable_error(exc: BaseException) -> bool:
if isinstance(exc, httpx.HTTPStatusError):
# Retry on 429, 5xx, but not 4xx (except 429)
return exc.response.status_code >= 500 or exc.response.status_code == 429
return isinstance(exc, RETRYABLE_EXCEPTIONS)
@retry(
wait=wait_exponential_jitter(initial=1, max=30),
stop=stop_after_attempt(3),
retry=retry_if_exception_type(RETRYABLE_EXCEPTIONS),
reraise=True,
)
async def resilient_agent_chat(agent, query: str) -> str:
"""
Call the agent with automatic retry on transient failures.
"""
response = await agent.achat(query)
return str(response)
async def main_with_fallback():
from agent import build_agent
agent = build_agent()
query = "Summarize the latest developments in AI agent frameworks as of Q1 2025."
try:
result = await resilient_agent_chat(agent, query)
print(result)
except httpx.HTTPStatusError as e:
if e.response.status_code == 429:
retry_after = e.response.headers.get("Retry-After", "60")
print(f"Rate limited. Retry after {retry_after}s")
else:
print(f"HTTP error: {e}")
except Exception as e:
print(f"Unexpected error: {e}")
if __name__ == "__main__":
asyncio.run(main_with_fallback())The tenacity configuration uses exponential backoff with jitter — standard practice for LLM APIs. n4n.ai forwards provider Retry-After headers, so you can honor them precisely.
Add structured output for downstream consumers
If another service consumes your agent’s output, define a Pydantic schema and use LlamaIndex’s StructuredOutputParser.
# structured.py
from pydantic import BaseModel, Field
from typing import List, Optional
from llama_index.core.output_parsers import PydanticOutputParser
from llama_index.core.prompts import PromptTemplate
from agent import build_agent
class ResearchFinding(BaseModel):
claim: str = Field(description="A single factual claim")
citation: str = Field(description="Inline citation in [title](url) format")
confidence: float = Field(ge=0.0, le=1.0, description="Confidence 0-1")
class ResearchReport(BaseModel):
question: str
summary: str
findings: List[ResearchFinding]
gaps: List[str] = Field(default_factory=list, description="What we couldn't verify")
parser = PydanticOutputParser(output_cls=ResearchReport)
FORMAT_PROMPT = PromptTemplate(
"Answer the user's question and return ONLY valid JSON matching this schema:\n"
"{schema}\n\n"
"Question: {query_str}\n"
"Answer:"
)
async def structured_research(query: str) -> ResearchReport:
agent = build_agent()
# Inject format instructions into the system prompt temporarily
format_instructions = parser.get_format_string()
full_prompt = FORMAT_PROMPT.format(schema=format_instructions, query_str=query)
response = await agent.achat(full_prompt)
return parser.parse(str(response))
async def main():
report = await structured_research(
"What are the top 3 LlamaIndex agent patterns in 2024?"
)
print(f"Question: {report.question}")
print(f"Summary: {report.summary}")
print("\nFindings:")
for i, f in enumerate(report.findings, 1):
print(f" {i}. {f.claim} {f.citation} (confidence: {f.confidence:.0%})")
if report.gaps:
print("\nGaps:")
for g in report.gaps:
print(f" - {g}")
if __name__ == "__main__":
asyncio.run(main())Expected output:
Question: What are the top 3 LlamaIndex agent patterns in 2024?
Summary: The three dominant patterns are ReAct agents with tool use, multi-agent orchestration, and RAG-augmented agents with citation enforcement.
Findings:
1. ReAct-style agents remain the most widely deployed pattern for single-agent tool use [LlamaIndex 2024 State of Agents](https://example.com/state) (confidence: 90%)
2. Multi-agent frameworks like AgentWorkflow gained traction for complex workflows [LlamaIndex Blog: Multi-Agent Patterns](https://example.com/multi) (confidence: 85%)
3. RAG-augmented agents with inline citations became standard for research tasks [LangChain vs LlamaIndex Benchmark](https://example.com/bench) (confidence: 80%)
Gaps:
- Limited public benchmark data for production-scale deploymentsObservability: log tool calls and token usage
Production agents need visibility. LlamaIndex’s callback system captures everything.
# observability.py
from llama_index.core.callbacks import CallbackManager, LlamaDebugHandler
from llama_index.core.global_handler import set_global_handler
from agent import build_agent
# Enable debug handler — captures every LLM call, tool call, and token count
debug_handler = LlamaDebugHandler(print_trace_on_end=True)
callback_manager = CallbackManager([debug_handler])
set_global_handler(callback_manager)
async def main():
agent = build_agent()
await agent.achat("What is the current status of Python 3.13 adoption?")
# After the call, inspect captured events
for event in debug_handler.get_llm_events():
print(f"LLM: {event.model_name} | "
f"prompt_tokens: {event.prompt_tokens} | "
f"completion_tokens: {event.completion_tokens} | "
f"total: {event.total_tokens}")
for event in debug_handler.get_tool_events():
print(f"TOOL: {event.tool_name} | "
f"input: {event.tool_input} | "
f"output_len: {len(str(event.tool_output))}")
if __name__ == "__main__":
import asyncio
asyncio.run(main())Sample output:
LLM: anthropic/claude-3.5-sonnet | prompt_tokens: 1,247 | completion_tokens: 312 | total: 1,559
LLM: anthropic/claude-3.5-sonnet | prompt_tokens: 2,103 | completion_tokens: 187 | total: 2,290
TOOL: web_search | input: {'query': 'Python 3.13 adoption rate 2024', 'max_results': 5} | output_len: 1,842
TOOL: fetch_page | input: {'url': 'https://docs.python.org/3/whatsnew/3.13.html'} | output_len: 8,192This data feeds directly into cost tracking and latency dashboards. n4n.ai meters per-token usage on their side too, so you can reconcile.
Swap models without code changes
Because n4n.ai normalizes the model identifier, switching to a different model — say, google/gemini-1.5-pro or openai/gpt-4o — is a one-line config change. No SDK swap, no prompt rewrites.
# config.py (addition)
def get_llm(model: str = "anthropic/claude-3.5-sonnet", **kwargs) -> OpenAI:
# Override via env for A/B testing or fallback
import os
model = os.getenv("N4N_MODEL", model)
return OpenAI(
model=model,
api_key=N4N_API_KEY,
base_url=N4N_BASE_URL,
temperature=0.1,
max_tokens=4096,
**kwargs,
)N4N_MODEL=openai/gpt-4o python run_agent.pyThe agent logic, tools, and output parsing stay identical. This is the practical benefit of a unified gateway: your application code remains provider-agnostic.
What to tackle next
- Persistent memory: Swap
ChatMemoryBufferfor a vector-backed memory store to retain context across sessions. - Human-in-the-loop: Add a
human_inputtool that pauses execution for approval before sensitive actions. - Evaluation: Build a golden dataset of questions and use LlamaIndex’s
AgentEvaluatorto regression-test prompt changes. - Deployment: Containerize with a lightweight ASGI server (FastAPI + uvicorn) and put the agent behind an API gateway.
The pattern scales: same tools, same agent loop, same observability — just swap the model identifier when benchmarks or costs shift.