每天用AI干活的技术人，该如何沉淀可复用的工程落地经验？

import os
import json
import hashlib
from datetime import datetime
from pathlib import Path

class PromptRegistry:
    """Prompt模板注册中心——把prompt当代码管"""
    
    def __init__(self, storage_dir="./prompt_registry"):
        self.storage_dir = Path(storage_dir)
        self.storage_dir.mkdir(parents=True, exist_ok=True)
        self.index_file = self.storage_dir / "index.json"
        self._load_index()
    
    def _load_index(self):
        if self.index_file.exists():
            with open(self.index_file, 'r', encoding='utf-8') as f:
                self.index = json.load(f)
        else:
            self.index = {"prompts": []}
    
    def _save_index(self):
        with open(self.index_file, 'w', encoding='utf-8') as f:
            json.dump(self.index, f, ensure_ascii=False, indent=2)
    
    def register(self, name: str, template: str, metadata: dict):
        """注册一个新的prompt模板"""
        prompt_hash = hashlib.md5(template.encode()).hexdigest()[:8]
        
        entry = {
            "id": f"prompt_{prompt_hash}",
            "name": name,
            "template": template,
            "metadata": {
                **metadata,
                "created_at": datetime.now().isoformat(),
                "version": metadata.get("version", "1.0.0"),
            },
            "tags": metadata.get("tags", []),
            "scene": metadata.get("scene", "general"),
            "model": metadata.get("model", "Qwen2.5-7B"),
            "effectiveness_score": metadata.get("effectiveness_score", None),
            "notes": metadata.get("notes", ""),
        }
        
        # 检查是否已有同名模板
        existing = [p for p in self.index["prompts"] if p["name"] == name]
        if existing:
            # 版本对比：如果内容变了，新增版本记录
            old = existing[-1]
            if old["template"] != template:
                old["metadata"]["superseded_at"] = datetime.now().isoformat()
                old["metadata"]["superseded_by"] = entry["id"]
        
        self.index["prompts"].append(entry)
        self._save_index()
        
        # 同时保存独立文件（方便git diff）
        prompt_file = self.storage_dir / f"{entry['id']}.md"
        with open(prompt_file, 'w', encoding='utf-8') as f:
            f.write(f"# {name}\n\n")
            f.write(f"- **场景**: {metadata.get('scene', 'general')}\n")
            f.write(f"- **模型**: {metadata.get('model', 'N/A')}\n")
            f.write(f"- **效果评分**: {metadata.get('effectiveness_score', 'N/A')}/10\n")
            f.write(f"- **版本**: {metadata.get('version', '1.0.0')}\n\n")
            f.write(f"## Template\n\n```\n{template}\n```\n")
            if metadata.get("notes"):
                f.write(f"\n## Notes\n\n{metadata['notes']}\n")
        
        return entry["id"]
    
    def search(self, query: str = None, tag: str = None, scene: str = None):
        """检索prompt模板"""
        results = self.index["prompts"]
        
        # 只看最新版本
        latest = {}
        for p in results:
            if p["name"] not in latest:
                latest[p["name"]] = p
            elif p["metadata"]["created_at"] > latest[p["name"]]["metadata"]["created_at"]:
                latest[p["name"]] = p
        results = list(latest.values())
        
        if scene:
            results = [p for p in results if p["scene"] == scene]
        if tag:
            results = [p for p in results if tag in p["tags"]]
        if query:
            results = [p for p in results 
                       if query.lower() in p["name"].lower() 
                       or query.lower() in p["template"].lower()
                       or query.lower() in p.get("notes", "").lower()]
        
        return results
    
    def get_by_name(self, name: str):
        """按名称获取最新版prompt"""
        matches = [p for p in self.index["prompts"] if p["name"] == name]
        if not matches:
            return None
        return sorted(matches, key=lambda x: x["metadata"]["created_at"])[-1]


