AI开发工程师全流程工作流

SKILL.md

---
name: ai-dev-engineer
description: "AI开发工程师全流程工作流。覆盖需求分析→技术选型→数据处理→AI核心开发(Prompt/Agent/RAG/微调)→后端服务→前端交互→测试评估→部署运维→性能优化8大阶段。面向LLM应用开发全链路，提供代码模板、架构决策框架、调试排查指南。触发词: AI开发, AI工程师, AI全栈, LLM应用开发, Agent开发, RAG开发, Prompt工程, 模型微调, AI部署, AI后端, AI前端, 搭建AI应用, AI系统设计, AI架构, MLOps, AI工程化, AI dev, LLM app, AI fullstack, build AI app, AI system design."
version: "1.0.0"
agent_created: true
metadata:
  emoji: "⚙️"
  tags: ["ai", "engineering", "llm", "rag", "agent", "fullstack", "devops", "mlops"]
  category: "development"
---

# AI Dev Engineer — AI开发工程师全流程工作流

> 从需求到上线的 AI 应用工程化全链路。侧重编码落地、架构设计、性能优化、运维监控——是 AI PM 的工程执行搭档。

---

## 概述

AI Dev Engineer Skill 覆盖 AI 应用开发的完整工程链路，共 **8 大阶段、32 个关键动作**。与 ai-pm（产品视角）互补，本 skill 聚焦**工程实现**。

```
①需求与技术选型 → ②数据工程 → ③AI核心开发 → ④后端服务开发
                                        ↓
⑧性能优化与迭代 ← ⑦部署与运维 ← ⑥测试与评估 ← ⑤前端交互开发
```

---

## 阶段一：需求分析与技术选型

**触发**: 拿到需求描述、PRD、或用户说"帮我设计一个AI系统的技术方案"

### 执行流程

#### 1.1 需求结构化拆解

将产品需求转化为工程任务清单：

```markdown
| 产品需求 | 工程任务 | 技术复杂度 | AI依赖度 | 优先级 |
|---------|---------|:--------:|:------:|:----:|
| 用户上传PDF并提问 | RAG文档问答 | 中 | 高 | P0 |
| 对话历史管理 | 会话存储+上下文窗口 | 低 | 中 | P0 |
| 多轮追问能力 | Agent工具调用 | 高 | 高 | P1 |
```

#### 1.2 技术架构决策

根据场景选择架构模式：

| 场景 | 推荐架构 | 适用条件 |
|------|---------|---------|
| 简单ChatBot | LLM API + 前端 | 无需外部数据、无需工具 |
| 知识库问答 | RAG架构 | 有私有文档/知识库 |
| 复杂工具调用 | Agent架构 (ReAct/Plan-Execute) | 需要多步推理+外部工具 |
| 多任务协同 | Multi-Agent架构 | 任务可拆解为独立子任务 |
| 高并发服务 | LLM API + 语义缓存 + 队列 | QPS > 100 |

#### 1.3 技术栈选型

快速决策矩阵：

```
┌─ 语言: Python (首选, AI生态最完善) / TypeScript (全栈偏好)
├─ 框架: FastAPI (高性能API) / LangChain (快速原型) / 自研 (灵活性最高)
├─ 模型API: OpenAI / DashScope / DeepSeek / 本地部署(vLLM/Ollama)
├─ 向量数据库: Milvus (生产级) / Chroma (原型) / pgvector (已有PG)
├─ 数据库: PostgreSQL + Redis (会话缓存)
├─ 部署: Docker + Docker Compose / K8s
└─ 监控: LangFuse / LangSmith / 自建 (Prometheus+Grafana)
```

#### 1.4 成本估算

```
月成本 = DAU × 平均对话轮次 × (输入Token × 输入单价 + 输出Token × 输出单价)
       + 向量数据库费用(云服务) 或 服务器费用(自建)
       + Embedding API调用费用
```

输出: 技术方案文档 + 成本估算表 + 架构图

---

## 阶段二：数据工程

