ai-engineer

Use when building production LLM applications — designing RAG pipelines, choosing vector databases, implementing agent orchestration, optimizing cost, or adding AI safety guardrails.

SKILL.md

GitHub

AI Engineering

Patterns for building production-grade LLM applications, RAG systems, and intelligent agents.

When to Activate

  • Building or improving RAG systems, LLM features, or AI agent workflows
  • Selecting models, vector databases, or embedding strategies
  • Optimizing retrieval quality, latency, or inference cost
  • Implementing AI safety guardrails, content moderation, or PII handling
  • Integrating multimodal inputs (images, audio, documents) into AI pipelines
  • Designing multi-agent coordination or agentic tool-use loops
  • Setting up AI observability, evaluation, or A/B testing

Model Selection

Model Best For Relative Cost
claude-opus-4-6 Complex reasoning, architecture, research High
claude-sonnet-4-6 Balanced coding, most development tasks Medium
claude-haiku-4-5 Classification, extraction, high-volume tasks Low
GPT-4o OpenAI tool ecosystem, function calling Medium-High
Llama 3.1 70B (local) Air-gapped, cost-sensitive, no PII risk None (infra cost)

Default to Sonnet-class models for development. Use Haiku/mini variants for high-throughput steps. Reserve Opus/GPT-4o for reasoning-heavy tasks.

RAG Architecture

Chunking Strategy

# BAD: Fixed-size splits break semantic units
text_splitter = CharacterTextSplitter(chunk_size=500)

# GOOD: Semantic chunking preserves context
from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=64,
    separators=["\n\n", "\n", ". ", " ", ""]
)
Strategy Use When
Recursive character General prose, code
Semantic (sentence-transformers) Mixed-length documents
Document-structure aware PDFs, HTML, Markdown
Sliding window Dense technical content

Vector Database Selection

DB Hosted Self-Hosted Hybrid Search Notes
Pinecone Yes No Yes Managed, serverless
Qdrant Yes Yes Yes Rust core, fast filtering
Weaviate Yes Yes Yes GraphQL API
pgvector Via Supabase Yes With tsvector Great if already on Postgres
Chroma No Yes No Local dev only

Hybrid Search (Vector + Keyword)

from qdrant_client import QdrantClient
from qdrant_client.models import SparseVector, NamedSparseVector

# Dense vector (semantic) + sparse vector (BM25)
results = client.query_points(
    collection_name="docs",
    prefetch=[
        models.Prefetch(query=dense_embedding, using="dense", limit=20),
        models.Prefetch(query=SparseVector(indices=bm25_indices, values=bm25_values),
                        using="sparse", limit=20),
    ],
    query=models.FusionQuery(fusion=models.Fusion.RRF),  # Reciprocal Rank Fusion
    limit=10,
)

Reranking

# BAD: Return top-k by vector similarity alone
results = index.query(vector=embedding, top_k=5)

# GOOD: Over-fetch then rerank for precision
candidates = index.query(vector=embedding, top_k=20)

import cohere
co = cohere.Client()
reranked = co.rerank(
    model="rerank-english-v3.0",
    query=user_query,
    documents=[r.metadata["text"] for r in candidates.matches],
    top_n=5,
)

RAG Pipeline Patterns

Pattern What It Solves
HyDE (Hypothetical Document Embeddings) Query/document embedding mismatch
RAG-Fusion Single query too narrow — runs multiple query variants
Self-RAG Model decides when retrieval is needed
GraphRAG Multi-hop reasoning across connected entities
Contextual compression Retrieved chunks too noisy; extract relevant spans only 
# HyDE: generate a hypothetical answer, embed it, retrieve similar docs
hyde_prompt = f"Write a paragraph that would answer: {query}"
hypothetical_doc = llm.invoke(hyde_prompt)
hyde_embedding = embedder.embed(hypothetical_doc)
results = vector_store.similarity_search_by_vector(hyde_embedding)

Agent Orchestration

Agentic Loop (LangGraph)

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    tool_calls_remaining: int

def agent_node(state: AgentState):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def tool_node(state: AgentState):
    last_message = state["messages"][-1]
    results = execute_tools(last_message.tool_calls)
    return {"messages": results, "tool_calls_remaining": state["tool_calls_remaining"] - 1}

def should_continue(state: AgentState):
    last = state["messages"][-1]
    if not last.tool_calls or state["tool_calls_remaining"] <= 0:
        return END
    return "tools"

graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.add_edge("tools", "agent")
graph.add_conditional_edges("agent", should_continue)

Agent Memory Patterns

Type Storage Use For
Short-term In-memory messages list Current conversation context
Long-term Vector store + summary Facts, preferences across sessions
Episodic Structured DB Past task outcomes
Procedural Prompt / tool definitions Skills, workflows 
# Summarize + compress conversation memory
from langchain.memory import ConversationSummaryBufferMemory

memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=1000,  # summarize once buffer exceeds limit
    return_messages=True,
)

Multi-Agent Pattern (CrewAI)

from crewai import Agent, Task, Crew

researcher = Agent(
    role="Senior Researcher",
    goal="Find accurate information",
    tools=[search_tool, browse_tool],
    llm=llm,
)
writer = Agent(
    role="Technical Writer",
    goal="Synthesize research into clear prose",
    llm=llm,
)

research_task = Task(description="Research {topic}", agent=researcher)
write_task = Task(description="Write a report based on research", agent=writer)

crew = Crew(agents=[researcher, writer], tasks=[research_task, write_task])
result = crew.kickoff(inputs={"topic": "vector databases"})

Prompt Engineering

Structured Output

# TypeScript: Zod schema → structured output
import Anthropic from "@anthropic-ai/sdk";
import { z } from "zod";
import zodToJsonSchema from "zod-to-json-schema";

const ExtractedData = z.object({
  entities: z.array(z.object({ name: z.string(), type: z.string() })),
  summary: z.string(),
});

const response = await client.messages.create({
  model: "claude-sonnet-4-6",
  max_tokens: 1024,
  tools: [{
    name: "extract_data",
    description: "Extract structured data from text",
    input_schema: zodToJsonSchema(ExtractedData),
  }],
  tool_choice: { type: "tool", name: "extract_data" },
  messages: [{ role: "user", content: `Extract from: ${text}` }],
});

Prompting Techniques

Technique When to Use
Chain-of-thought (Think step by step) Multi-step reasoning, math, logic
Tree-of-thoughts Exploring multiple solution paths
Self-consistency Sample multiple outputs, majority vote
Few-shot examples Consistent formatting, specialized tasks
Constitutional self-critique Safety checks, tone alignment 
# BAD: Vague instruction
"Summarize this document"

# GOOD: Structured, constrained prompt
"""Summarize the following document in exactly 3 bullet points.
Each bullet must start with a verb and be under 20 words.
Focus only on actionable findings.

Document:
{document}"""

Production Patterns

Semantic Caching

from semantic_router.encoders import OpenAIEncoder
from semantic_router.layer import RouteLayer

# Cache responses for semantically similar queries
cache = RedisSemanticCache(
    redis_url="redis://localhost:6379",
    embedding=OpenAIEmbeddings(),
    score_threshold=0.95,  # cosine similarity threshold
)

@cache
def get_answer(query: str) -> str:
    return llm.invoke(query)

Streaming with FastAPI

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import anthropic

app = FastAPI()
client = anthropic.Anthropic()

@app.post("/chat")
async def chat(query: str):
    async def generate():
        with client.messages.stream(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            messages=[{"role": "user", "content": query}],
        ) as stream:
            for text in stream.text_stream:
                yield f"data: {text}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

Cost Controls

# Estimate cost before execution
def estimate_cost(prompt: str, model: str = "claude-sonnet-4-6") -> float:
    input_tokens = len(prompt) // 4  # rough estimate
    # Sonnet 4.6: $3/M input, $15/M output
    return (input_tokens / 1_000_000) * 3.0

# Hard cap: reject if estimated cost exceeds threshold
if estimate_cost(prompt) > 0.10:
    raise ValueError("Prompt too large for single request — chunk it")

AI Safety & Guardrails

Prompt Injection Detection

INJECTION_PATTERNS = [
    r"ignore (previous|above|all) instructions",
    r"you are now",
    r"disregard your",
    r"new persona",
    r"act as (if you are|a)?",
]

def detect_injection(user_input: str) -> bool:
    import re
    return any(re.search(p, user_input, re.IGNORECASE) for p in INJECTION_PATTERNS)

# Wrap all user inputs
if detect_injection(user_message):
    return {"error": "Input rejected"}

PII Redaction

import re

PII_PATTERNS = {
    "email": r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b",
    "ssn": r"\b\d{3}-\d{2}-\d{4}\b",
    "credit_card": r"\b(?:\d{4}[- ]?){3}\d{4}\b",
    "phone": r"\b\+?1?\s?\(?\d{3}\)?[\s.-]?\d{3}[\s.-]?\d{4}\b",
}