# ---- 使用示例 ----
if __name__ == "__main__":
    registry = PromptRegistry("./prompt_registry")
    
    # 注册一个代码审查prompt（第一次版本）
    v1 = registry.register(
        name="code_review_python",
        template="你是一个代码审查专家，请审查以下Python代码：\n{code}",
        metadata={
            "scene": "code_review",
            "tags": ["python", "review"],
            "model": "Qwen2.5-7B-Instruct",
            "effectiveness_score": 6,
            "notes": "基础版本，能识别简单问题但漏报较多",
            "version": "1.0.0",
        }
    )
    
    # 调优后注册第二个版本
    v2 = registry.register(
        name="code_review_python",
        template="""你是一位资深Python代码审查专家，精通PEP 8和常见安全漏洞。

请审查以下代码，重点关注：
1. 逻辑正确性和边界条件
2. SQL注入/XSS等安全问题
3. 异常处理是否完善
4. 性能问题（N+1查询、不必要循环）

以JSON格式输出：issues数组，每个issue含severity/file/line/description/suggestion。

代码：
{code}""",
        metadata={
            "scene": "code_review",
            "tags": ["python", "review", "security"],
            "model": "Qwen2.5-7B-Instruct",
            "effectiveness_score": 8.5,
            "notes": "增加结构化输出和安全审查维度，准确率提升明显。temperature=0.1效果最佳",
            "version": "2.0.0",
        }
    )
    
    # 检索
    results = registry.search(scene="code_review")
    for r in results:
        print(f"[{r['metadata']['version']}] {r['name']} - 评分: {r['effectiveness_score']}/10")
        print(f"  备注: {r['notes']}")

import json
from datetime import datetime
from pathlib import Path

class IncidentKnowledgeBase:
    """AI工程踩坑知识库"""
    
    # 标准化的故障记录模板
    RECORD_TEMPLATE = {
        "id": "",
        "title": "",
        "category": "",           # deployment / performance / prompt / rag / finetune / integration
        "severity": "",            # blocker / major / minor
        "symptoms": [],            # 表象：报错信息/异常行为
        "root_cause": "",          # 根因分析
        "solution": "",            # 解决方案
        "solution_code": "",       # 解决方案代码（如有）
        "environment": {           # 环境信息
            "model": "",
            "framework": "",
            "gpu": "",
            "os": "",
        },
        "tags": [],
        "related_incidents": [],   # 关联的其他记录ID
        "created_at": "",
        "verified_at": "",         # 最后验证有效的时间
        "status": "active",        # active / outdated / deprecated
    }
    
    def __init__(self, storage_path="./ai_incidents_kb.json"):
        self.storage_path = Path(storage_path)
        if self.storage_path.exists():
            with open(self.storage_path, 'r', encoding='utf-8') as f:
                self.records = json.load(f)
        else:
            self.records = []
    
    def add(self, title: str, category: str, severity: str, 
            symptoms: list, root_cause: str, solution: str,
            environment: dict = None, tags: list = None, 
            solution_code: str = ""):
        """添加一条踩坑记录"""
        record = {
            "id": f"INC-{len(self.records)+1:04d}",
            "title": title,
            "category": category,
            "severity": severity,
            "symptoms": symptoms,
            "root_cause": root_cause,
            "solution": solution,
            "solution_code": solution_code,
            "environment": environment or {},
            "tags": tags or [],
            "related_incidents": [],
            "created_at": datetime.now().isoformat(),
            "verified_at": datetime.now().isoformat(),
            "status": "active",
        }
        self.records.append(record)
        self._save()
        return record["id"]
    
    def search(self, query: str = None, category: str = None, 
               severity: str = None, tags: list = None):
        """多维度检索"""
        results = [r for r in self.records if r["status"] == "active"]
        
        if category:
            results = [r for r in results if r["category"] == category]
        if severity:
            results = [r for r in results if r["severity"] == severity]
        if tags:
            results = [r for r in results if any(t in r["tags"] for t in tags)]
        if query:
            q = query.lower()
            results = [r for r in results 
                       if q in r["title"].lower() 
                       or q in r["root_cause"].lower()
                       or q in " ".join(r["symptoms"]).lower()
                       or q in r["solution"].lower()]
        
        return results
    
    def link_incidents(self, id1: str, id2: str):
        """关联两条记录（同类根因/连锁反应）"""
        for r in self.records:
            if r["id"] == id1 and id2 not in r["related_incidents"]:
                r["related_incidents"].append(id2)
            if r["id"] == id2 and id1 not in r["related_incidents"]:
                r["related_incidents"].append(id1)
        self._save()
    
    def _save(self):
        with open(self.storage_path, 'w', encoding='utf-8') as f:
            json.dump(self.records, f, ensure_ascii=False, indent=2)