**触发**: 需要准备训练数据、构建知识库、处理标注数据

### 执行流程

#### 2.1 数据采集与清洗

```python
# 典型的数据清洗 pipeline
import re
from typing import List, Dict

def clean_documents(raw_docs: List[Dict]) -> List[Dict]:
    """文档清洗: 去重、去噪、标准化"""
    cleaned = []
    seen = set()
    for doc in raw_docs:
        # 去重
        content_hash = hash(doc['content'][:200])
        if content_hash in seen:
            continue
        seen.add(content_hash)
        # 去噪: 移除多余空白、特殊字符
        doc['content'] = re.sub(r'\s+', ' ', doc['content']).strip()
        # 过滤过短/过长文档
        if 50 < len(doc['content']) < 10000:
            cleaned.append(doc)
    return cleaned
```

#### 2.2 RAG文档切片策略

```
切片决策树:
├─ 结构化文档 (Markdown/HTML)
│   └─ 按标题层级切片 (MarkdownHeaderTextSplitter)
├─ 非结构化文本
│   ├─ 短文档 (< 2000字): 整篇不切
│   ├─ 中文文档: Chunk=512 tokens, Overlap=50
│   └─ 代码: Chunk=256 tokens, 按函数边界切
└─ 表格数据
    └─ 保留为结构化JSON, 不加切片
```

#### 2.3 向量化与入库

```python
# 标准 RAG 入库流程
from openai import OpenAI

def embed_and_store(docs: List[Dict], collection_name: str):
    """文档向量化并存入向量数据库"""
    client = OpenAI()
    embeddings = []
    for doc in docs:
        resp = client.embeddings.create(
            model="text-embedding-3-small",  # 性价比最优
            input=doc['content']
        )
        embeddings.append({
            'id': doc['id'],
            'vector': resp.data[0].embedding,
            'metadata': doc['metadata']
        })
    # 批量写入向量数据库
    vector_db.upsert(collection_name, embeddings)
```

#### 2.4 微调数据准备

```jsonl
{"messages": [{"role": "system", "content": "你是AI客服"}, {"role": "user", "content": "如何退货?"}, {"role": "assistant", "content": "在订单详情页点击申请退货..."}]}
```

检查清单:
- [ ] 数据去重 (MinHash/SimHash)
- [ ] 格式校验 (JSONL 完整性)
- [ ] 质量抽样 (人工抽检 5%)
- [ ] 分布检查 (类别均衡性)

---

## 阶段三：AI核心开发

> 这是 AI 开发工程师的核心战场。

### 3.1 Prompt Engineering (工程化)

**三层 Prompt 架构:**

```
┌─────────────────────────────┐
│  System Prompt (角色+约束)    │  ← 固定, 版本管理
├─────────────────────────────┤
│  Context (RAG结果/用户画像)   │  ← 动态注入
├─────────────────────────────┤
│  User Message (当前输入)     │  ← 原始用户输入
└─────────────────────────────┘
```

**Prompt 版本管理模板:**

```python
# prompts/v1/customer_service.py
SYSTEM_PROMPT_V1 = """你是{company_name}的AI客服专家。
## 能力范围
- 回答产品相关问题
- 处理退换货咨询
- 查询订单状态

## 约束
- 不确定时明确告知，不要编造
- 涉及退款金额时必须确认后回答
- 语气友好但不谄媚"""

# prompts/v1/__init__.py 中做版本注册
PROMPT_REGISTRY = {
    "customer_service": {"v1": SYSTEM_PROMPT_V1, "default": "v1"},
}
```

**调试技巧:**
- 用 `{ }` 标记动态变量，便于检查遗漏
- 每个 Prompt 写单元测试 (给定输入断言输出格式)
- 记录每次 Prompt 变更的 A/B 结果

### 3.2 Agent 开发

**ReAct Agent 标准实现:**

