Llm Integration Patterns

Temperature Guidelines

Prompt Engineering

JSON Output Pattern

prompt = f"""You are helping determine {field} for {context}.

Requirements:
- Return ONLY valid JSON
- No markdown formatting
- Include all required fields

Return JSON:
{{
    "field1": "...",
    "field2": [...],
    "mappings": {{...}}
}}

Generate now:"""

Parsing JSON from LLM

def _parse_llm_response(content: str) -> dict:
    import json
    
    try:
        # Try markdown block first
        if "```json" in content:
            json_str = content.split("```json")[1].split("```")[0]
        elif "{" in content:
            # Find JSON bounds
            start = content.index("{")
            end = content.rindex("}") + 1
            json_str = content[start:end]
        else:
            raise ValueError("No JSON found")
        
        return json.loads(json_str)
    except Exception:
        # Return minimal fallback
        return {"error": "parse_failed", "raw": content[:200]}

Fallback Pattern

ALWAYS have fallback when LLM fails:

async def generate_with_llm(input_data: InputData) -> Result:
    try:
        response = await llm.ainvoke(prompt)
        result = parse_response(response.content)
        return Result(**result)
        
    except Exception as e:
        logger.warning(f"LLM failed, using fallback: {e}")
        return _fallback_result(input_data)

def _fallback_result(input_data: InputData) -> Result:
    """Template-based fallback when LLM unavailable."""
    return Result(
        content=f"Default for {input_data.type}...",
        source="fallback",
    )

Activity Logging & Cost Tracking (REQUIRED)

Every LLM call MUST be logged with model details and AUD cost estimation.

from src.app.core.llm.config import LLMConfig, LLMTier
from src.app.core.activity.logger import get_activity_logger

logger = get_activity_logger()

# 1. Before call
await logger.log_activity(
    trace_id=trace_id,
    agent="module.component",
    activity_type="llm_call_started",
    details={
        "tier": LLMTier.PREMIUM,
        "purpose": "query_analysis",
    }
)

# 2. After call (calculate and log cost)
tokens_in = len(prompt.split()) * 1.3
tokens_out = len(response.content.split()) * 1.3
cost_aud = LLMConfig.estimate_cost(LLMTier.PREMIUM, int(tokens_in), int(tokens_out), currency="AUD")

await logger.log_llm_call(
    trace_id=trace_id,
    model_id="anthropic/claude-sonnet-4.5",
    prompt=prompt[:500],
    response=response.content[:500],
    duration_ms=duration,
    cost_aud=cost_aud, # Log AUD for user transparency
)

Error Handling

from langchain_core.exceptions import OutputParserException