TTA.dev LLM Primitives

TTA.dev exposes LLMs through two primitives:

make_resilient_llm() is the one-liner factory that composes LiteLLMPrimitive + RetryPrimitive + CachePrimitive into a production-ready stack.

When to Use This Skill

• Making any LLM API call in TTA.dev
• Choosing the right model string for Groq, Anthropic, Ollama, Gemini, or OpenRouter
• Adding streaming to an LLM call
• Using LLM tool-calling (function calling)
• Setting up Langfuse zero-config tracing
• Building a primary/fallback LLM architecture

Prerequisites

uv sync                        # core + litellm included
uv sync --extra groq           # Groq-specific SDK (optional — litellm handles it)
uv sync --extra anthropic      # Anthropic SDK (optional)

Environment variables:

Model String Format

litellm uses "<provider>/<model-name>":

"groq/llama-3.1-8b-instant"
"groq/llama-3.3-70b-versatile"
"anthropic/claude-3-5-haiku-20241022"
"anthropic/claude-3-5-sonnet-20241022"
"ollama/llama3.2"
"ollama/mistral"
"gemini/gemini-2.0-flash"
"openrouter/meta-llama/llama-3-70b-instruct"

See references/provider-reference.md for the full table.

LLMRequest and LLMResponse

from ttadev.primitives.llm import LLMRequest, LLMResponse

request = LLMRequest(
    model="groq/llama-3.1-8b-instant",   # full litellm string
    messages=[{"role": "user", "content": "Hello!"}],
    system="You are a helpful assistant.",  # optional system prompt
    temperature=0.7,                        # default 0.7
    max_tokens=512,                         # optional
)

# LLMResponse fields
response: LLMResponse
response.content    # str — generated text
response.model      # str — model that responded
response.provider   # str — provider name
response.usage      # dict — {"prompt_tokens": ..., "completion_tokens": ...}
response.tool_calls # list[ToolCall] | None — when tool-calling is used

Quick Start — Single LLM Call

import asyncio
from ttadev.primitives import LiteLLMPrimitive, WorkflowContext
from ttadev.primitives.llm import LLMRequest

llm = LiteLLMPrimitive()
ctx = WorkflowContext.root("llm-demo")

response = asyncio.run(llm.execute(
    LLMRequest(
        model="groq/llama-3.1-8b-instant",
        messages=[{"role": "user", "content": "What is 2 + 2?"}],
    ),
    ctx,
))
print(response.content)   # "4"

make_resilient_llm — Production One-Liner

make_resilient_llm() composes CachePrimitive → RetryPrimitive → LiteLLMPrimitive in one call:

from ttadev.primitives import make_resilient_llm, WorkflowContext
from ttadev.primitives.llm import LLMRequest

llm = make_resilient_llm(
    "groq",                                              # provider prefix
    litellm_fallbacks=["anthropic/claude-3-haiku-20240307", "ollama/llama3.2"],
    retry_attempts=3,       # default
    cache=True,             # default — SHA-256 keyed on messages+model+params
    cache_ttl_seconds=3600, # default — 1 hour
)

ctx      = WorkflowContext.root("resilient-demo")
response = asyncio.run(llm.execute(
    LLMRequest(model="llama-3.1-8b-instant", messages=[...]),
    ctx,
))

The composed stack:

CachePrimitive       ← keyed on messages + model + params
  └─ RetryPrimitive  ← 3 attempts, exponential back-off
       └─ LiteLLMPrimitive(provider="groq", fallbacks=[...])

Streaming

Use llm.stream(request, ctx) — returns an AsyncIterator[str]:

import asyncio
from ttadev.primitives import LiteLLMPrimitive, WorkflowContext
from ttadev.primitives.llm import LLMRequest

async def stream_demo() -> None:
    llm = LiteLLMPrimitive()
    ctx = WorkflowContext.root("stream-demo")
    req = LLMRequest(
        model="groq/llama-3.1-8b-instant",
        messages=[{"role": "user", "content": "Tell me a short story."}],
    )
    async for chunk in llm.stream(req, ctx):
        print(chunk, end="", flush=True)
    print()

asyncio.run(stream_demo())

Tool Calls

Define tools with ToolSchema, pass them in LLMRequest.tools:

from ttadev.primitives import LiteLLMPrimitive, WorkflowContext
from ttadev.primitives.llm import LLMRequest, ToolSchema
import asyncio

weather_tool = ToolSchema(
    name="get_weather",
    description="Return current weather for a city.",
    parameters={
        "type": "object",
        "properties": {
            "location": {"type": "string", "description": "City name"},
        },
        "required": ["location"],
    },
)

llm = LiteLLMPrimitive()
ctx = WorkflowContext.root("tool-demo")
req = LLMRequest(
    model="groq/llama-3.3-70b-versatile",
    messages=[{"role": "user", "content": "What is the weather in Paris?"}],
    tools=[weather_tool],
    tool_choice="auto",
)
response = asyncio.run(llm.execute(req, ctx))
if response.tool_calls:
    for tc in response.tool_calls:
        print(tc.name, tc.arguments)  # get_weather {"location": "Paris"}

Langfuse Observability — Zero Config

Set env vars; tracing is automatic:

export LANGFUSE_SECRET_KEY=sk-lf-...
export LANGFUSE_PUBLIC_KEY=pk-lf-...
# LANGFUSE_HOST defaults to https://cloud.langfuse.com

Every LiteLLMPrimitive.execute() call is traced. No code changes needed.

Primary / Fallback Architecture

Use FallbackPrimitive to fall back from LiteLLMPrimitive to UniversalLLMPrimitive:

from ttadev.primitives import FallbackPrimitive, WorkflowContext
from ttadev.primitives.llm import LiteLLMPrimitive, UniversalLLMPrimitive, LLMProvider

primary  = LiteLLMPrimitive()                             # litellm — primary
fallback = UniversalLLMPrimitive(provider=LLMProvider.GROQ)  # direct SDK — fallback

resilient = FallbackPrimitive(primary, fallback)

ctx      = WorkflowContext.root("primary-fallback")
response = asyncio.run(resilient.execute(request, ctx))

Anti-Patterns

Provider Reference

Full table of model strings by provider → references/provider-reference.md