hamilton-llm

name	hamilton-llm
description	LLM and AI workflow patterns for Hamilton including RAG pipelines, embeddings, vector databases, and prompt engineering. Use for building AI applications with Hamilton.
allowed-tools	Read, Grep, Glob, Bash(python:*), Bash(pytest:*)
user-invocable	true
disable-model-invocation	false

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[]]:
    
    client = openai.AsyncOpenAI()
    response =  client.embeddings.create(
        =document_chunks,
        model=embedding_model
    )
     [item.embedding  item  response.data]


  () -> []:
    
     pinecone
    index = pinecone.Index()

    vectors = [
        (, emb, {: chunk})
         i, (emb, chunk)  ((embeddings, document_chunks))
    ]
     index.upsert(vectors)
     [v[]  v  vectors]


  () -> []:
    
    client = openai.AsyncOpenAI()
    response =  client.embeddings.create(
        =[query],
        model=embedding_model
    )
     response.data[].embedding

  () -> []:
    
     pinecone
    index = pinecone.Index()

    results =  index.query(
        vector=query_embedding,
        top_k=top_k,
        include_metadata=
    )
     [[][]    results[]]


 () -> :
    
    context = .join(retrieved_chunks)
     


  () -> :
    
    client = openai.AsyncOpenAI()
    response =  client.chat.completions.create(
        model=model,
        messages=[{: , : rag_prompt}]
    )
     response.choices[].message.content


 hamilton  async_driver
 rag_pipeline

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


 dr.execute(
    [],
    inputs={: }
)


result =  dr.execute(
    [],
    inputs={: }
)

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=,
        max_tokens=,
        messages=[{: , : prompt}]
    )
     response.content[].text


dr =  async_driver.Builder()\
    .with_config({: })\
    .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}]
    )
     response.choices[].message.content

  () -> []:
    
    response =  llm_client.chat.completions.create(
        model=,
        messages=[{: , : keyword_prompt}]
    )
     response.choices[].message.content.split()

  () -> :
    
    response =  llm_client.chat.completions.create(
        model=,
        messages=[{: , : sentiment_prompt}]
    )
     response.choices[].message.content

 () -> :
    
     {: summary, : keywords, : sentiment}

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[]:
    
     [
        {: , : system_message},
        {: , : }
    ]

  () -> :
    
    response =  llm_client.chat.completions.create(
        model=,
        messages=messages
    )
     response.choices[].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."""
    
     


 () -> :
    
    
     

  () -> :
    
    response =  llm_client.chat.completions.create(
        model=,
        messages=[{
            : ,
            : 
        }]
    )
     response.choices[].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