<!-- SPDX-License-Identifier: Apache-2.0 -->

Hamilton for LLM & AI Workflows

This skill covers patterns for building LLM applications, RAG pipelines, and AI workflows with Apache Hamilton.

Why Hamilton for LLM Workflows?

Key Benefits:

• Modular prompts: Each prompt component is a testable function
• Dependency tracking: Clear lineage from data → embeddings → retrieval → generation
• Async parallelization: Multiple LLM calls happen concurrently
• Caching: Avoid redundant expensive API calls
• Observability: Track LLM calls, costs, and performance
• Testing: Unit test prompts, retrieval, and generation separately

RAG Pipeline Pattern

Complete RAG Implementation:

"""RAG pipeline with Hamilton."""

import openai
from typing import List, Dict
import aiohttp


# 1. Document Loading & Processing
async def document_text(document_url: str) -> str:
    """Fetch document from URL."""
    async with aiohttp.ClientSession() as session:
        async with session.get(document_url) as resp:
            return await resp.text()


def document_chunks(document_text: str, chunk_size: int = 1000, overlap: int = 100) -> List[str]:
    """Split document into overlapping chunks."""
    chunks = []
    start = 0
    while start < len(document_text):
        end = start + chunk_size
        chunks.append(document_text[start:end])
        start = end - overlap
    return chunks


# 2. Embedding Generation
async def embeddings(
    document_chunks: List[str], embedding_model: str = "text-embedding-3-small"
) -> List[List[float]]:
    """Generate embeddings for chunks."""
    client = openai.AsyncOpenAI()
    response = await client.embeddings.create(input=document_chunks, model=embedding_model)
    return [item.embedding for item in response.data]


# 3. Vector Store
async def vector_store_ids(embeddings: List[List[float]], document_chunks: List[str]) -> List[str]:
    """Store embeddings in vector database."""
    import pinecone

    index = pinecone.Index("documents")

    vectors = [
        (f"chunk_{i}", emb, {"text": chunk})
        for i, (emb, chunk) in enumerate(zip(embeddings, document_chunks))
    ]
    await index.upsert(vectors)
    return [v[0] for v in vectors]


# 4. Query & Retrieval
async def query_embedding(
    query: str, embedding_model: str = "text-embedding-3-small"
) -> List[float]:
    """Generate embedding for query."""
    client = openai.AsyncOpenAI()
    response = await client.embeddings.create(input=[query], model=embedding_model)
    return response.data[0].embedding


async def retrieved_chunks(query_embedding: List[float], top_k: int = 5) -> List[str]:
    """Retrieve relevant chunks from vector store."""
    import pinecone

    index = pinecone.Index("documents")

    results = await index.query(vector=query_embedding, top_k=top_k, include_metadata=True)
    return [match["metadata"]["text"] for match in results["matches"]]


# 5. Prompt Construction
def rag_prompt(query: str, retrieved_chunks: List[str]) -> str:
    """Construct RAG prompt with context."""
    context = "\n\n".join(retrieved_chunks)
    return f"""Answer the question based on the context below.

Context:
{context}

Question: {query}

Answer:"""


# 6. LLM Generation
async def llm_response(rag_prompt: str, model: str = "gpt-4") -> str:
    """Generate response using LLM."""
    client = openai.AsyncOpenAI()
    response = await client.chat.completions.create(
        model=model, messages=[{"role": "user", "content": rag_prompt}]
    )
    return response.choices[0].message.content


# Execute the RAG pipeline
from hamilton import async_driver
import rag_pipeline

dr = await async_driver.Builder().with_modules(rag_pipeline).build()

# Indexing phase
await dr.execute(["vector_store_ids"], inputs={"document_url": "https://example.com/doc.pdf"})

# Query phase
result = await dr.execute(["llm_response"], inputs={"query": "What is the main topic?"})

Multi-Provider Pattern

Support multiple LLM providers:

"""Multi-provider LLM configuration."""

from hamilton.function_modifiers import config
import openai
import anthropic


@config.when(provider="openai")
def llm_client__openai() -> openai.AsyncOpenAI:
    """OpenAI client."""
    return openai.AsyncOpenAI()


@config.when(provider="anthropic")
def llm_client__anthropic() -> anthropic.AsyncAnthropic:
    """Anthropic client."""
    return anthropic.AsyncAnthropic()


