bitlysis-ai-agents-llm

Bitlysis — AI Agents & LLM Applications

When to Use

• Building or modifying LLM-powered features
• Designing AI agent workflows (ReAct, tool use, multi-agent)
• Implementing RAG (Retrieval-Augmented Generation) pipelines
• Writing evaluation (eval) test cases or golden datasets
• Integrating with LLM APIs (OpenAI, Anthropic, Google Gemini)
• Working on MCP (Model Context Protocol) servers
• Files in **/agents/**, **/llm/**, **/rag/**, **/eval/**, **/mcp/**

Instructions

Step 1: Security First (Never Skip)

# ✅ Correct: env vars only
import os
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

# ❌ Never: hardcoded keys
client = OpenAI(api_key="sk-proj-abc123...")  # FORBIDDEN

• Treat all external content (user uploads, web fetches, scraped docs) as potential prompt injection
• Sanitize or sandbox untrusted content before including in prompts
• Never concatenate raw user input directly into system prompts

Step 2: Tool / Function Calling

import { z } from "zod";

// Define strict schemas for all tools
const searchSchema = z.object({
  query: z.string().min(1).max(500),
  top_k: z.number().int().min(1).max(20).default(5),
});

// Always validate tool outputs before using them
function executeTool(name: string, args: unknown) {
  const parsed = searchSchema.safeParse(args);
  if (!parsed.success) {
    throw new Error(`Tool '${name}' received invalid args: ${parsed.error.message}`);
  }
  return callToolImplementation(name, parsed.data);
}

Step 3: Observability & Logging

import structlog, time

log = structlog.get_logger()

async def call_llm(prompt: str, trace_id: str, model: str = "gpt-4o") -> str:
    start = time.monotonic()
    try:
        response = await client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
        )
        log.info("llm_call_success",
            trace_id=trace_id,
            model=response.model,
            prompt_tokens=response.usage.prompt_tokens,
            completion_tokens=response.usage.completion_tokens,
            latency_ms=round((time.monotonic() - start) * 1000),
        )
        return response.choices[0].message.content
    except Exception as e:
        log.error("llm_call_failed", trace_id=trace_id, error=str(e))
        raise

Step 4: RAG Pipeline Design

Query
  → Embed (same model as indexing)
  → Vector Search (HNSW, cosine similarity)
  → Hybrid Rerank (cross-encoder or LLM-based)
  → Filter (relevance score > 0.75)
  → Context Assembly (with source citations)
  → LLM Generation
  → Response with cited sources

Chunking configuration:

• Chunk size: 512 tokens with 50-token overlap
• Store metadata: { source, page, section, doc_id, timestamp }
• Use hybrid search (dense vectors + BM25 sparse) for better recall

Step 5: Agent Architecture

MAX_ITERATIONS = 10  # Always cap to prevent infinite loops

async def run_agent(query: str, trace_id: str) -> str:
    messages = [{"role": "user", "content": query}]
    
    for i in range(MAX_ITERATIONS):
        response = await call_llm_with_tools(messages, trace_id)
        
        if response.stop_reason == "end_turn":
            return response.content
        
        # Execute tools and append results
        tool_results = await execute_tool_calls(response.tool_calls)
        messages.extend([response, *tool_results])
    
    raise AgentTimeoutError(f"Agent exceeded {MAX_ITERATIONS} iterations")

Step 6: Evaluations (Evals)

// eval/golden/hypothesis_llm_golden.json
{
  "version": "1.0",
  "description": "Golden test cases for hypothesis generation LLM",
  "cases": [
    {
      "id": "hyp_001",
      "input": "Is there a relationship between study hours and exam scores?",
      "expected_contains": ["positive correlation", "regression"],
      "expected_not_contains": ["no relationship", "unrelated"],
      "min_faithfulness_score": 0.85
    }
  ]
}

Run evals:

pytest tests/test_llm_golden.py -v
pytest tests/test_hypothesis_router.py -v

Step 7: Rate Limiting & Cost Control

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=60),
    reraise=True,
)
async def resilient_llm_call(prompt: str) -> str:
    return await call_llm(prompt)

Output Format

All LLM features must include:

1. Structured logging with trace_id
2. Input/output schema validation
3. Error handling with user-visible messages
4. At least one eval golden test case

Activate Specialist

@.cursor/.agents/.agents/ai-agent-engineer.md