Skip to main content ホーム クリエイター diegosouzapw awesome-omni-skill llm-app-patterns
llm-app-patterns Production-ready patterns for building LLM applications. Covers RAG pipelines, agent architectures, prompt IDEs, and LLMOps monitoring. Use when designing AI applications, implementing RAG, building agents, or setting up LLM observability.
インストールへ移動 Skills Marketplace コミュニティが作成したAIスキルを発見・探索
Codex または Claude でインストール この Prompt をコピーして Codex、Claude、または他のアシスタントに貼り付けると、Skill ページを確認してインストールできます。
直接コマンドでは確認用 Prompt が省略されます。実行前にソースを確認してください。
npx skills add https://github.com/diegosouzapw/awesome-omni-skill --skill llm-app-patternsコマンドは1行のまま表示されます。コピー前に横へスクロールして全体を確認してください。
ローカルで確認しますか?SkillsMP が現在取得できるファイルをダウンロードできます。
Zipをダウンロード ダウンロード中... このリポジトリの他の Skills Token-efficient tracking for AI orchestration. CLI-first for status updates (~50 tokens), agent fallback for complex ops (~1KB). Use when: updating task status, querying blockers, creating progress files, validating phases.
AshAi extension guidelines for integrating AI capabilities with Ash Framework. Use when implementing vectorization/embeddings, exposing Ash actions as LLM tools, creating prompt-backed actions, or setting up MCP servers. Covers semantic search, LangChain integration, and structured outputs.
This skill should be used when solving hard questions, complex architectural problems, or debugging issues that benefit from GPT-5 Pro or GPT-5.1 thinking models with large file context. Use when standard Claude analysis needs deeper reasoning or extended context windows.
name llm-app-patterns description Production-ready patterns for building LLM applications. Covers RAG pipelines, agent architectures, prompt IDEs, and LLMOps monitoring. Use when designing AI applications, implementing RAG, building agents, or setting up LLM observability.
🤖 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 =
RECURSIVE =
DOCUMENT_AWARE =
CHUNK_CONFIG = {
: ,
: ,
: [ , , , ],
}
"semantic"
"recursive"
"document_aware"
"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" ](output)
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