@config.when(provider="openai")
async def llm_response__openai(llm_client: openai.AsyncOpenAI, prompt: str) -> str:
    """Generate with OpenAI."""
    response = await llm_client.chat.completions.create(
        model="gpt-4", messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content


@config.when(provider="anthropic")
async def llm_response__anthropic(llm_client: anthropic.AsyncAnthropic, prompt: str) -> str:
    """Generate with Anthropic."""
    response = await llm_client.messages.create(
        model="claude-sonnet-4-6", max_tokens=1024, messages=[{"role": "user", "content": prompt}]
    )
    return response.content[0].text


# Switch providers via config
dr = (
    await async_driver.Builder()
    .with_config({"provider": "anthropic"})
    .with_modules(llm_module)
    .build()
)

Parallel LLM Calls

Multiple analyses in parallel:

"""Run multiple LLM analyses concurrently."""

import openai


async def llm_client() -> openai.AsyncOpenAI:
    """Shared LLM client."""
    return openai.AsyncOpenAI()


def summarization_prompt(document: str) -> str:
    """Prompt for summarization."""
    return f"Summarize this document:\n\n{document}"


def keyword_prompt(document: str) -> str:
    """Prompt for keyword extraction."""
    return f"Extract 5 key topics from this document:\n\n{document}"


def sentiment_prompt(document: str) -> str:
    """Prompt for sentiment analysis."""
    return f"Analyze the sentiment of this document:\n\n{document}"


# These three run in parallel!
async def summary(llm_client: openai.AsyncOpenAI, summarization_prompt: str) -> str:
    """Generate summary."""
    response = await llm_client.chat.completions.create(
        model="gpt-4", messages=[{"role": "user", "content": summarization_prompt}]
    )
    return response.choices[0].message.content


async def keywords(llm_client: openai.AsyncOpenAI, keyword_prompt: str) -> List[str]:
    """Extract keywords."""
    response = await llm_client.chat.completions.create(
        model="gpt-4", messages=[{"role": "user", "content": keyword_prompt}]
    )
    return response.choices[0].message.content.split("\n")


async def sentiment(llm_client: openai.AsyncOpenAI, sentiment_prompt: str) -> str:
    """Analyze sentiment."""
    response = await llm_client.chat.completions.create(
        model="gpt-4", messages=[{"role": "user", "content": sentiment_prompt}]
    )
    return response.choices[0].message.content


def document_analysis(summary: str, keywords: List[str], sentiment: str) -> dict:
    """Combine all analyses."""
    return {"summary": summary, "keywords": keywords, "sentiment": sentiment}


# Result: 3 LLM calls happen concurrently, ~3x faster!

Caching LLM Calls

Save API costs with caching:

"""Cache expensive LLM calls."""

from hamilton.function_modifiers import cache
import openai


@cache(behavior="default")  # Cache LLM responses
async def document_summary(document_text: str, llm_client: openai.AsyncOpenAI) -> str:
    """Generate summary (cached).

    First call: Makes API request, caches result
    Subsequent calls: Retrieves from cache (free & instant!)
    """
    response = await llm_client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": f"Summarize this document:\n\n{document_text}"}],
    )
    return response.choices[0].message.content


# First run: Costs $0.01
# Second run: $0.00, instant retrieval
# Savings on 1000 documents: $10!

Prompt Engineering Patterns

Modular prompts:

"""Build complex prompts from components."""


def system_message(task_type: str) -> str:
    """System message based on task."""
    templates = {
        "summarize": "You are an expert at creating concise summaries.",
        "extract": "You are an expert at extracting structured information.",
        "analyze": "You are an expert at analyzing content and providing insights.",
    }
    return templates[task_type]


def user_context(document: str, max_length: int = 2000) -> str:
    """Truncate document if needed."""
    return document[:max_length] if len(document) > max_length else document


def instruction(task_type: str) -> str:
    """Task-specific instruction."""
    instructions = {
        "summarize": "Provide a 3-sentence summary.",
        "extract": "Extract key entities and dates.",
        "analyze": "Analyze the main themes and sentiment.",
    }
    return instructions[task_type]


def messages(system_message: str, user_context: str, instruction: str) -> List[dict]:
    """Combine into message list."""
    return [
        {"role": "system", "content": system_message},
        {"role": "user", "content": f"{user_context}\n\n{instruction}"},
    ]


