Ch 2 — Agent Frameworks¶

from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

llm = ChatAnthropic(model="claude-opus-4-5", temperature=0)

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a concise technical writer."),
        ("human", "Explain {topic} in exactly two sentences."),
    ]
)

chain = prompt | llm | StrOutputParser()

result: str = chain.invoke({"topic": "vector databases"})
print(result)

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_anthropic import ChatAnthropic
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_core.prompts import ChatPromptTemplate

llm = ChatAnthropic(model="claude-opus-4-5")
tools = [DuckDuckGoSearchRun()]

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant with access to web search."),
        ("human", "{input}"),
        ("placeholder", "{agent_scratchpad}"),
    ]
)

agent = create_tool_calling_agent(llm=llm, tools=tools, prompt=prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=8)

result = executor.invoke({"input": "What is the population of Tokyo as of 2024?"})
print(result["output"])

from langchain_anthropic import ChatAnthropic
from langchain_openai import ChatOpenAI  # fallback provider
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

primary_llm = ChatAnthropic(model="claude-opus-4-5")
fallback_llm = ChatOpenAI(model="gpt-4o")

prompt = ChatPromptTemplate.from_template("Summarise: {text}")
parser = StrOutputParser()

# If the primary LLM raises an exception, fall back to OpenAI
chain = (prompt | primary_llm.with_fallbacks([fallback_llm]) | parser)

summary = chain.invoke({"text": "LangChain is a framework for building LLM applications..."})

from llama_index.core.query_engine import RouterQueryEngine
from llama_index.core.selectors import LLMSingleSelector
from llama_index.core.tools import QueryEngineTool

financial_engine = ...  # query engine over financial documents
technical_engine = ...  # query engine over API docs

tools = [
    QueryEngineTool.from_defaults(
        query_engine=financial_engine,
        description="Use for questions about revenue, costs, and financial metrics.",
    ),
    QueryEngineTool.from_defaults(
        query_engine=technical_engine,
        description="Use for questions about APIs, code, and technical specifications.",
    ),
]

router_engine = RouterQueryEngine(
    selector=LLMSingleSelector.from_defaults(),
    query_engine_tools=tools,
)

from llama_index.core.query_engine import SubQuestionQueryEngine

sub_question_engine = SubQuestionQueryEngine.from_defaults(
    query_engine_tools=tools,
    verbose=True,
)

response = sub_question_engine.query(
    "Compare Q3 revenue and explain the API rate-limit changes introduced in the same quarter."
)

"""
doc_qa_agent.py — LlamaIndex document Q&A agent with ReAct loop.

Requirements:
    pip install llama-index>=0.10 llama-index-llms-anthropic \
                llama-index-embeddings-openai pypdf
"""

from __future__ import annotations

import os
from pathlib import Path

from llama_index.core import (
    Settings,
    SimpleDirectoryReader,
    VectorStoreIndex,
)
from llama_index.core.agent import ReActAgent
from llama_index.core.tools import QueryEngineTool, ToolMetadata
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.llms.anthropic import Anthropic


def build_document_agent(docs_dir: str | Path) -> ReActAgent:
    """
    Ingest documents from `docs_dir`, build a vector index,
    and wrap it in a LlamaIndex ReActAgent.

    Args:
        docs_dir: Directory containing PDF, TXT, or Markdown files.

    Returns:
        A ReActAgent ready to answer questions about the documents.
    """
    docs_path = Path(docs_dir)
    if not docs_path.exists():
        raise FileNotFoundError(f"Document directory not found: {docs_path}")

    # Configure global LLM and embedding model
    Settings.llm = Anthropic(
        model="claude-opus-4-5",
        api_key=os.environ["ANTHROPIC_API_KEY"],
        max_tokens=4096,
    )
    Settings.embed_model = OpenAIEmbedding(
        model="text-embedding-3-small",
        api_key=os.environ["OPENAI_API_KEY"],
    )

    # Load and index documents
    print(f"Loading documents from {docs_path} ...")
    documents = SimpleDirectoryReader(
        input_dir=str(docs_path),
        recursive=True,
        required_exts=[".pdf", ".txt", ".md"],
    ).load_data()
    print(f"Loaded {len(documents)} document chunks.")

    index = VectorStoreIndex.from_documents(documents, show_progress=True)
    query_engine = index.as_query_engine(similarity_top_k=6)

    # Wrap query engine as an agent tool
    query_tool = QueryEngineTool(
        query_engine=query_engine,
        metadata=ToolMetadata(
            name="document_search",
            description=(
                "Search the document corpus and retrieve relevant information. "
                "Input should be a precise natural-language question."
            ),
        ),
    )

    agent = ReActAgent.from_tools(
        tools=[query_tool],
        llm=Settings.llm,
        verbose=True,
        max_iterations=10,
        context=(
            "You are a knowledgeable assistant that answers questions "
            "strictly based on the provided documents. "
            "If the answer is not in the documents, say so clearly."
        ),
    )

    return agent


def main() -> None:
    docs_directory = os.environ.get("DOCS_DIR", "./documents")
    agent = build_document_agent(docs_directory)

    print("\nDocument Q&A Agent ready. Type 'quit' to exit.\n")
    while True:
        question = input("Question: ").strip()
        if question.lower() in {"quit", "exit", "q"}:
            break
        if not question:
            continue
        response = agent.chat(question)
        print(f"\nAnswer: {response}\n")