```python
from typing import List, Dict, Any, Callable
import json

class ReActAgent:
    """标准 ReAct Agent 实现"""
    
    def __init__(self, llm_call: Callable, tools: Dict[str, Callable]):
        self.llm = llm_call
        self.tools = tools
        self.max_steps = 10
    
    def run(self, user_input: str) -> str:
        messages = [{"role": "system", "content": self._build_system_prompt()}]
        messages.append({"role": "user", "content": user_input})
        
        for step in range(self.max_steps):
            response = self.llm(messages)
            action = self._parse_action(response)
            
            if action['type'] == 'final_answer':
                return action['content']
            elif action['type'] == 'tool_call':
                tool_result = self._execute_tool(
                    action['tool'], action['input']
                )
                messages.append({"role": "assistant", "content": response})
                messages.append({
                    "role": "user", 
                    "content": f"工具返回: {tool_result}"
                })
        
        return "达到最大步数限制，请简化问题重试"
    
    def _build_system_prompt(self) -> str:
        tool_desc = "\n".join([
            f"- {name}: {desc}" 
            for name, (_, desc) in self.tools.items()
        ])
        return f"""你是AI助手，可使用以下工具完成任务:
{tool_desc}

回复格式:
- 使用工具: {{"action": "tool", "tool": "工具名", "input": "参数"}}
- 最终回答: {{"action": "final", "content": "答案"}}"""
```

**Multi-Agent 编排模式:**

```
模式A - 顺序流水线: Agent1 → Agent2 → Agent3
  适用: 任务有明确先后依赖 (分析→设计→实现)

模式B - 并行+汇总: Agent1 ↘
                  Agent2 → 汇总Agent → 输出
                  Agent3 ↗
  适用: 独立子任务可并行处理

模式C - 辩论模式: Agent1 ↔ Agent2 (多轮辩论) → 裁判Agent
  适用: 需要多角度验证的决策场景
```

### 3.3 RAG 系统开发

**完整 RAG Pipeline:**

```python
class RAGPipeline:
    """生产级 RAG Pipeline"""
    
    def __init__(self, embedding_model: str, llm_model: str, 
                 vector_db, reranker=None):
        self.embedding_model = embedding_model
        self.llm_model = llm_model
        self.vector_db = vector_db
        self.reranker = reranker
    
    def query(self, question: str, top_k: int = 10, rerank_k: int = 3):
        # Step 1: Query 改写 (处理指代消解、拼写纠错)
        query = self._rewrite_query(question)
        
        # Step 2: 向量检索 + 关键词检索 (混合检索)
        vector_results = self.vector_db.search(query, top_k=top_k)
        keyword_results = self._bm25_search(query, top_k=top_k)
        
        # Step 3: 融合排序 (RRF - Reciprocal Rank Fusion)
        fused = self._reciprocal_rank_fusion(
            [vector_results, keyword_results], k=60
        )
        
        # Step 4: Rerank (可选)
        if self.reranker:
            fused = self.reranker.rerank(query, fused)[:rerank_k]
        
        # Step 5: 生成回答
        context = "\n\n".join([doc['content'] for doc in fused])
        answer = self.llm.generate(
            prompt=f"基于以下参考资料回答问题:\n{context}\n\n问题: {question}"
        )
        
        return {
            'answer': answer,
            'sources': [doc['source'] for doc in fused],
            'confidence': self._estimate_confidence(answer, fused)
        }
```

**RAG 优化清单:**
- [ ] 混合检索 (向量 + BM25) 优于单一检索
- [ ] Query Rewriting 显著提升召回率
- [ ] Small-to-Big 检索: 检索小片段，返回大上下文
- [ ] Re-ranking 是性价比最高的精度提升手段
- [ ] 加入引用标注提高可信度

### 3.4 模型微调 (Fine-tuning)

**微调决策框架:**

```
需要微调吗?
├─ Prompt优化能解决? → 不需要, 改Prompt
├─ Few-shot示例能解决? → 不需要, 加示例
├─ 特定领域知识? → 先试RAG → 不够再微调
├─ 特定格式/风格? → 微调 (数据量 100-1000条)
├─ 新能力/任务? → 微调 (数据量 1000+条)
└─ 成本和延迟优化? → 蒸馏/微调小模型
```