async def llm_result(llm_client: openai.AsyncOpenAI, messages: List[dict]) -> str:
    """Execute LLM call."""
    response = await llm_client.chat.completions.create(model="gpt-4", messages=messages)
    return response.choices[0].message.content

Structured Output with Pydantic

Parse LLM output into structured data:

"""Structured extraction with validation."""

from pydantic import BaseModel, Field
from typing import List
import openai


class ExtractedEntity(BaseModel):
    """Structured entity."""

    name: str = Field(description="Entity name")
    type: str = Field(description="Entity type (person, org, location)")
    relevance: float = Field(description="Relevance score 0-1", ge=0, le=1)


class ExtractionResult(BaseModel):
    """Complete extraction result."""

    entities: List[ExtractedEntity]
    summary: str


async def structured_extraction(document: str, llm_client: openai.AsyncOpenAI) -> ExtractionResult:
    """Extract structured data from document."""
    response = await llm_client.beta.chat.completions.parse(
        model="gpt-4o-2024-08-06",
        messages=[{"role": "user", "content": f"Extract entities from:\n\n{document}"}],
        response_format=ExtractionResult,
    )
    return response.choices[0].message.parsed

Agent Patterns

Multi-step agent with Hamilton:

"""Agent with tool use."""

from typing import Literal


def agent_prompt(query: str, available_tools: List[str]) -> str:
    """Create agent prompt with tools."""
    tools_desc = "\n".join(f"- {tool}" for tool in available_tools)
    return f"""You have access to these tools:
{tools_desc}

User query: {query}

What tool should be used? Respond with just the tool name."""


async def tool_selection(
    llm_client: openai.AsyncOpenAI, agent_prompt: str
) -> Literal["search", "calculate", "summarize"]:
    """LLM selects which tool to use."""
    response = await llm_client.chat.completions.create(
        model="gpt-4", messages=[{"role": "user", "content": agent_prompt}]
    )
    return response.choices[0].message.content.strip().lower()


@config.when_in(tool_selection=["search"])
async def tool_result__search(query: str) -> str:
    """Execute search tool."""
    # Implement search logic
    return "Search results..."


@config.when_in(tool_selection=["calculate"])
def tool_result__calculate(query: str) -> str:
    """Execute calculation tool."""
    # Implement calculation logic
    return "Calculation result..."


async def final_response(llm_client: openai.AsyncOpenAI, query: str, tool_result: str) -> str:
    """Generate final response with tool result."""
    response = await llm_client.chat.completions.create(
        model="gpt-4",
        messages=[
            {
                "role": "user",
                "content": f"User asked: {query}\nTool returned: {tool_result}\n\nProvide final answer:",
            }
        ],
    )
    return response.choices[0].message.content

Testing LLM Workflows

Unit test prompts and logic:

"""Test LLM components."""

import pytest


def test_prompt_construction():
    """Test prompt building logic."""
    from llm_module import rag_prompt

    query = "What is Hamilton?"
    chunks = ["Hamilton is a framework", "It builds DAGs"]

    prompt = rag_prompt(query, chunks)

    assert "Hamilton is a framework" in prompt
    assert "What is Hamilton?" in prompt
    assert "Context:" in prompt


async def test_retrieval():
    """Test retrieval logic (mock vector store)."""
    # Mock vector store responses
    # Test retrieval function
    pass


def test_structured_output():
    """Test Pydantic parsing."""
    from llm_module import ExtractionResult, ExtractedEntity

    result = ExtractionResult(
        entities=[ExtractedEntity(name="Hamilton", type="product", relevance=0.9)],
        summary="A framework for building DAGs",
    )

    assert len(result.entities) == 1
    assert result.entities[0].name == "Hamilton"

Best Practices

1. Modularize prompts - Each component is a testable function
2. Cache aggressively - LLM calls are expensive
3. Use async - Parallelize independent LLM calls
4. Structure outputs - Use Pydantic for parsing
5. Handle failures - Add retry logic and fallbacks
6. Track costs - Monitor token usage
7. Version prompts - Use config for prompt variants

Additional Resources

• For async patterns, use /hamilton-scale
• For observability, use /hamilton-observability
• Apache Hamilton LLM examples: github.com/apache/hamilton/tree/main/examples/LLM_Workflows