Skip to main content 홈 크리에이터 beko2210 firstbrain llm-app-patterns
llm-app-patterns Production-ready patterns for building LLM applications, inspired by [Dify](https://github.com/langgenius/dify) and industry best practices.
설치로 이동 Skills Marketplace 커뮤니티가 만든 AI 스킬을 발견하고 탐색하세요.
Codex 또는 Claude로 설치 이 Prompt를 복사해 Codex, Claude 또는 다른 어시스턴트에 붙여 넣으면 Skill 페이지를 검토하고 설치를 진행할 수 있습니다.
직접 명령은 검토 Prompt를 거치지 않습니다. 실행하기 전에 소스를 확인하세요.
npx skills add https://github.com/BEKO2210/Firstbrain --skill llm-app-patterns명령은 한 줄로 유지됩니다. 복사하기 전에 가로로 스크롤해 전체 내용을 확인하세요.
로컬 사본을 원하시나요? SkillsMP에서 현재 제공할 수 있는 파일을 다운로드하세요.
Zip 다운로드 다운로드 중... 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]]