# ---- 使用示例 ----
if __name__ == "__main__":
    kb = IncidentKnowledgeBase("./ai_incidents_kb.json")
    
    # 记录一个真实踩坑：vLLM显存OOM
    inc1 = kb.add(
        title="vLLM并发请求时CUDA Out of Memory",
        category="deployment",
        severity="blocker",
        symptoms=[
            "vLLM服务启动正常，低并发时运行正常",
            "并发请求>10时报错: CUDA error: out of memory",
            "nvidia-smi显示显存占用接近100%",
        ],
        root_cause="vLLM默认gpu-memory-utilization=0.9，预分配KV Cache过大。"
                   "高并发时多个请求的KV Cache叠加，超出显存上限。",
        solution="降低gpu-memory-utilization到0.85，同时限制max-num-seqs控制并发批次大小",
        solution_code="""python -m vllm.entrypoints.openai.api_server \\
    --model Qwen/Qwen2.5-7B-Instruct \\
    --gpu-memory-utilization 0.85 \\
    --max-num-seqs 16 \\
    --max-model-len 4096""",
        environment={
            "model": "Qwen2.5-7B-Instruct",
            "framework": "vLLM 0.6.0",
            "gpu": "RTX 4090 24GB",
            "os": "Ubuntu 22.04",
        },
        tags=["vllm", "oom", "gpu-memory", "concurrency"],
    )
    
    # 记录另一个关联问题：量化后精度下降
    inc2 = kb.add(
        title="AWQ量化后模型输出质量明显下降",
        category="performance",
        severity="major",
        symptoms=[
            "量化前回答准确率88%",
            "AWQ量化后准确率降至72%",
            "模型出现格式错乱和重复输出",
        ],
        root_cause="使用的是社区版AWQ量化模型，量化校准数据集与业务领域不匹配。"
                   "通用校准数据对中文业务场景覆盖不足。",
        solution="使用业务领域数据自行做AWQ量化校准，或改用GPTQ量化（对本场景影响更小）",
        environment={
            "model": "Qwen2.5-7B-Instruct",
            "framework": "vLLM 0.6.0 + AutoAWQ",
            "gpu": "RTX 4090 24GB",
        },
        tags=["quantization", "awq", "accuracy", "quality"],
    )
    
    # 关联两条记录
    kb.link_incidents(inc1, inc2)
    
    # 搜索
    results = kb.search(query="显存")
    for r in results:
        print(f"[{r['id']}] {r['title']}")
        print(f"  根因: {r['root_cause'][:60]}...")
        print(f"  方案: {r['solution'][:60]}...")

from datetime import datetime
from pathlib import Path
import json

class DecisionRecord:
    """技术决策记录(ADR)——记录为什么这么做"""
    
    TEMPLATE = """# {title}

## 状态
{status}  # proposed / accepted / deprecated / superseded

## 背景
{context}

## 决策
{decision}

## 方案对比
{alternatives}

## 理由
{rationale}

## 影响
{consequences}

## 验证
{validation}

---
- 决策时间: {date}
- 决策参与者: {participants}
- 关联文档: {related}
"""
    
    def __init__(self, storage_dir="./decisions"):
        self.storage_dir = Path(storage_dir)
        self.storage_dir.mkdir(parents=True, exist_ok=True)
    
    def create(self, title: str, context: str, decision: str,
               alternatives: str, rationale: str,
               consequences: str = "", validation: str = "",
               participants: list = None, related: list = None):
        """创建一条决策记录"""
        date_str = datetime.now().strftime("%Y-%m-%d")
        file_name = f"{date_str}_{title[:20].replace(' ', '_')}.md"
        file_path = self.storage_dir / file_name
        
        content = self.TEMPLATE.format(
            title=title,
            status="accepted",
            context=context,
            decision=decision,
            alternatives=alternatives,
            rationale=rationale,
            consequences=consequences,
            validation=validation,
            date=date_str,
            participants=", ".join(participants or []),
            related=", ".join(related or []),
        )
        
        with open(file_path, 'w', encoding='utf-8') as f:
            f.write(content)
        
        return file_path