def redact_pii(text: str) -> str:
    for label, pattern in PII_PATTERNS.items():
        text = re.sub(pattern, f"[{label.upper()}_REDACTED]", text)
    return text

Content Moderation

# Use OpenAI Moderation API as a pre-filter (free)
import openai

def is_safe(text: str) -> bool:
    result = openai.moderations.create(input=text)
    return not result.results[0].flagged

# Gate all user inputs
if not is_safe(user_message):
    return {"error": "Message violates content policy"}

AI Observability

LangSmith Tracing

import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-key"
os.environ["LANGCHAIN_PROJECT"] = "my-rag-app"

# All LangChain calls now auto-traced — no code changes needed
chain = prompt | llm | output_parser
result = chain.invoke({"query": user_query})

Custom Metrics (Prometheus)

from prometheus_client import Counter, Histogram

llm_requests = Counter("llm_requests_total", "Total LLM API calls", ["model", "status"])
llm_latency = Histogram("llm_latency_seconds", "LLM response latency", ["model"])
retrieval_score = Histogram("retrieval_relevance_score", "RAG retrieval scores")

with llm_latency.labels(model="claude-sonnet-4-6").time():
    response = client.messages.create(...)
llm_requests.labels(model="claude-sonnet-4-6", status="success").inc()

RAG Evaluation

Metric Tool Measures
Context Precision RAGAS Are retrieved chunks relevant?
Context Recall RAGAS Did retrieval miss needed chunks?
Answer Faithfulness RAGAS Does answer match retrieved context?
Answer Relevance RAGAS Does answer address the question? 
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision

dataset = Dataset.from_dict({
    "question": questions,
    "answer": answers,
    "contexts": retrieved_contexts,
    "ground_truth": expected_answers,
})
scores = evaluate(dataset, metrics=[faithfulness, answer_relevancy, context_precision])

Red Flags

  • RAG without evaluating retrieval quality — high embedding similarity doesn't mean the retrieved chunks answer the question; evaluate retrieval precision/recall separately from generation quality
  • Chunk size chosen arbitrarily — too large floods context with irrelevant content, too small loses coherence; benchmark chunk sizes against real queries before committing
  • No evals before deploying prompt changes — changing a prompt in production without a regression suite is the AI equivalent of deploying untested code; build evals first
  • Unlimited agent loops — an agent without a max-turn ceiling can loop indefinitely on ambiguous tasks; always set a hard limit and define graceful stopping behavior
  • User-provided text injected directly into system prompts — prompt injection via user content can override instructions; sanitize input and keep it in the user message role, never the system role
  • Cost estimation deferred until after launch — LLM costs scale with tokens × requests; estimate per-request cost at the architecture stage, not post-launch when it's too expensive to change
  • Single embedding model for all content types — code, prose, and tables have different semantic spaces; benchmark domain-appropriate models or separate indexes per content type

Checklist

Before shipping an AI feature:

  • Model pinned to specific version, not latest
  • max_tokens set explicitly; generation budget validated against cost
  • Prompt injection detection applied to all user-controlled inputs
  • PII redaction runs before sending data to external models
  • Content moderation gate in place for user-facing endpoints
  • Retrieval quality measured (context precision/recall via RAGAS or equivalent)
  • Semantic caching enabled for repeated or near-duplicate queries
  • Streaming used for responses >200 tokens to avoid client timeouts
  • Retry with exponential backoff on rate-limit and transient errors
  • LLM calls traced (LangSmith, Phoenix, or custom spans)
  • Latency and token-usage metrics emitted to monitoring stack
  • Fallback model or graceful degradation path defined
  • Chunking strategy validated on representative documents
  • Reranker in place if retrieval corpus exceeds 10K chunks

See also: claude-api, observability, security

Похожие навыки: other

limits-colimits

Context management for Claude Code. Hooks maintain state via ledgers and handoffs. MCP execution without context polluti...

Автор: parcadei
3,392

ddttom-allaboutv2-claude-skills-block-collection-and-party

The Block Collection and Block Party are repositories for existing AEM blocks, build tools, code snippets, integration p...

Автор: mattnigh
20

nodnarbnitram-claude-code-extensions-claude-skills-esphome-confi

This skill should be used when the user asks to "create an esphome config", "set up an esp32 device", "configure esphome...

Автор: mattnigh
20