```python
# LoRA 微调核心流程 (使用 QLoRA 降低显存)
# 关键参数:
# - r=8~64 (秩, 越大能力越强但越慢)
# - lora_alpha=16~32 (缩放因子)
# - target_modules=["q_proj", "v_proj"] (Qwen/Llama 通用)
# - 学习率: 2e-4 ~ 5e-5
```

---

## 阶段四：后端服务开发

### 4.1 API 架构设计

```python
# FastAPI 标准项目结构
project/
├── app/
│   ├── api/
│   │   ├── v1/
│   │   │   ├── chat.py        # 对话接口
│   │   │   ├── rag.py         # RAG 接口
│   │   │   └── agent.py       # Agent 接口
│   │   └── deps.py            # 依赖注入
│   ├── core/
│   │   ├── config.py          # 配置管理
│   │   ├── llm.py             # LLM 调用封装
│   │   └── security.py        # 鉴权
│   ├── models/                # 数据模型
│   ├── services/              # 业务逻辑层
│   └── main.py                # 入口
├── tests/
├── Dockerfile
└── docker-compose.yml
```

### 4.2 流式输出 (SSE)

```python
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio
import json

@app.post("/api/v1/chat/stream")
async def chat_stream(request: ChatRequest):
    async def generate():
        async for chunk in llm_service.stream_chat(
            messages=request.messages,
            temperature=request.temperature
        ):
            yield f"data: {json.dumps({'delta': chunk, 'finish': False})}\n\n"
        yield f"data: {json.dumps({'delta': '', 'finish': True})}\n\n"
    
    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",  # Nginx 禁用缓冲
        }
    )
```

### 4.3 关键中间件

```python
# 1. 请求限流
from slowapi import Limiter
limiter = Limiter(key_func=lambda: "global")

# 2. 语义缓存 (避免重复调用LLM)
class SemanticCache:
    def get_or_compute(self, query: str, func, threshold=0.92):
        cached = self.vector_db.search(query, top_k=1)
        if cached and cached[0]['score'] > threshold:
            return cached[0]['response']
        result = func(query)
        self.store(query, result)
        return result

# 3. Token 计数与成本追踪
def estimate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
    pricing = {
        "gpt-4o": (2.5/1e6, 10/1e6),
        "gpt-4o-mini": (0.15/1e6, 0.6/1e6),
        "deepseek-chat": (0.14/1e6, 0.28/1e6),
    }
    in_price, out_price = pricing.get(model, (0, 0))
    return input_tokens * in_price + output_tokens * out_price
```

---

## 阶段五：前端交互开发

### 5.1 AI聊天界面核心要素

```
┌─────────────────────────────────┐
│  Header: 标题 + 新建对话 + 设置   │
├─────────────────────────────────┤
│                                 │
│  Message List (虚拟滚动)         │
│  ┌─ User Bubble ────────────┐  │
│  │ 用户消息                    │  │
│  └──────────────────────────┘  │
│  ┌─ AI Bubble ──────────────┐  │
│  │ 流式渲染中的AI回复...       │  │
│  │ 📎 来源: doc1.pdf, doc2   │  │
│  └──────────────────────────┘  │
│                                 │
├─────────────────────────────────┤
│  Input: [文本框] [上传] [发送]   │
└─────────────────────────────────┘
```

### 5.2 流式渲染实现

```typescript
// React SSE 流式接收
async function* streamChat(messages: Message[]) {
  const response = await fetch('/api/v1/chat/stream', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ messages }),
  });
  
  const reader = response.body!.getReader();
  const decoder = new TextDecoder();
  let buffer = '';
  
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    buffer += decoder.decode(value, { stream: true });
    
    // 解析 SSE 事件
    const lines = buffer.split('\n');
    buffer = lines.pop() || '';
    for (const line of lines) {
      if (line.startsWith('data: ')) {
        const data = JSON.parse(line.slice(6));
        if (data.finish) return;
        yield data.delta;
      }
    }
  }
}
```

