Senior Prompt Engineer

Expert-level guidance for LLM integration, prompt engineering, RAG systems, and multi-model orchestration.

When to Apply

• Designing or optimizing LLM prompts
• Implementing model routing/orchestration
• Managing context windows efficiently
• Building RAG pipelines
• Optimizing response quality and cost

1. Prompt Engineering Principles (CRITICAL)

Structure prompts clearly

<system>
You are a senior crypto market analyst. Analyze the provided market data 
and provide actionable insights.
</system>

<context>
{market_data}
{technical_indicators}
{news_summary}
</context>

<task>
Analyze {symbol} based on the provided context.
Focus on: trend direction, key levels, risk factors.
Format: structured markdown with sections.
</task>

Be specific about output format

• Define exact sections expected
• Specify language and tone
• Set length constraints
• Provide examples of ideal output

Use structured context injection

def build_prompt(context: dict, query: str) -> str:
    sections = []
    if context.get("prices"):
        sections.append(f"## Current Prices\n{format_prices(context['prices'])}")
    if context.get("technical"):
        sections.append(f"## Technical Analysis\n{format_ta(context['technical'])}")
    if context.get("news"):
        sections.append(f"## Recent News\n{format_news(context['news'])}")
    
    return f"""You are analyzing crypto markets.

MARKET CONTEXT:
{chr(10).join(sections)}

USER QUERY: {query}

Provide analysis based ONLY on the data above. Do not invent data."""

2. Context Window Management (CRITICAL)

Prioritize context by relevance

1. Most recent market data (highest priority)
2. Technical indicators relevant to query
3. News within last 24h
4. Historical patterns (compressed)

Compress context efficiently

def compress_context(full_context: dict, max_tokens: int = 4000) -> str:
    """Compress context to fit token budget, prioritizing recency and relevance."""
    sections = []
    budget = max_tokens
    
    # Always include: current prices (small, critical)
    prices = format_prices_compact(full_context["prices"])
    sections.append(prices)
    budget -= estimate_tokens(prices)
    
    # Include TA if budget allows
    if budget > 500:
        ta = format_ta_compact(full_context["technical"])
        sections.append(ta)
        budget -= estimate_tokens(ta)
    
    # Include news headlines only if budget allows
    if budget > 300:
        news = format_news_headlines(full_context["news"][:5])
        sections.append(news)
    
    return "\n\n".join(sections)

Token estimation

• English: ~4 chars per token
• Numbers/symbols: ~2-3 chars per token
• JSON: more tokens than plain text
• Always leave 20% buffer for response

3. Multi-Model Orchestration (HIGH)

Model selection by task

MODEL_ROUTING = {
    "smalltalk": "fast_model",        # Quick, cheap
    "price_query": "fast_model",       # Simple lookup
    "technical_analysis": "analytical_model",  # Deep reasoning
    "best_setups": "analytical_model",  # Complex multi-step
    "deep_research": "deep_model",     # Maximum quality
    "orchestration": "fast_model",     # Query classification
}

Orchestration pattern

1. Classify query with fast model (< 1s)
2. Route to appropriate model and context level
3. Assemble context based on query type
4. Generate response with selected model
5. Post-process for formatting and validation

Fallback chain

async def generate_with_fallback(prompt, models):
    for model in models:
        try:
            return await call_llm(model, prompt)
        except (RateLimitError, TimeoutError, ModelUnavailableError):
            logger.warning(f"Model {model} failed, trying next")
    raise AllModelsFailedError()

4. Response Quality (HIGH)

Prevent hallucination

• Include "Based ONLY on the provided data" in prompts
• Require source references for claims
• Post-validate numbers against context data
• Use temperature 0.3-0.5 for factual analysis

Streaming optimization

• Start streaming immediately on first token
• Yield partial results as they arrive
• Don't buffer entire response before sending
• Handle mid-stream errors gracefully

Output validation

def validate_analysis(response: str, context: dict) -> bool:
    """Check response doesn't contain hallucinated data."""
    # Check mentioned prices are close to actual
    for symbol, price in extract_prices(response):
        actual = context["prices"].get(symbol)
        if actual and abs(price - actual) / actual > 0.05:
            logger.warning(f"Price mismatch: {symbol} response={price} actual={actual}")
            return False
    return True

5. Cost Optimization (MEDIUM)

Cache common responses

• Cache smalltalk responses for 1 hour
• Cache news summaries for 4 hours
• Cache TA reports for cache-TTL period
• Return cached with "cached" flag

Reduce token usage

• Compress context (remove duplicates, use abbreviations)
• Use shorter system prompts for simple tasks
• Batch similar queries
• Skip context assembly for simple factual queries

Monitor costs

• Track tokens per request type
• Alert on cost spikes
• Optimize expensive query types first
• A/B test prompt efficiency

6. RAG Patterns

Vector search for news

async def get_relevant_news(query: str, top_k: int = 5):
    embedding = await get_embedding(query)
    results = qdrant_client.search(
        collection_name="telegram_news_v2",
        query_vector=embedding,
        limit=top_k,
        score_threshold=0.7
    )
    return [r.payload for r in results]

Context enrichment pipeline

1. Query → embedding
2. Vector search for relevant docs
3. Score and filter by relevance
4. Combine with structured data
5. Compress to fit context window