# Step 05-07: RAG / Self-Correction / Multi-Agent

> 这些步骤是进阶内容，包含核心概念和实现代码。

## Step 05: RAG - 知识检索

### 核心概念

RAG = Retrieval-Augmented Generation（检索增强生成）

```
┌─────────────────────────────────────────────────────┐
│                    RAG 流程                          │
├─────────────────────────────────────────────────────┤
│                                                      │
│  用户问题 ──▶ 编码为向量 ──▶ 向量数据库检索           │
│                   │                                  │
│                   ▼                                  │
│           找到最相关的文档                            │
│                   │                                  │
│                   ▼                                  │
│      把文档和问题一起发送给 LLM                       │
│                   │                                  │
│                   ▼                                  │
│              生成答案                                 │
│                                                      │
└─────────────────────────────────────────────────────┘
```

### 为什么需要 RAG？

1. LLM 的知识有截止日期
2. LLM 不知道你私有的数据
3. RAG 让 LLM 能"查阅"外部知识

### 关键组件

| 组件 | 作用 |
|------|------|
| Embedding Model | 把文本变成向量 |
| Vector Database | 存储和检索向量 |
| Retrieval | 找到最相关的文档 |
| Generation | 用检索结果生成答案 |

### 简化实现

```python
class SimpleRAG:
    """简化版 RAG 系统"""

    def __init__(self):
        # 文档存储
        self.documents = []
        # 向量存储（简化版，用关键词）
        self.vectors = {}

    def add_document(self, text: str, metadata: dict = None):
        """添加文档"""
        self.documents.append({
            "text": text,
            "metadata": metadata or {}
        })

    def retrieve(self, query: str, top_k: int = 3) -> list:
        """检索相关文档"""
        # 简化版：基于关键词匹配
        results = []
        for doc in self.documents:
            # 计算简单相关性分数
            score = sum(1 for word in query if word in doc["text"].lower())
            if score > 0:
                results.append((score, doc))
        # 排序并返回 top_k
        results.sort(key=lambda x: x[0], reverse=True)
        return [doc for _, doc in results[:top_k]]

    def generate(self, query: str, llm) -> str:
        """RAG 生成"""
        docs = self.retrieve(query)
        context = "\n".join([d["text"] for d in docs])

        prompt = f"""
根据以下上下文回答问题：

上下文：
{context}

问题：{query}

答案：
"""
        return llm.invoke(prompt)
```

---

## Step 06: Self-Correction - 自我修正

### 核心概念

Self-Correction = 让 Agent 能够自我发现并修复错误

```
┌─────────────────────────────────────────────────────┐
│                Self-Correction 流程                  │
├─────────────────────────────────────────────────────┤
│                                                      │
│  生成结果 ──▶ 验证 ──▶ 有问题？                       │
│                   │                                  │
│              ┌────┴────┐                            │
│              │         │                            │
│             是         否                            │
│              │         │                            │
│              ▼         ▼                            │
│        分析错误    返回结果                          │
│              │                                        │
│              ▼                                        │
│        生成修复方案                                   │
│              │                                        │
│              ▼                                        │
│        重新生成 ──▶ 再次验证                          │
│                                                      │
└─────────────────────────────────────────────────────┘
```

### 实现要点

```python
class SelfCorrectingAgent:
    """自我修正 Agent"""

    def __init__(self):
        self.max_retries = 3

    def try_generate(self, requirement: str) -> str:
        """带自我修正的生成"""
        for attempt in range(self.max_retries):
            # 1. 生成
            result = self.generate(requirement)

            # 2. 验证
            validation = self.validate(result)

            # 3. 检查是否通过
            if validation["passed"]:
                return result

            # 4. 分析错误
            error = validation["error"]
            print(f"尝试 {attempt + 1} 失败: {error}")

            # 5. 准备修复
            requirement = self.prepare_fix(requirement, error, result)

        return f"经过 {self.max_retries} 次尝试仍失败"
```

---

## Step 07: Multi-Agent - 多 Agent 协作

### 核心概念

Multi-Agent = 多个专门的 Agent 协同工作

```
┌─────────────────────────────────────────────────────┐
│                  Multi-Agent 架构                   │
├─────────────────────────────────────────────────────┤
│                                                      │
│                  ┌──────────────┐                    │
│                  │ Orchestrator │                    │
│                  │  (协调者)    │                    │
│                  └──────┬───────┘                    │
│                         │                            │
│     ┌───────────────────┼───────────────────┐         │
│     │                   │                   │         │
│     ▼                   ▼                   ▼         │
│ ┌──────────┐      ┌──────────┐      ┌──────────┐   │
│ │ Generator│      │ Validator│      │ Searcher │   │
│ │ (生成者) │      │ (验证者) │      │ (搜索者) │   │
│ └──────────┘      └──────────┘      └──────────┘   │
│                                                      │
└─────────────────────────────────────────────────────┘
```

### 协作模式

| 模式 | 说明 | 适用场景 |
|------|------|---------|
| 串行 | A → B → C 依次执行 | 步骤有依赖 |
| 并行 | A / B / C 同时执行 | 步骤独立 |
| 循环 | A → B → A → B 循环 | 需要反复验证 |

### 简化实现

```python
class MultiAgentSystem:
    """多 Agent 协作系统"""

    def __init__(self):
        # 注册各个 Agent
        self.agents = {
            "generator": GeneratorAgent(),
            "validator": ValidatorAgent(),
            "searcher": SearcherAgent(),
        }
        # 协调器
        self.orchestrator = Orchestrator(self.agents)

    def process(self, requirement: str) -> str:
        """协调多个 Agent 处理请求"""
        # 1. 搜索相关知识
        context = self.agents["searcher"].search(requirement)

        # 2. 生成（可能需要多轮）
        for attempt in range(3):
            draft = self.agents["generator"].generate(requirement, context)

            # 3. 验证
            validation = self.agents["validator"].validate(draft)

            if validation["passed"]:
                return validation["result"]

        return "处理失败"
```

---

## 📚 学习资源

- [LangGraph 文档](https://langchain-ai.github.io/langgraph/)
- [RAG 最佳实践](https://www.pinecone.io/learn/rag/)
- [Multi-Agent 系统设计](https://arxiv.org/abs/2308.03688)