### 5.3 关键交互细节

- **Markdown 渲染**: 使用 react-markdown + rehype-highlight 支持代码高亮
- **思考过程展示**: Agent 模式下展示 `思考 → 行动 → 观察` 循环
- **来源引用**: RAG 回答附带原文链接，点击可展开
- **停止生成**: 提供 AbortController 中断流式请求
- **错误重试**: 网络异常时自动重试 (最多3次，指数退避)

---

## 阶段六：测试与评估

### 6.1 AI 应用的测试分层

```
┌──────────────────────────┐
│  E2E 测试 (端到端场景)     │  ← 少量, 覆盖核心用户旅程
├──────────────────────────┤
│  Eval 测试 (AI质量评估)    │  ← AI应用特有, Golden Dataset
├──────────────────────────┤
│  集成测试 (API + DB + LLM) │  ← Mock LLM 响应
├──────────────────────────┤
│  单元测试 (纯逻辑)         │  ← 大量, 快速反馈
└──────────────────────────┘
```

### 6.2 LLM Eval 框架

```python
# 使用 Golden Dataset 进行评估
def evaluate_prompt(prompt_template, test_cases, llm):
    """评估Prompt效果"""
    results = []
    for case in test_cases:
        response = llm(prompt_template.format(query=case['query']))
        results.append({
            'query': case['query'],
            'expected': case['expected'],
            'actual': response,
            'pass': case['expected'].lower() in response.lower(),
            'latency_ms': response.latency,
        })
    
    accuracy = sum(r['pass'] for r in results) / len(results)
    avg_latency = sum(r['latency_ms'] for r in results) / len(results)
    
    return {
        'accuracy': accuracy,
        'avg_latency_ms': avg_latency,
        'total': len(results),
        'failed_cases': [r for r in results if not r['pass']],
    }
```

### 6.3 测试检查清单

- [ ] 单元测试覆盖率 > 80% (非 LLM 调用部分)
- [ ] Golden Dataset 覆盖所有核心场景 (至少 50 条)
- [ ] 安全测试: 注入攻击、PII泄露、有害内容过滤
- [ ] 性能测试: P50/P95/P99 延迟、并发压测
- [ ] 降级测试: LLM 不可用时的兜底行为
- [ ] Prompt 变更回归测试 (每次改 Prompt 必须跑 Eval)

---

## 阶段七：部署与运维

### 7.1 Docker 容器化

```dockerfile
# 多阶段构建示例
FROM python:3.11-slim AS builder
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

FROM python:3.11-slim
WORKDIR /app
COPY --from=builder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages
COPY app/ ./app/
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
```

### 7.2 CI/CD Pipeline

```
Git Push → Lint + Type Check → Unit Tests → Eval Tests → Build Image
                                                              ↓
                                              Deploy Staging → Smoke Test → Deploy Prod
```

### 7.3 监控体系

```python
# 关键监控指标
metrics = {
    'latency': {
        'p50_ms', 'p95_ms', 'p99_ms',  # 延迟分布
        'first_token_ms',               # 首Token时间(流式)
    },
    'quality': {
        'thumbs_up_rate',   # 用户点赞率
        'thumbs_down_rate', # 用户点踩率
        'regeneration_rate',# 重新生成率
    },
    'cost': {
        'tokens_per_request', # 每次请求Token消耗
        'daily_cost_usd',     # 日成本
        'cache_hit_rate',     # 缓存命中率
    },
    'reliability': {
        'error_rate',      # 错误率
        'timeout_rate',    # 超时率
        'fallback_rate',   # 降级触发率
    }
}
```

### 7.4 灰度发布策略

```
阶段1: 内部测试 (1%流量) → 观察 2小时
阶段2: 小流量 (5%) → 观察 24小时
阶段3: 中流量 (25%) → 观察 24小时
阶段4: 大流量 (50%) → 观察 12小时
阶段5: 全量 (100%)

每个阶段需验证: 错误率不升、延迟不升、点赞率不降
回滚条件: 任意核心指标恶化 > 20% 或 错误率 > 5%
```

