Skip to main content Skills Marketplace Discover and explore AI skills built by the community.
Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.
Copy promptShow prompt details A direct command skips the review prompt. Inspect the source before running it.
npx skills add https://github.com/BEKO2210/Firstbrain --skill llm-app-patternsThe command stays on one line. Scroll horizontally to inspect it before copying.
Prefer a local copy? Download the files currently available to SkillsMP.
Download Zip Downloading... More from this repository
Related occupations SOC
Based on SOC occupation classification
name llm-app-patterns description Production-ready patterns for building LLM applications, inspired by [Dify](https://github.com/langgenius/dify) and industry best practices. type skill created 2026-02-27T00:00:00.000Z domain ai-ml category llm-agents risk unknown source community tags ["skill","ai-ml","llm-agents","llm","app"]
🤖 LLM Application Patterns
Production-ready patterns for building LLM applications, inspired by Dify and industry best practices.
When to Use This Skill
Use this skill when:
Designing LLM-powered applications
Implementing RAG (Retrieval-Augmented Generation)
Building AI agents with tools
Setting up LLMOps monitoring
Choosing between agent architectures
1. RAG Pipeline Architecture
Overview
RAG (Retrieval-Augmented Generation) grounds LLM responses in your data.
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Ingest │────▶│ Retrieve │────▶│ Generate │
│ Documents │ │ Context │ │ Response │
└─────────────┘ └─────────────┘ └─────────────┘
│ │ │
▼ ▼ ▼
┌─────────┐ ┌───────────┐ ┌───────────┐
│ Chunking│ │ Vector │ │ LLM │
│Embedding│ │ Search │ │ + Context│
└─────────┘ └───────────┘ └───────────┘
1.1 Document Ingestion
class ChunkingStrategy :
FIXED_SIZE = "fixed_size"
SEMANTIC = "semantic"
RECURSIVE = "recursive"
DOCUMENT_AWARE = "document_aware"
CHUNK_CONFIG = {
"chunk_size" : 512 ,
"chunk_overlap" : 50 ,
: [ , , , ],
}
"separators"
"\n\n"
"\n"
". "
" "
1.2 Embedding & Storage
VECTOR_DB_OPTIONS = {
"pinecone" : {
"use_case" : "Production, managed service" ,
"scale" : "Billions of vectors" ,
"features" : ["Hybrid search" , "Metadata filtering" ]
},
"weaviate" : {
"use_case" : "Self-hosted, multi-modal" ,
"scale" : "Millions of vectors" ,
"features" : ["GraphQL API" , "Modules" ]
},
"chromadb" : {
"use_case" : "Development, prototyping" ,
"scale" : "Thousands of vectors" ,
"features" : ["Simple API" , "In-memory option" ]
},
"pgvector" : {
"use_case" : "Existing Postgres infrastructure" ,
"scale" : "Millions of vectors" ,
"features" : ["SQL integration" , "ACID compliance" ]
}
}
EMBEDDING_MODELS = {
"openai/text-embedding-3-small" : {
"dimensions" : 1536 ,
"cost" : "$0.02/1M tokens" ,
"quality" : "Good for most use cases"
},
"openai/text-embedding-3-large" : {
"dimensions" : 3072 ,
"cost" : "$0.13/1M tokens" ,
"quality" : "Best for complex queries"
},
"local/bge-large" : {
"dimensions" : 1024 ,
"cost" : "Free (compute only)" ,
"quality" : "Comparable to OpenAI small"
}
}
1.3 Retrieval Strategies
def semantic_search (query: str , top_k: int = 5 ):
query_embedding = embed(query)
results = vector_db.similarity_search(
query_embedding,
top_k=top_k
)
return results
def hybrid_search (query: str , top_k: int = 5 , alpha: float = 0.5 ):
"""
alpha=1.0: Pure semantic
alpha=0.0: Pure keyword (BM25)
alpha=0.5: Balanced
"""
semantic_results = vector_db.similarity_search(query)
keyword_results = bm25_search(query)
return rrf_merge(semantic_results, keyword_results, alpha)
def multi_query_retrieval (query: str ):
"""Generate multiple query variations for better recall"""
queries = llm.generate_query_variations(query, n=3 )
all_results = []
for q in queries:
all_results.extend(semantic_search(q))
return deduplicate(all_results)
def compressed_retrieval (query: str ):
"""Retrieve then compress to relevant parts only"""
docs = semantic_search(query, top_k=10 )
compressed = llm.extract_relevant_parts(docs, query)
return compressed
1.4 Generation with Context RAG_PROMPT_TEMPLATE = """
Answer the user's question based ONLY on the following context.
If the context doesn't contain enough information, say "I don't have enough information to answer that."
Context:
{context}
Question: {question}
Answer:"""
def generate_with_rag (question: str ):
context_docs = hybrid_search(question, top_k=5 )
context = "\n\n" .join([doc.content for doc in context_docs])
prompt = RAG_PROMPT_TEMPLATE.format (
context=context,
question=question
)
response = llm.generate(prompt)
return {
"answer" : response,
"sources" : [doc.metadata for doc in context_docs]
}
2. Agent Architectures
2.1 ReAct Pattern (Reasoning + Acting) Thought: I need to search for information about X
Action: search("X")
Observation: [search results]
Thought: Based on the results, I should...
Action: calculate(...)
Observation: [calculation result]
Thought: I now have enough information
Action: final_answer("The answer is...")
REACT_PROMPT = """
You are an AI assistant that can use tools to answer questions.
Available tools:
{tools_description}
Use this format:
Thought: [your reasoning about what to do next]
Action: [tool_name(arguments)]
Observation: [tool result - this will be filled in]
... (repeat Thought/Action/Observation as needed)
Thought: I have enough information to answer
Final Answer: [your final response]
Question: {question}
"""
class ReActAgent :
def __init__ (self, tools: list , llm ):
self .tools = {t.name: t for t in tools}
self .llm = llm
self .max_iterations = 10
def run (self, question: str ) -> str :
prompt = REACT_PROMPT.format (
tools_description=self ._format_tools(),
question=question
)
for _ in range (self .max_iterations):
response = self .llm.generate(prompt)
if "Final Answer:" in response:
return self ._extract_final_answer(response)
action = self ._parse_action(response)
observation = self ._execute_tool(action)
prompt += f"\nObservation: {observation} \n"
return "Max iterations reached"
2.2 Function Calling Pattern
TOOLS = [
{
"name" : "search_web" ,
"description" : "Search the web for current information" ,
"parameters" : {
"type" : "object" ,
"properties" : {
"query" : {
"type" : "string" ,
"description" : "Search query"
}
},
"required" : ["query" ]
}
},
{
"name" : "calculate" ,
"description" : "Perform mathematical calculations" ,
"parameters" : {
"type" : "object" ,
"properties" : {
"expression" : {
"type" : "string" ,
"description" : "Math expression to evaluate"
}
},
"required" : ["expression" ]
}
}
]
class FunctionCallingAgent :
def run (self, question: str ) -> str :
messages = [{"role" : "user" , "content" : question}]
while True :
response = self .llm.chat(
messages=messages,
tools=TOOLS,
tool_choice="auto"
)
if response.tool_calls:
for tool_call in response.tool_calls:
result = self ._execute_tool(
tool_call.name,
tool_call.arguments
)
messages.append({
"role" : "tool" ,
"tool_call_id" : tool_call.id ,
"content" : str (result)
})
else :
return response.content
2.3 Plan-and-Execute Pattern class PlanAndExecuteAgent :
"""
1. Create a plan (list of steps)
2. Execute each step
3. Replan if needed
"""
def run (self, task: str ) -> str :
plan = self .planner.create_plan(task)
results = []
for step in plan:
result = self .executor.execute(step, context=results)
results.append(result)
if self ._needs_replan(task, results):
new_plan = self .planner.replan(
task,
completed=results,
remaining=plan[len (results):]
)
plan = new_plan
return self .synthesizer.summarize(task, results)
2.4 Multi-Agent Collaboration class AgentTeam :
"""
Specialized agents collaborating on complex tasks
"""
def __init__ (self ):
self .agents = {
"researcher" : ResearchAgent(),
"analyst" : AnalystAgent(),
"writer" : WriterAgent(),
"critic" : CriticAgent()
}
self .coordinator = CoordinatorAgent()
def solve (self, task: str ) -> str :
assignments = self .coordinator.decompose(task)
results = {}
for assignment in assignments:
agent = self .agents[assignment.agent]
result = agent.execute(
assignment.subtask,
context=results
)
results[assignment.id ] = result
critique = self .agents["critic" ].review(results)
if critique.needs_revision:
return self .solve_with_feedback(task, results, critique)
return self .coordinator.synthesize(results)
3. Prompt IDE Patterns
3.1 Prompt Templates with Variables class PromptTemplate :
def __init__ (self, template: str , variables: list [str ] ):
self .template = template
self .variables = variables
def format (self, **kwargs ) -> str :
missing = set (self .variables) - set (kwargs.keys())
if missing:
raise ValueError(f"Missing variables: {missing} " )
return self .template.format (**kwargs)
def with_examples (self, examples: list [dict ] ) -> str :
"""Add few-shot examples"""
example_text = "\n\n" .join([
f"Input: {ex['input' ]} \nOutput: {ex['output' ]} "
for ex in examples
])
return f"{example_text} \n\n{self.template} "
summarizer = PromptTemplate(
template="Summarize the following text in {style} style:\n\n{text}" ,
variables=["style" , "text" ]
)
prompt = summarizer.format (
style="professional" ,
text="Long article content..."
)
3.2 Prompt Versioning & A/B Testing class PromptRegistry :
def __init__ (self, db ):
self .db = db
def register (self, name: str , template: str , version: str ):
"""Store prompt with version"""
self .db.save({
"name" : name,
"template" : template,
"version" : version,
"created_at" : datetime.now(),
"metrics" : {}
})
def get (self, name: str , version: str = "latest" ) -> str :
"""Retrieve specific version"""
return self .db.get(name, version)
def ab_test (self, name: str , user_id: str ) -> str :
"""Return variant based on user bucket"""
variants = self .db.get_all_versions(name)
bucket = hash (user_id) % len (variants)
return variants[bucket]
def record_outcome (self, prompt_id: str , outcome: dict ):
"""Track prompt performance"""
self .db.update_metrics(prompt_id, outcome)
3.3 Prompt Chaining class PromptChain :
"""
Chain prompts together, passing output as input to next
"""
def __init__ (self, steps: list [dict ] ):
self .steps = steps
def run (self, initial_input: str ) -> dict :
context = {"input" : initial_input}
results = []
for step in self .steps:
prompt = step["prompt" ].format (**context)
output = llm.generate(prompt)
if step.get("parser" ):
output = step"parser"
context[step["output_key" ]] = output
results.append({
"step" : step["name" ],
"output" : output
})
return {
"final_output" : context[self .steps[-1 ]["output_key" ]],
"intermediate_results" : results
}
chain = PromptChain([
{
"name" : "research" ,
"prompt" : "Research the topic: {input}" ,
"output_key" : "research"
},
{
"name" : "analyze" ,
"prompt" : "Analyze these findings:\n{research}" ,
"output_key" : "analysis"
},
{
"name" : "summarize" ,
"prompt" : "Summarize this analysis in 3 bullet points:\n{analysis}" ,
"output_key" : "summary"
}
])
4. LLMOps & Observability
4.1 Metrics to Track LLM_METRICS = {
"latency_p50" : "50th percentile response time" ,
"latency_p99" : "99th percentile response time" ,
"tokens_per_second" : "Generation speed" ,
"user_satisfaction" : "Thumbs up/down ratio" ,
"task_completion" : "% tasks completed successfully" ,
"hallucination_rate" : "% responses with factual errors" ,
"cost_per_request" : "Average $ per API call" ,
"tokens_per_request" : "Average tokens used" ,
"cache_hit_rate" : "% requests served from cache" ,
"error_rate" : "% failed requests" ,
"timeout_rate" : "% requests that timed out" ,
"retry_rate" : "% requests needing retry"
}
4.2 Logging & Tracing import logging
from opentelemetry import trace
tracer = trace.get_tracer(__name__)
class LLMLogger :
def log_request (self, request_id: str , data: dict ):
"""Log LLM request for debugging and analysis"""
log_entry = {
"request_id" : request_id,
"timestamp" : datetime.now().isoformat(),
"model" : data["model" ],
"prompt" : data["prompt" ][:500 ],
"prompt_tokens" : data["prompt_tokens" ],
"temperature" : data.get("temperature" , 1.0 ),
"user_id" : data.get("user_id" ),
}
logging.info(f"LLM_REQUEST: {json.dumps(log_entry)} " )
def log_response (self, request_id: str , data: dict ):
"""Log LLM response"""
log_entry = {
"request_id" : request_id,
"completion_tokens" : data["completion_tokens" ],
"total_tokens" : data["total_tokens" ],
"latency_ms" : data["latency_ms" ],
"finish_reason" : data["finish_reason" ],
"cost_usd" : self ._calculate_cost(data),
}
logging.info(f"LLM_RESPONSE: {json.dumps(log_entry)} " )
@tracer.start_as_current_span("llm_call" )
def call_llm (prompt: str ) -> str :
span = trace.get_current_span()
span.set_attribute("prompt.length" , len (prompt))
response = llm.generate(prompt)
span.set_attribute("response.length" , len (response))
span.set_attribute("tokens.total" , response.usage.total_tokens)
return response.content
4.3 Evaluation Framework class LLMEvaluator :
"""
Evaluate LLM outputs for quality
"""
def evaluate_response (self,
question: str ,
response: str ,
ground_truth: str = None ) -> dict :
scores = {}
scores["relevance" ] = self ._score_relevance(question, response)
scores["coherence" ] = self ._score_coherence(response)
scores["groundedness" ] = self ._score_groundedness(response)
if ground_truth:
scores["accuracy" ] = self ._score_accuracy(response, ground_truth)
scores["safety" ] = self ._score_safety(response)
return scores
def run_benchmark (self, test_cases: list [dict ] ) -> dict :
"""Run evaluation on test set"""
results = []
for case in test_cases:
response = llm.generate(case ["prompt" ])
scores = self .evaluate_response(
question=case ["prompt" ],
response=response,
ground_truth=case .get("expected" )
)
results.append(scores)
return self ._aggregate_scores(results)
5. Production Patterns
5.1 Caching Strategy import hashlib
from functools import lru_cache
class LLMCache :
def __init__ (self, redis_client, ttl_seconds=3600 ):
self .redis = redis_client
self .ttl = ttl_seconds
def _cache_key (self, prompt: str , model: str , **kwargs ) -> str :
"""Generate deterministic cache key"""
content = f"{model} :{prompt} :{json.dumps(kwargs, sort_keys=True )} "
return hashlib.sha256(content.encode()).hexdigest()
def get_or_generate (self, prompt: str , model: str , **kwargs ) -> str :
key = self ._cache_key(prompt, model, **kwargs)
cached = self .redis.get(key)
if cached:
return cached.decode()
response = llm.generate(prompt, model=model, **kwargs)
if kwargs.get("temperature" , 1.0 ) == 0 :
self .redis.setex(key, self .ttl, response)
return response
5.2 Rate Limiting & Retry import time
from tenacity import retry, wait_exponential, stop_after_attempt
class RateLimiter :
def __init__ (self, requests_per_minute: int ):
self .rpm = requests_per_minute
self .timestamps = []
def acquire (self ):
"""Wait if rate limit would be exceeded"""
now = time.time()
self .timestamps = [t for t in self .timestamps if now - t < 60 ]
if len (self .timestamps) >= self .rpm:
sleep_time = 60 - (now - self .timestamps[0 ])
time.sleep(sleep_time)
self .timestamps.append(time.time())
@retry(
wait=wait_exponential(multiplier=1 , min =4 , max =60 ),
stop=stop_after_attempt(5 )
)
def call_llm_with_retry (prompt: str ) -> str :
try :
return llm.generate(prompt)
except RateLimitError:
raise
except APIError as e:
if e.status_code >= 500 :
raise
raise
5.3 Fallback Strategy class LLMWithFallback :
def __init__ (self, primary: str , fallbacks: list [str ] ):
self .primary = primary
self .fallbacks = fallbacks
def generate (self, prompt: str , **kwargs ) -> str :
models = [self .primary] + self .fallbacks
for model in models:
try :
return llm.generate(prompt, model=model, **kwargs)
except (RateLimitError, APIError) as e:
logging.warning(f"Model {model} failed: {e} " )
continue
raise AllModelsFailedError("All models exhausted" )
llm_client = LLMWithFallback(
primary="gpt-4-turbo" ,
fallbacks=["gpt-3.5-turbo" , "claude-3-sonnet" ]
)
Architecture Decision Matrix Pattern Use When Complexity Cost Simple RAG FAQ, docs search Low Low Hybrid RAG Mixed queries Medium Medium ReAct Agent Multi-step tasks Medium Medium Function Calling Structured tools Low Low Plan-Execute Complex tasks High High Multi-Agent Research tasks Very High Very High
Resources
Connections
Domain: [[KI & Machine Learning]]
Kategorie: [[LLM & KI-Agenten]]
Navigation: [[Skills Uebersicht]], [[Home]]