# RAG Agent

> Pipeline memory and continuous learning system using RAG (Retrieval-Augmented Generation). Stores all pipeline artifacts (research, ADRs, solutions, results) in vector database for semantic retrieval. Enables pipeline to learn from history, avoid re-researching, and improve recommendations over time. Use this agent continuously throughout pipeline execution to build institutional knowledge.

- Skill: `majiayu000/rag-agent` (Agent Skill, multi-file: 2 files)
- Install (CLI): `npx skillmds@latest add majiayu000/rag-agent`
- Raw SKILL.md: https://api.skillmd.com/api/skills/majiayu000/rag-agent/raw
- Safety review: pending (external: skill-scanner PASS, skillspector PASS)
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: DevOps & Infra
- Author: majiayu000 (https://skillmd.com/u/majiayu000)
- Updated: 2026-09-09
- Page: https://skillmd.com/skills/majiayu000/rag-agent

---


# RAG Agent - Pipeline Memory & Continuous Learning

## Role

The RAG (Retrieval-Augmented Generation) Agent is the **institutional memory** of the pipeline. It captures, stores, and retrieves all pipeline artifacts using semantic search, enabling the pipeline to learn from past experiences and continuously improve.

---

## Core Responsibilities

### 1. **Capture Everything**

Store all pipeline artifacts in vector database:

**What Gets Stored:**
- ✅ Research reports (topics, findings, recommendations)
- ✅ ADRs (architectural decisions, reasoning)
- ✅ Developer solutions (code, tests, approach)
- ✅ Validation results (pass/fail, issues found)
- ✅ Arbitration scores (what won, what lost, why)
- ✅ Integration results (deployment success/failure)
- ✅ Testing results (quality gates, performance)
- ✅ Error logs (what went wrong, how fixed)
- ✅ User feedback (satisfaction, issues reported)

**Storage Format:**
```python
{
  "artifact_id": "research-card-123-oauth",
  "artifact_type": "research_report",
  "card_id": "card-123",
  "task_title": "Add OAuth authentication",
  "content": "Research Report: ... authlib recommended...",
  "metadata": {
    "technologies": ["authlib", "Flask-Login", "OAuth2"],
    "recommendations": ["Use authlib", "Encrypt tokens"],
    "timestamp": "2025-10-22T14:00:00Z",
    "priority": "high",
    "complexity": "complex"
  },
  "embeddings": [0.234, -0.567, 0.891, ...]  # Vector for semantic search
}
```

### 2. **Semantic Search & Retrieval**

Enable agents to find relevant past experiences:

**Query Types:**
```python
# Research Agent asks:
"Show me research about OAuth libraries we've done before"
→ Returns: Previous authlib vs python-social-auth research

# Architecture Agent asks:
"What did we decide for similar database tasks?"
→ Returns: Past ADRs for customer database, user database

# Developer Agent asks:
"Show me similar authentication implementations"
→ Returns: Past OAuth solutions with high scores

# Validation Agent asks:
"What security issues appeared in similar code?"
→ Returns: Past validation blockers for auth tasks
```

**Semantic Search Examples:**
- Query: "WebSocket performance issues"
  - Finds: Past research on WebSocket scaling, similar real-time features
- Query: "PostgreSQL vs SQLite decision"
  - Finds: Database comparison ADRs, production deployment results
- Query: "High-scoring CRUD implementations"
  - Finds: Top developer solutions for CRUD tasks

### 3. **Learn & Improve**

Extract patterns and insights from history:

**Learning Patterns:**
```python
# Pattern 1: What Works
"When we used authlib for OAuth, arbitration scores averaged 96/100"
"When we used SQLite for production, integration failed 80% of time"
→ Recommendation: Prefer authlib, avoid SQLite production

# Pattern 2: Common Issues
"SQL injection found in 60% of solutions without ORM"
"Tests failed when coverage < 85%"
→ Recommendation: Require ORM, enforce 85%+ coverage

# Pattern 3: Technology Success Rates
"Flask tasks: 95% success rate, avg score 94/100"
"Django tasks: 85% success rate, avg score 88/100"
→ Recommendation: Prefer Flask for simple APIs

# Pattern 4: User Satisfaction
"Tasks with research stage: 4.8/5 user rating"
"Tasks without research: 3.2/5 user rating"
→ Recommendation: Run research for all complex tasks
```

### 4. **Assist All Agents**

Provide contextual knowledge to every pipeline agent:

**Research Agent:**
- "Did we research this technology before?"
- "What recommendations did we make last time?"
- "What security issues did we find?"

**Architecture Agent:**
- "What did we decide for similar problems?"
- "What patterns worked well?"
- "What should we avoid?"

**Developer Agents:**
- "Show me similar implementations"
- "What test strategies worked?"
- "What libraries did we use?"

**Validation Agent:**
- "What issues appeared in similar code?"
- "What test coverage was sufficient?"
- "What blockers occurred before?"

**Arbitration Agent:**
- "What scored well on similar tasks?"
- "What patterns correlate with high scores?"
- "What approaches failed?"

---

## When to Use This Agent

### ✅ Use RAG Agent:

**ALWAYS** - It runs continuously throughout pipeline:

1. **Pipeline Start** - Query for similar past tasks
2. **Research Stage** - Check if topic researched before
3. **Architecture Stage** - Retrieve similar ADRs
4. **Development Stage** - Find similar implementations
5. **Validation Stage** - Check common issues
6. **Arbitration Stage** - Compare to past scores
7. **Testing Stage** - Reference past test results
8. **Pipeline End** - Store all artifacts for future

**Every single pipeline execution** uses and updates RAG!

---

## RAG Agent Operations

### Operation 1: Store Artifact

**When:** After each pipeline stage completes

```python
rag_agent.store_artifact(
    artifact_type="research_report",
    card_id="card-123",
    task_title="Add OAuth authentication",
    content=research_report_text,
    metadata={
        "technologies": ["authlib", "OAuth2", "Flask"],
        "recommendations": ["Use authlib", "Encrypt tokens"],
        "confidence": "HIGH"
    }
)
```

**What Happens:**
1. Generate text embedding using sentence-transformers
2. Extract keywords and entities
3. Store in ChromaDB with metadata
4. Update knowledge graph connections
5. Index for fast retrieval

### Operation 2: Query Similar

**When:** Before each stage to get context

```python
# Research Agent queries before researching
similar = rag_agent.query_similar(
    query_text="OAuth library comparison",
    artifact_types=["research_report", "adr"],
    top_k=5,
    filters={"technologies": ["OAuth", "authentication"]}
)

# Returns:
[
    {
        "artifact_id": "research-card-098-oauth",
        "similarity": 0.94,
        "task_title": "Add Google OAuth login",
        "content": "Research found authlib is best...",
        "metadata": {...},
        "date": "2025-09-15"
    },
    ...
]
```

**What Happens:**
1. Generate query embedding
2. Vector similarity search in ChromaDB
3. Apply metadata filters
4. Rank by relevance + recency
5. Return top matches

### Operation 3: Extract Patterns

**When:** Periodically (daily/weekly) or on-demand

```python
# Extract learning patterns
patterns = rag_agent.extract_patterns(
    pattern_type="technology_success_rates",
    time_window_days=90
)

# Returns:
{
    "authlib": {
        "tasks_count": 12,
        "avg_score": 96.3,
        "success_rate": 0.92,
        "recommendation": "HIGHLY_RECOMMENDED"
    },
    "python-social-auth": {
        "tasks_count": 3,
        "avg_score": 78.5,
        "success_rate": 0.67,
        "recommendation": "CONSIDER_ALTERNATIVES"
    }
}
```

### Operation 4: Get Recommendations

**When:** Any agent needs guidance

```python
# Get RAG-informed recommendations
recommendations = rag_agent.get_recommendations(
    task_description="Add real-time chat feature",
    context={
        "technologies_mentioned": ["WebSocket", "chat"],
        "priority": "high",
        "complexity": "complex"
    }
)

# Returns:
{
    "based_on_history": [
        "Used Flask-SocketIO in 4 past chat features (avg score: 94/100)",
        "Redis worked well for message queue (3 tasks, 100% success)",
        "Common issue: WebSocket scaling at >1000 users (found in 2 tasks)"
    ],
    "recommendations": [
        "Consider Flask-SocketIO (proven success)",
        "Plan for Redis message queue",
        "Research horizontal scaling early"
    ],
    "avoid": [
        "Long polling (failed performance tests in task card-087)",
        "In-memory storage (lost messages on restart in card-104)"
    ]
}
```

---

## Vector Database Schema

### ChromaDB Collections

**Collection 1: research_reports**
```python
{
    "id": "research-card-123",
    "embedding": [vector],
    "metadata": {
        "card_id": "card-123",
        "task_title": "Add OAuth authentication",
        "technologies": ["authlib", "OAuth2"],
        "recommendations": ["Use authlib"],
        "timestamp": "2025-10-22T14:00:00Z",
        "priority": "high",
        "user_prompts_count": 3
    },
    "document": "Full research report text..."
}
```

**Collection 2: architecture_decisions**
```python
{
    "id": "adr-card-123",
    "embedding": [vector],
    "metadata": {
        "card_id": "card-123",
        "adr_number": "003",
        "task_title": "Add OAuth authentication",
        "technologies": ["authlib", "Flask-Login"],
        "decision": "Use authlib for OAuth",
        "timestamp": "2025-10-22T14:05:00Z"
    },
    "document": "Full ADR text..."
}
```

**Collection 3: developer_solutions**
```python
{
    "id": "solution-card-123-developer-b",
    "embedding": [vector],
    "metadata": {
        "card_id": "card-123",
        "developer": "developer-b",
        "task_title": "Add OAuth authentication",
        "approach": "comprehensive",
        "test_coverage": 92,
        "arbitration_score": 98,
        "winner": true,
        "technologies": ["authlib", "Flask-Login", "AES-256"]
    },
    "document": "Solution description and key code snippets..."
}
```

**Collection 4: issues_and_fixes**
```python
{
    "id": "issue-card-123-validation",
    "embedding": [vector],
    "metadata": {
        "card_id": "card-123",
        "stage": "validation",
        "issue_type": "security",
        "severity": "high",
        "resolved": true,
        "fix": "Added token encryption"
    },
    "document": "Issue: Tokens stored unencrypted. Fix: Implemented AES-256..."
}
```

---

## Learning & Improvement Examples

### Example 1: Avoid Repeated Research

**Scenario:** New task needs OAuth research

**Without RAG:**
```
Research Agent researches OAuth libraries again (2-3 minutes)
→ Finds authlib is best (again)
→ Researches token security (again)
```

**With RAG:**
```python
# Research Agent queries RAG first
past_research = rag.query_similar("OAuth library comparison")

# Finds:
"We researched this 2 weeks ago (card-123)
 - authlib recommended (4.3k stars, active)
 - python-social-auth less maintained
 - Token encryption required
 Confidence: HIGH, Recency: Recent"

# Research Agent decision:
if past_research.similarity > 0.90 and past_research.age_days < 30:
    # Use existing research!
    return past_research.content
else:
    # Re-research (info might be outdated)
    conduct_new_research()
```

**Result:** Saves 2-3 minutes, ensures consistency

### Example 2: Learn From Mistakes

**Scenario:** Task requires production database

**Past Experience (stored in RAG):**
```
Task card-087: "Customer database"
Decision: Used SQLite for production
Result: FAILED integration (concurrent write issues)
Lesson: SQLite not suitable for production
```

**New Task:** "Add user profile database"

**Architecture Agent queries RAG:**
```python
similar_tasks = rag.query_similar("database production deployment")

# Finds card-087 failure
# Extracts lesson: "Avoid SQLite for production"
```

**ADR Created:**
```markdown
## Database Decision

**Choice:** PostgreSQL

**Reasoning:**
Past experience (card-087) showed SQLite fails in production
due to concurrent write limitations. PostgreSQL recommended
based on successful deployments in card-091, card-102.

**Evidence from RAG:**
- SQLite: 0/3 production tasks succeeded
- PostgreSQL: 5/5 production tasks succeeded
```

**Result:** Learns from mistakes, avoids repeating errors

### Example 3: Improve Recommendations

**Scenario:** Multiple OAuth tasks over time

**RAG Learns:**
```python
# After 10 OAuth-related tasks:
Technology Success Rates:
  authlib: 10 tasks, 96 avg score, 90% success
  python-social-auth: 2 tasks, 78 avg score, 50% success

Common Patterns:
  - authlib tasks: 3.2 days avg implementation
  - python-social-auth tasks: 5.1 days avg implementation

Issues Found:
  - authlib: 2 minor issues (documentation clarity)
  - python-social-auth: 8 issues (maintenance, bugs)
```

**Research Agent on new OAuth task:**
```python
# Queries RAG for recommendations
rag_insights = rag.get_recommendations("OAuth implementation")

# RAG provides data-backed recommendation:
"""
Based on 10 past OAuth tasks:
- authlib: 96/100 avg score, 90% success rate
- STRONG RECOMMENDATION for authlib
- Evidence: Faster implementation, fewer issues
- Confidence: VERY HIGH (10 data points)
"""
```

**Result:** Recommendations improve with experience

---

## Communication Protocol Integration

### Receives Messages From:
- **All Agents** - Store artifact requests, query requests

### Sends Messages To:
- **All Agents** - Query results, recommendations, insights

### Message Types:

**Store Artifact:**
```python
messenger.send_data_update(
    to_agent="rag-agent",
    message_type="store_artifact",
    data={
        "artifact_type": "research_report",
        "card_id": "card-123",
        "content": research_report,
        "metadata": {...}
    }
)
```

**Query Similar:**
```python
messenger.send_request(
    to_agent="rag-agent",
    request_type="query_similar",
    requirements={
        "query_text": "OAuth library comparison",
        "artifact_types": ["research_report", "adr"],
        "top_k": 5
    }
)
```

**Response:**
```python
messenger.send_response(
    to_agent="research-agent",
    response_type="query_results",
    data={
        "results": [similar_artifacts],
        "count": 5,
        "max_similarity": 0.94
    }
)
```

---

## Success Criteria

### ✅ RAG Agent is Successful When:

1. **Complete Storage**
   - All artifacts from every pipeline run captured
   - No data loss
   - Proper embeddings generated

2. **Accurate Retrieval**
   - Query returns relevant results
   - Similarity scores > 0.8 for matches
   - Results ranked by relevance + recency

3. **Useful Learning**
   - Patterns extracted are actionable
   - Recommendations improve over time
   - Success rates increase

4. **Performance**
   - Query response < 100ms
   - Storage operation < 50ms
   - Scales to 10,000+ artifacts

5. **Improved Pipeline**
   - Fewer repeated errors
   - Better technology choices
   - Higher arbitration scores
   - Faster development (less research)

---

## Implementation Stack

**Vector Database:** ChromaDB
- Embedded (no server needed)
- Fast semantic search
- Metadata filtering
- Python native

**Embeddings:** sentence-transformers
- Model: all-MiniLM-L6-v2 (384 dimensions)
- Fast inference
- Good quality for code/text

**Storage Location:**
- `/tmp/rag_db/` - ChromaDB persistent storage
- Survives pipeline restarts
- Grows over time (institutional knowledge)

---

## RAG Agent Activation

**Always Active:**
```python
# Pipeline Start
rag_agent.initialize()

# Every Stage
rag_agent.query_before_stage(stage_name)
rag_agent.store_after_stage(stage_name, results)

# Pipeline End
rag_agent.finalize()
```

**No activation logic needed - RAG is always on!**

---

## Example RAG-Enhanced Pipeline Flow

```
Task: "Add payment processing"
    ↓
RAG: Query similar payment tasks
    → Found 3 past Stripe integrations
    → Found common issue: webhook security
    → Recommendation: Use stripe-python, validate webhooks
    ↓
Research Agent:
    - Checks RAG first
    - Found recent Stripe research (card-095, 10 days ago)
    - Uses existing research (saves 3 minutes)
    - Adds: webhook security (from RAG insight)
    ↓
Architecture Agent:
    - Queries RAG for payment ADRs
    - Found stripe-python used in 3 tasks (100% success)
    - Creates ADR citing past successes
    ↓
Developers:
    - Query RAG for Stripe implementations
    - Get code examples from past solutions
    - Avoid known issues (from RAG)
    ↓
Validation:
    - Queries RAG for payment validation issues
    - Finds: "Always test webhook signature validation"
    - Adds specific test
    ↓
Pipeline End:
    - Stores new payment implementation in RAG
    - Future payment tasks benefit from this experience
```

---

## Benefits

**Time Savings:**
- ✅ Avoid re-researching (2-3 min per task)
- ✅ Reuse past solutions (5-10 min per task)
- ✅ Learn from mistakes (hours saved debugging)

**Quality Improvement:**
- ✅ Data-backed decisions (not guesses)
- ✅ Avoid known issues (from past experience)
- ✅ Consistent technology choices

**Continuous Learning:**
- ✅ Pipeline gets smarter over time
- ✅ Success rates increase
- ✅ Better recommendations
- ✅ Institutional knowledge preserved

**Developer Experience:**
- ✅ Less repetitive work
- ✅ Best practices built-in
- ✅ Faster onboarding (examples available)

---

**Note:** RAG Agent is the **memory and learning system** that makes the entire pipeline continuously improve. Without RAG, the pipeline forgets everything after each task. With RAG, the pipeline builds expertise over time.