if __name__ == "__main__":
    main()

from llama_index.core import StorageContext, load_index_from_storage
from llama_index.core.storage.docstore import SimpleDocumentStore
from llama_index.core.storage.index_store import SimpleIndexStore
from llama_index.core.vector_stores import SimpleVectorStore

INDEX_PERSIST_DIR = "./index_storage"

def save_index(index: VectorStoreIndex) -> None:
    index.storage_context.persist(persist_dir=INDEX_PERSIST_DIR)

def load_index() -> VectorStoreIndex:
    storage_context = StorageContext.from_defaults(
        persist_dir=INDEX_PERSIST_DIR
    )
    return load_index_from_storage(storage_context)

import anthropic
import os

client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

def stream_agent_step(
    messages: list[dict],
    tools: list[dict],
    system: str,
) -> tuple[list, str]:
    """
    Run one streaming step of the agent loop.

    Returns:
        (content_blocks, stop_reason)
    """
    content_blocks = []
    stop_reason = ""

    with client.messages.stream(
        model="claude-opus-4-5",
        max_tokens=4096,
        system=system,
        tools=tools,
        messages=messages,
    ) as stream:
        for event in stream:
            if hasattr(event, "type"):
                if event.type == "content_block_delta":
                    if hasattr(event.delta, "text"):
                        print(event.delta.text, end="", flush=True)

        final_message = stream.get_final_message()
        content_blocks = final_message.content
        stop_reason = final_message.stop_reason

    print()  # newline after streamed text
    return content_blocks, stop_reason

from langfuse import Langfuse
from langfuse.decorators import langfuse_context, observe
import os

langfuse = Langfuse(
    public_key=os.environ["LANGFUSE_PUBLIC_KEY"],
    secret_key=os.environ["LANGFUSE_SECRET_KEY"],
    host=os.environ.get("LANGFUSE_HOST", "https://cloud.langfuse.com"),
)


@observe(name="agent_run")
def traced_agent_run(user_message: str) -> str:
    """Agent run decorated with Langfuse tracing."""
    langfuse_context.update_current_trace(
        name="ReAct Agent",
        user_id="user-123",
        tags=["production", "v2"],
    )

    # ... your agent loop here ...
    answer = run_agent(user_message)

    langfuse_context.update_current_observation(
        output=answer,
        metadata={"model": "claude-opus-4-5"},
    )
    return answer

# Fetch a prompt from Langfuse prompt management
prompt_template = langfuse.get_prompt("react-agent-system", version=3)
system_prompt: str = prompt_template.compile(agent_name="Research Assistant")

from jinja2 import Template

SYSTEM_TEMPLATE = Template(
    """\
You are {{ agent_name }}, an AI assistant with the following tools:
{% for tool in tools %}
- **{{ tool.name }}**: {{ tool.description }}
{% endfor %}

Current date: {{ current_date }}
User tier: {{ user_tier }}
"""
)

system_prompt = SYSTEM_TEMPLATE.render(
    agent_name="ResearchBot",
    tools=TOOL_SCHEMAS,
    current_date="2025-07-15",
    user_tier="enterprise",
)

import time
import random
import anthropic

client = anthropic.Anthropic()


def call_with_retry(
    messages: list[dict],
    max_retries: int = 3,
    base_delay: float = 1.0,
) -> anthropic.types.Message:
    """Call the Anthropic API with exponential backoff on transient errors."""
    for attempt in range(max_retries):
        try:
            return client.messages.create(
                model="claude-opus-4-5",
                max_tokens=4096,
                messages=messages,
            )
        except anthropic.RateLimitError:
            if attempt == max_retries - 1:
                raise
            delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
            print(f"Rate limited. Retrying in {delay:.1f}s ...")
            time.sleep(delay)
        except anthropic.APIStatusError as exc:
            if exc.status_code >= 500:  # Server error — retry
                if attempt == max_retries - 1:
                    raise
                time.sleep(base_delay * (2 ** attempt))
            else:
                raise  # Client error — don't retry
    raise RuntimeError("Unreachable")

import threading
import time
from collections import deque


class RateLimiter:
    """Sliding window rate limiter for API calls."""

    def __init__(self, max_requests: int, window_seconds: float) -> None:
        self.max_requests = max_requests
        self.window = window_seconds
        self._timestamps: deque[float] = deque()
        self._lock = threading.Lock()

    def acquire(self) -> None:
        """Block until a request slot is available."""
        with self._lock:
            now = time.monotonic()
            # Evict timestamps outside the window
            while self._timestamps and self._timestamps[0] < now - self.window:
                self._timestamps.popleft()
            if len(self._timestamps) >= self.max_requests:
                sleep_for = self.window - (now - self._timestamps[0])
                time.sleep(max(sleep_for, 0))
            self._timestamps.append(time.monotonic())


# Example: 60 requests per minute
limiter = RateLimiter(max_requests=60, window_seconds=60.0)

COST_PER_MILLION_INPUT = 3.00   # USD for claude-opus-4-5
COST_PER_MILLION_OUTPUT = 15.00

def compute_cost(input_tokens: int, output_tokens: int) -> float:
    """Return the cost in USD for a single API call."""
    return (
        input_tokens * COST_PER_MILLION_INPUT / 1_000_000
        + output_tokens * COST_PER_MILLION_OUTPUT / 1_000_000
    )

# After each API call:
# response.usage.input_tokens, response.usage.output_tokens