# 使用示例
dr = DecisionRecord("./decisions")

dr.create(
    title="推理框架选型: vLLM vs TGI vs Ollama",
    context="""团队需要部署Qwen2.5-7B-Instruct作为内部AI服务的推理引擎。
预估日均调用量5万次，峰值QPS约50，要求P99延迟<3秒。
部署环境: 单台A10(24GB) GPU服务器。""",
    
    decision="选择vLLM作为推理框架",
    
    alternatives="""### 方案A: vLLM
- 优势: PagedAttention技术，高并发吞吐量最优；OpenAI API兼容
- 劣势: 社区相对年轻，偶有bug；配置参数较多

### 方案B: HuggingFace TGI
- 优势: 生态成熟，与transformers库无缝对接；文档完善
- 劣势: 吞吐量比vLLM低约30%(实测)；启动慢

### 方案C: Ollama
- 优势: 部署极简，一行命令搞定；适合个人/小团队
- 劣势: 不支持高并发；无 batching；不适合生产环境""",
    
    rationale="""核心决策因素是高并发吞吐量。压测结果:
- vLLM: QPS=50时P99=2.1s，吞吐量1200 tokens/s
- TGI: QPS=50时P99=3.8s，吞吐量850 tokens/s  
- Ollama: QPS=10时P99=4.5s，无法满足50 QPS需求

vLLM在吞吐量和延迟上都明显领先，且API兼容OpenAI格式，
团队现有代码迁移成本最低。""",
    
    consequences="""正面:
- 满足性能SLA要求
- API兼容性好，迁移成本低

负面:
- 需要关注vLLM版本更新，偶有breaking change
- 需要专人维护vLLM配置参数调优""",
    
    validation="""部署后压测: QPS=50, P99=2.3s, 吞吐量1150 tokens/s
与预估基本一致，SLA达标。连续运行7天无异常。""",
    
    participants=["架构师A", "后端负责人B", "运维C"],
    related=["部署架构设计文档", "vLLM配置参数调优记录"],
)

import json
import time
from datetime import datetime
from openai import OpenAI