---

## 阶段八：性能优化与迭代

### 8.1 延迟优化清单

| 优化手段 | 预期提升 | 实现复杂度 | 适用场景 |
|---------|:------:|:--------:|---------|
| 语义缓存 | 50-80% | 中 | 重复问题多 |
| Prompt 精简 | 10-30% | 低 | 所有场景 |
| 流式输出 | 感知延迟降70% | 低 | 所有场景 |
| 模型降级 (强→弱) | 30-50% | 低 | 简单任务 |
| 并行调用 | 40-60% | 中 | 多步独立Agent |
| 预测用户意图 | 20-30% | 高 | 高频固定场景 |
| 模型量化/蒸馏 | 50-70% | 高 | 大规模部署 |

### 8.2 成本优化

```python
# 成本优化策略
class CostOptimizer:
    
    def route_model(self, query: str) -> str:
        """智能路由: 简单问题用小模型，复杂问题用大模型"""
        complexity = self.estimate_complexity(query)
        if complexity == "simple":
            return "gpt-4o-mini"  # $0.15/M input
        elif complexity == "medium":
            return "gpt-4o"       # $2.50/M input  
        else:
            return "claude-opus"  # 最强模型
    
    def should_use_cache(self, query: str) -> bool:
        """判断是否走缓存"""
        return self.semantic_cache.has_similar(query, threshold=0.95)
```

### 8.3 Bad Case 驱动迭代

```
发现问题 → 归类根因 → 选择修复策略:
├─ Prompt 问题 (40%)
│   └─ 修改Prompt → 回归测试 → 灰度上线
├─ 知识库缺失 (30%)
│   └─ 补充文档 → 重新入库 → 验证召回
├─ 模型能力不足 (20%)
│   └─ 换模型 / 微调 / 加约束
└─ 架构缺陷 (10%)
    └─ 重构Agent流程 / 增加校验层
```

---

## 使用模式

### 快捷模式

用户提出具体开发任务，直接匹配对应阶段:

- `帮我设计一个AI客服的技术方案` → 阶段一
- `清洗这批文档准备RAG入库` → 阶段二
- `写一个ReAct Agent` / `优化这段Prompt` → 阶段三
- `搭建FastAPI后端` / `实现流式输出` → 阶段四
- `写一个AI聊天前端组件` → 阶段五
- `搭建LLM评估体系` / `写Golden Dataset测试` → 阶段六
- `Docker部署这个AI应用` / `配置监控` → 阶段七
- `分析线上Bad Case` / `优化延迟` → 阶段八

### 全流程模式

用户说"帮我从零搭建一个XXX的AI应用"时，按顺序推进八个阶段，每个阶段产出后确认。

### 调试模式

遇到具体问题时:
- Prompt 效果差 → 3.1 检查版本、跑 Eval、加约束
- RAG 召回低 → 3.3 检查切片策略、试混合检索
- Agent 跑偏 → 3.2 检查工具描述、加 step 限制
- 高延迟 → 8.1 加缓存、精简 Prompt、切小模型
- 部署问题 → 7.1 Docker 日志排查

---

## 核心原则

1. **先简单后复杂**: Prompt → RAG → Agent → 微调, 循序渐进
2. **可观测性内置**: 每个服务必须有日志、指标、追踪
3. **降级必有**: 每个 AI 调用必须有非 AI 的兜底方案
4. **成本可视化**: Token消耗实时追踪, 异常告警
5. **Eval 驱动迭代**: 没有评估的优化是盲目的
6. **安全第一**: 输入过滤 + 输出审核 + 速率限制

---

## 附加资源

- `references/tech_stack_guide.md` — 详细技术栈对比与选型指南
- `references/common_pitfalls.md` — AI 开发常见坑与解决方案
- `scripts/eval_runner.py` — LLM 评估测试运行脚本
- `scripts/cost_tracker.py` — Token 成本追踪脚本