class EvaluationBenchmark:
    """AI方案效果评估基准"""
    
    def __init__(self, benchmark_path="./benchmarks"):
        self.benchmark_path = benchmark_path
        self.client = OpenAI(base_url="http://localhost:8000/v1", api_key="empty")
        self.benchmarks = self._load_benchmarks()
    
    def _load_benchmarks(self):
        """加载评估数据集"""
        path = f"{self.benchmark_path}/golden_set.json"
        try:
            with open(path, 'r', encoding='utf-8') as f:
                return json.load(f)
        except FileNotFoundError:
            return {"test_cases": []}
    
    def add_test_case(self, question: str, expected_answer: str, 
                      category: str, keywords: list = None):
        """添加评估用例（黄金标准QA对）"""
        case = {
            "id": f"TC-{len(self.benchmarks['test_cases'])+1:03d}",
            "question": question,
            "expected_answer": expected_answer,
            "category": category,
            "keywords": keywords or [],
            "added_at": datetime.now().isoformat(),
        }
        self.benchmarks["test_cases"].append(case)
        self._save_benchmarks()
    
    def run_evaluation(self, model_name: str, prompt_template: str = ""):
        """对指定模型+prompt运行评估"""
        results = []
        
        for case in self.benchmarks["test_cases"]:
            start_time = time.time()
            
            messages = [
                {"role": "system", "content": prompt_template or "你是技术助手"},
                {"role": "user", "content": case["question"]},
            ]
            
            try:
                response = self.client.chat.completions.create(
                    model=model_name,
                    messages=messages,
                    temperature=0.0,
                    max_tokens=512,
                )
                answer = response.choices[0].message.content
                latency = time.time() - start_time
                
                # 自动评分：关键词覆盖率
                keyword_coverage = self._calc_keyword_coverage(
                    answer, case.get("keywords", [])
                )
                
                results.append({
                    "case_id": case["id"],
                    "category": case["category"],
                    "question": case["question"],
                    "expected": case["expected_answer"],
                    "actual": answer,
                    "keyword_coverage": keyword_coverage,
                    "latency_ms": round(latency * 1000, 1),
                    "status": "pass" if keyword_coverage >= 0.6 else "review",
                })
            except Exception as e:
                results.append({
                    "case_id": case["id"],
                    "status": "error",
                    "error": str(e),
                })
        
        # 汇总指标
        summary = self._summarize(results, model_name)
        return summary
    
    def _calc_keyword_coverage(self, answer: str, keywords: list):
        """计算关键词覆盖率"""
        if not keywords:
            return 1.0
        hit = sum(1 for kw in keywords if kw.lower() in answer.lower())
        return hit / len(keywords)
    
    def _summarize(self, results: list, model_name: str):
        passed = [r for r in results if r["status"] == "pass"]
        review = [r for r in results if r["status"] == "review"]
        errors = [r for r in results if r["status"] == "error"]
        latencies = [r["latency_ms"] for r in results if "latency_ms" in r]
        
        return {
            "model": model_name,
            "total_cases": len(results),
            "passed": len(passed),
            "needs_review": len(review),
            "errors": len(errors),
            "pass_rate": round(len(passed) / max(len(results), 1) * 100, 1),
            "avg_latency_ms": round(sum(latencies) / max(len(latencies), 1), 1) if latencies else 0,
            "p99_latency_ms": round(sorted(latencies)[int(len(latencies)*0.99)] if latencies else 0, 1),
            "evaluated_at": datetime.now().isoformat(),
            "details": results,
        }
    
    def _save_benchmarks(self):
        path = f"{self.benchmark_path}/golden_set.json"
        import os
        os.makedirs(self.benchmark_path, exist_ok=True)
        with open(path, 'w', encoding='utf-8') as f:
            json.dump(self.benchmarks, f, ensure_ascii=False, indent=2)
    
    def save_result(self, result: dict, name: str):
        """保存评估结果用于历史对比"""
        path = f"{self.benchmark_path}/results/{name}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
        import os
        os.makedirs(os.path.dirname(path), exist_ok=True)
        with open(path, 'w', encoding='utf-8') as f:
            json.dump(result, f, ensure_ascii=False, indent=2)
        print(f"评估结果已保存: {path}")


# 使用示例
bench = EvaluationBenchmark("./benchmarks")

# 添加黄金标准测试用例（从业务真实QA中精选）
bench.add_test_case(
    question="公司的退换货政策是什么？",
    expected_answer="7天无理由退货，15天质量问题换货",
    category="policy",
    keywords=["7天", "无理由", "退货", "15天", "换货"],
)

bench.add_test_case(
    question="如何修改收货地址？",
    expected_answer="在订单详情页点击修改地址，已发货订单不可修改",
    category="operation",
    keywords=["订单详情", "修改地址", "已发货", "不可修改"],
)

# 运行评估
result = bench.run_evaluation(
    model_name="Qwen/Qwen2.5-7B-Instruct",
    prompt_template="你是XX电商平台的客服助手，基于参考资料回答用户问题。",
)
bench.save_result(result, "qwen25_7b_baseline")

print(f"通过率: {result['pass_rate']}%")
print(f"平均延迟: {result['avg_latency_ms']}ms")
print(f"P99延迟: {result['p99_latency_ms']}ms")