以实战论 AI：诚邀一线工程师分享模型调优、工程落地真实案例

```python
#!/usr/bin/env python3
"""
RAG 检索优化工具包
- 多阶段检索：向量召回 → Rerank 精排 → 上下文组装
- 分块策略：按 Markdown 结构分块 + 滑动窗口重叠
"""

import re
import json
import numpy as np
from sentence_transformers import SentenceTransformer, CrossEncoder

# ============ 1. 智能分块 ============
class MarkdownChunker:
    """按 Markdown 标题结构分块，保留语义完整性"""
    
    def __init__(self, max_chunk_size=800, overlap=100):
        self.max_chunk_size = max_chunk_size
        self.overlap = overlap
    
    def chunk(self, text, source=""):
        """将文档按标题+段落分块"""
        # 按 Markdown 标题切分
        sections = self._split_by_headers(text)
        
        chunks = []
        for section in sections:
            title = section["title"]
            content = section["content"].strip()
            
            if not content:
                continue
            
            # 如果单个 section 太长，再按段落切
            if len(content) > self.max_chunk_size:
                paragraphs = content.split("\n\n")
                current_chunk = ""
                
                for para in paragraphs:
                    if len(current_chunk) + len(para) > self.max_chunk_size:
                        if current_chunk:
                            chunks.append({
                                "content": f"{title}\n{current_chunk}",
                                "source": source,
                                "section": title,
                            })
                        # 重叠窗口：保留上一块末尾的内容
                        overlap_text = current_chunk[-self.overlap:] if len(current_chunk) > self.overlap else ""
                        current_chunk = overlap_text + "\n" + para
                    else:
                        current_chunk += "\n\n" + para if current_chunk else para
                
                if current_chunk.strip():
                    chunks.append({
                        "content": f"{title}\n{current_chunk}",
                        "source": source,
                        "section": title,
                    })
            else:
                chunks.append({
                    "content": f"{title}\n{content}",
                    "source": source,
                    "section": title,
                })
        
        return chunks
    
    def _split_by_headers(self, text):
        """按 # ## ### 标题切分"""
        sections = []
        current_title = ""
        current_content = ""
        
        for line in text.split("\n"):
            if re.match(r'^#{1,4}\s', line):
                if current_content.strip():
                    sections.append({"title": current_title, "content": current_content})
                current_title = line.strip()
                current_content = ""
            else:
                current_content += line + "\n"
        
        if current_content.strip():
            sections.append({"title": current_title, "content": current_content})
        
        return sections


# ============ 2. 多阶段检索 ============
class HybridRetriever:
    """向量召回 + Rerank 精排"""
    
    def __init__(self):
        # 向量模型（召回阶段）
        self.encoder = SentenceTransformer("BAAI/bge-large-zh-v1.5")
        # Rerank 模型（精排阶段）
        self.reranker = CrossEncoder("BAAI/bge-reranker-large")
        
        self.documents = []
        self.embeddings = None
    
    def index(self, chunks):
        """构建向量索引"""
        self.documents = chunks
        texts = [c["content"] for c in chunks]
        self.embeddings = self.encoder.encode(texts, normalize_embeddings=True)
        print(f"索引完成: {len(chunks)} 个文档块")
    
    def retrieve(self, query, top_k=3, recall_k=20):
        """
        两阶段检索：
        1. 向量召回 top-k*5（宽召回）
        2. Rerank 精排取 top-k（精筛选）
        """
        # === 第一阶段：向量召回 ===
        query_emb = self.encoder.encode([query], normalize_embeddings=True)
        scores = np.dot(self.embeddings, query_emb.T).flatten()
        
        # 宽召回：取 recall_k 个候选
        candidate_indices = np.argsort(scores)[-recall_k:][::-1]
        candidates = [self.documents[i] for i in candidate_indices]
        
        # === 第二阶段：Rerank 精排 ===
        pairs = [[query, c["content"]] for c in candidates]
        rerank_scores = self.reranker.predict(pairs)
        
        # 按 rerank 分数排序，取 top_k
        ranked = sorted(zip(candidates, rerank_scores), key=lambda x: x[1], reverse=True)
        
        results = []
        for doc, score in ranked[:top_k]:
            results.append({
                **doc,
                "rerank_score": float(score),
            })
        
        return results


# ============ 3. 使用示例 ============
if __name__ == "__main__":
    # 模拟知识库文档
    docs = [
        {
            "content": """## Redis 集群脑裂处理
当 Redis 集群出现网络分区时，可能出现脑裂问题。
处理步骤：
1. 配置 min-slaves-to-write=1，确保主节点至少有一个从节点确认写入
2. 启用 Sentinel 自动故障转移
3. 检查网络分区恢复后，确保旧主节点降级为从节点
4. 验证数据一致性""",
            "source": "Redis运维手册.md"
        },
        {
            "content": """## MySQL 慢查询优化
慢查询排查步骤：
1. 开启 slow_query_log
2. 设置 long_query_time=1
3. 使用 EXPLAIN 分析执行计划
4. 检查索引覆盖情况
5. 优化 SQL 写法（避免 SELECT *）""",
            "source": "MySQL调优指南.md"
        },
    ]
    
    # 分块
    chunker = MarkdownChunker(max_chunk_size=800, overlap=100)
    all_chunks = []
    for doc in docs:
        chunks = chunker.chunk(doc["content"], source=doc["source"])
        all_chunks.extend(chunks)
    
    # 检索
    retriever = HybridRetriever()
    retriever.index(all_chunks)
    
    # 测试
    results = retriever.retrieve("Redis 脑裂怎么处理？", top_k=3)
    for r in results:
        print(f"来源: {r['source']} | 分数: {r['rerank_score']:.4f}")
        print(f"内容: {r['content'][:100]}...")
        print()
```

```
请根据以下参考信息回答问题。
参考信息：{context}
问题：{question}
回答：
```

```
你是一个专业的运维工程师。请严格按照以下要求回答问题。

## 参考信息
{context}

## 回答要求
1. 只能基于参考信息回答，不得编造参考信息中不存在的内容
2. 如果参考信息不足以回答问题，请明确回答"当前知识库未覆盖此问题，建议查阅以下文档：[列出可能相关的文档名]"
3. 回答时先给出结论，再给出操作步骤
4. 操作步骤必须编号，每步不超过两句话
5. 如果涉及风险操作，必须标注「⚠️ 风险提示」

## 示例
问题：Redis 内存满了怎么办？
回答：
Redis 内存满了可以通过以下方式处理：

1. 检查内存使用情况：执行 INFO memory 查看 used_memory 和 maxmemory
2. 配置淘汰策略：设置 maxmemory-policy 为 allkeys-lru
⚠️ 风险提示：设置淘汰策略会导致部分 key 被自动删除
3. 扩容：如果数据量持续增长，考虑增加 Redis 节点

## 当前问题
{question}
```

```python
#!/usr/bin/env python3
"""
LoRA 微调数据准备脚本
- 从历史问答数据中构造 SFT 训练集
- 格式化为 Qwen 模型要求的格式
"""

import json
import random
from pathlib import Path

# 微调数据模板
SFT_TEMPLATE = {
    "instruction": "你是一个专业的运维工程师，请基于团队知识库回答问题。",
    "input_format": "问题：{question}\n\n参考信息：{context}",
    "output_format": "{answer}",
}

def build_sft_dataset(raw_qa_pairs, output_file):
    """
    将原始问答对转换为 SFT 训练数据
    
    raw_qa_pairs 格式：
    [
        {
            "question": "灰度发布怎么做？",
            "context": "相关文档片段...",
            "answer": "标准回答..."
        },
    ]
    """
    sft_data = []
    
    for pair in raw_qa_pairs:
        # 构造指令
        instruction = SFT_TEMPLATE["instruction"]
        # 构造输入
        user_input = SFT_TEMPLATE["input_format"].format(
            question=pair["question"],
            context=pair["context"],
        )
        # 构造输出（人工标注的标准答案）
        output = SFT_TEMPLATE["output_format"].format(
            answer=pair["answer"],
        )
        
        sft_data.append({
            "instruction": instruction,
            "input": user_input,
            "output": output,
            "history": [],  # 单轮对话
        })
    
    # 打乱顺序
    random.shuffle(sft_data)
    
    # 划分训练集和验证集（9:1）
    split = int(len(sft_data) * 0.9)
    train_data = sft_data[:split]
    val_data = sft_data[split:]
    
    # 保存
    with open(output_file.replace(".json", "_train.json"), "w", encoding="utf-8") as f:
        json.dump(train_data, f, ensure_ascii=False, indent=2)
    
    with open(output_file.replace(".json", "_val.json"), "w", encoding="utf-8") as f:
        json.dump(val_data, f, ensure_ascii=False, indent=2)
    
    print(f"SFT 数据集生成完成:")
    print(f"  训练集: {len(train_data)} 条 → {output_file.replace('.json', '_train.json')}")
    print(f"  验证集: {len(val_data)} 条 → {output_file.replace('.json', '_val.json')}")


# 数据增强：从知识库自动生成问答对
def auto_generate_qa(knowledge_docs, llm_client):
    """用大模型从知识库文档自动生成问答对"""
    generated = []
    
    for doc in knowledge_docs:
        prompt = f"""请根据以下技术文档，生成 3 个问答对。
要求：
1. 问题应该是运维工程师实际会遇到的问题
2. 答案必须完全基于文档内容，不得编造
3. 答案格式：先结论，后步骤

文档内容：
{doc["content"]}

输出 JSON 数组格式：
[{{"question": "...", "answer": "..."}}, ...]
"""
        response = llm_client.chat.completions.create(
            model="qwen2.5-14b-instruct",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.5,
        )
        
        try:
            qa_pairs = json.loads(response.choices[0].message.content)
            for qa in qa_pairs:
                generated.append({
                    "question": qa["question"],
                    "context": doc["content"][:500],  # 取前500字符作为上下文
                    "answer": qa["answer"],
                    "source": doc["source"],
                })
        except Exception as e:
            print(f"解析失败: {e}")
    
    print(f"自动生成 {len(generated)} 个问答对")
    return generated


if __name__ == "__main__":
    # 模拟人工标注的问答对
    manual_qa = [
        {
            "question": "灰度发布的标准流程是什么？",
            "context": "灰度发布分为四个阶段：准备阶段、灰度阶段、观察阶段、全量阶段...",
            "answer": "灰度发布标准流程：\n1. 准备阶段：确认灰度比例（通常 5%→20%→50%→100%）\n2. 灰度阶段：通过流量网关将指定比例流量导入新版本\n3. 观察阶段：每级灰度后观察 30 分钟，检查错误率和延迟\n4. 全量阶段：确认无异常后全量切换",
            "source": "发布流程规范.md",
        },
    ]
    
    build_sft_dataset(manual_qa, "sft_dataset.json")
```

```python
#!/usr/bin/env python3
"""
生产级 AI 服务封装
- 多模型实例负载均衡
- 请求超时 + 重试 + 降级
- 全链路监控
- 成本控制
"""

import time
import json
import random
import threading
from collections import defaultdict
from dataclasses import dataclass, field
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeout
from openai import OpenAI

# ============ 1. 模型实例池 ============
@dataclass
class ModelInstance:
    """单个模型实例"""
    name: str
    base_url: str
    model: str
    client: OpenAI = None
    request_count: int = 0
    error_count: int = 0
    avg_latency: float = 0
    healthy: bool = True
    
    def __post_init__(self):
        self.client = OpenAI(base_url=self.base_url, api_key="internal", timeout=30)
    
    def chat(self, messages, temperature=0.3, max_tokens=2048):
        start = time.time()
        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                temperature=temperature,
                max_tokens=max_tokens,
            )
            latency = time.time() - start
            self.request_count += 1
            self.avg_latency = (self.avg_latency * (self.request_count - 1) + latency) / self.request_count
            return response.choices[0].message.content, latency
        except Exception as e:
            self.error_count += 1
            if self.error_count > 10:
                self.healthy = False
                print(f"⚠️ 实例 {self.name} 标记为不健康: {e}")
            raise


class ModelInstancePool:
    """模型实例池 + 负载均衡"""
    
    def __init__(self, instances):
        self.instances = instances
        self.lock = threading.Lock()
    
    def get_instance(self, strategy="least_loaded"):
        """选择一个健康的实例"""
        healthy = [i for i in self.instances if i.healthy]
        if not healthy:
            # 全部不健康，尝试恢复
            for i in self.instances:
                i.healthy = True
                i.error_count = 0
            healthy = self.instances
            print("⚠️ 所有实例不健康，尝试全部恢复")
        
        if strategy == "round_robin":
            return random.choice(healthy)
        elif strategy == "least_loaded":
            return min(healthy, key=lambda x: x.request_count)
        else:
            return healthy[0]
    
    def call_with_failover(self, messages, **kwargs):
        """带故障转移的调用"""
        max_retries = 3
        last_error = None
        
        for attempt in range(max_retries):
            instance = self.get_instance()
            try:
                result, latency = instance.chat(messages, **kwargs)
                return result, latency, instance.name
            except Exception as e:
                last_error = e
                print(f"实例 {instance.name} 调用失败 (attempt {attempt+1}): {e}")
                continue
        
        raise last_error


# ============ 2. 超时 + 降级 ============
class AIServiceWithFallback:
    """带超时和降级的 AI 服务"""
    
    def __init__(self, instance_pool, timeout_seconds=15):
        self.pool = instance_pool
        self.timeout = timeout_seconds
        self.executor = ThreadPoolExecutor(max_workers=10)
        
        # 降级策略缓存（常见问题的标准答案）
        self.fallback_cache = {}
        
        # 监控指标
        self.metrics = {
            "total_requests": 0,
            "success_count": 0,
            "timeout_count": 0,
            "fallback_count": 0,
            "latency_list": [],
        }
    
    def chat(self, messages, question=""):
        """带超时和降级的对话"""
        self.metrics["total_requests"] += 1
        start = time.time()
        
        try:
            # 提交到线程池，设置超时
            future = self.executor.submit(
                self.pool.call_with_failover, messages
            )
            result, latency, instance_name = future.result(timeout=self.timeout)
            
            self.metrics["success_count"] += 1
            self.metrics["latency_list"].append(time.time() - start)
            
            # 缓存高频问题的答案
            if question and self.metrics["total_requests"] % 100 == 0:
                self._update_fallback_cache(question, result)
            
            return {
                "answer": result,
                "latency": latency,
                "instance": instance_name,
                "fallback": False,
            }
        
        except FutureTimeout:
            self.metrics["timeout_count"] += 1
            print(f"⏱️ 请求超时 ({self.timeout}s)，尝试降级")
            return self._fallback(question)
        
        except Exception as e:
            self.metrics["timeout_count"] += 1
            print(f"❌ 请求失败: {e}，尝试降级")
            return self._fallback(question)
    
    def _fallback(self, question):
        """降级策略"""
        self.metrics["fallback_count"] += 1
        
        # 策略1：检查缓存
        if question in self.fallback_cache:
            return {
                "answer": self.fallback_cache[question],
                "latency": 0,
                "instance": "cache",
                "fallback": True,
            }
        
        # 策略2：返回友好提示 + 建议操作
        return {
            "answer": "AI 服务当前响应较慢，请稍后重试。如紧急问题请联系值班运维。",
            "latency": 0,
            "instance": "fallback",
            "fallback": True,
        }
    
    def _update_fallback_cache(self, question, answer):
        """更新降级缓存（只缓存高频问题）"""
        if len(answer) > 50:  # 太短的不缓存
            self.fallback_cache[question] = answer
            # 缓存大小限制
            if len(self.fallback_cache) > 200:
                # 淘汰最旧的（简化版，实际用 LRU）
                oldest = next(iter(self.fallback_cache))
                del self.fallback_cache[oldest]
    
    def get_metrics(self):
        """获取监控指标"""
        m = self.metrics.copy()
        if m["latency_list"]:
            m["avg_latency"] = sum(m["latency_list"]) / len(m["latency_list"])
            m["p95_latency"] = sorted(m["latency_list"])[int(len(m["latency_list"]) * 0.95)]
            m["p99_latency"] = sorted(m["latency_list"])[int(len(m["latency_list"]) * 0.99)]
        m["success_rate"] = m["success_count"] / max(m["total_requests"], 1)
        m["fallback_rate"] = m["fallback_count"] / max(m["total_requests"], 1)
        return m


# ============ 3. 使用示例 ============
if __name__ == "__main__":
    # 创建多个模型实例（模拟多副本）
    instances = [
        ModelInstance("llm-node-1", "http://10.10.100.50:8000/v1", "qwen2.5-14b-instruct"),
        ModelInstance("llm-node-2", "http://10.10.100.51:8000/v1", "qwen2.5-14b-instruct"),
        ModelInstance("llm-node-3", "http://10.10.100.52:8000/v1", "qwen2.5-14b-instruct"),
    ]
    
    pool = ModelInstancePool(instances)
    service = AIServiceWithFallback(pool, timeout_seconds=15)
    
    # 模拟并发请求
    def make_request(question):
        messages = [
            {"role": "system", "content": "你是运维助手"},
            {"role": "user", "content": question},
        ]
        result = service.chat(messages, question=question)
        status = "✅" if not result["fallback"] else "🔄降级"
        print(f"{status} [{result['instance']}] {result['latency']:.2f}s | {question[:30]}")
    
    # 并发测试
    questions = [
        "Redis 连接池打满怎么处理？",
        "K8s Pod 频繁重启是什么原因？",
        "MySQL 慢查询怎么排查？",
        "如何配置 Nginx 负载均衡？",
        "Docker 容器无法启动怎么办？",
    ]
    
    with ThreadPoolExecutor(max_workers=5) as executor:
        for q in questions:
            executor.submit(make_request, q)
    
    # 打印监控指标
    print("\n" + "=" * 60)
    print("服务监控指标:")
    metrics = service.get_metrics()
    for k, v in metrics.items():
        if isinstance(v, float):
            print(f"  {k}: {v:.2f}")
        else:
            print(f"  {k}: {v}")
```

```python
#!/usr/bin/env python3
"""
模型路由器
- 根据问题复杂度路由到不同大小的模型
- 简单问题用 7B，复杂问题用 14B
- 降低平均推理成本
"""

import re
from openai import OpenAI

# 路由判断模型（用小模型做分类，成本低）
ROUTER_CLIENT = OpenAI(base_url="http://10.10.100.50:8000/v1", api_key="internal")

# 业务模型池
MODEL_POOL = {
    "simple": OpenAI(base_url="http://10.10.100.50:8001/v1", api_key="internal"),   # 7B
    "complex": OpenAI(base_url="http://10.10.100.50:8000/v1", api_key="internal"),  # 14B
}

ROUTER_PROMPT = """请判断以下问题的复杂度等级。

复杂度判断标准：
- simple: 简单事实查询、单一步骤操作、常见问题（如"Redis 怎么重启"）
- complex: 多步骤排查、需要推理分析、跨系统关联（如"数据库连接池打满同时 Redis 延迟升高怎么排查"）

问题：{question}

只回答 simple 或 complex，不要其他内容。
"""

# 高频问题缓存（命中缓存不走模型）
QUERY_CACHE = {}


def route_and_answer(question, context=""):
    """路由 + 回答"""
    # 1. 检查缓存
    cache_key = question.strip().lower()
    if cache_key in QUERY_CACHE:
        print(f"  → 缓存命中")
        return QUERY_CACHE[cache_key], 0, "cache"
    
    # 2. 路由判断
    route_response = ROUTER_CLIENT.chat.completions.create(
        model="qwen2.5-7b-instruct",
        messages=[{"role": "user", "content": ROUTER_PROMPT.format(question=question)}],
        temperature=0,
        max_tokens=10,
    )
    complexity = route_response.choices[0].message.content.strip().lower()
    
    # 3. 路由到对应模型
    if "simple" in complexity:
        model_key = "simple"
        model_name = "qwen2.5-7b-instruct"
    else:
        model_key = "complex"
        model_name = "qwen2.5-14b-instruct"
    
    print(f"  → 路由到 {model_name}")
    
    # 4. 调用业务模型
    messages = [
        {"role": "system", "content": "你是运维工程师，请基于参考信息回答问题。"},
        {"role": "user", "content": f"参考信息：{context}\n\n问题：{question}"},
    ]
    
    import time
    start = time.time()
    response = MODEL_POOL[model_key].chat.completions.create(
        model=model_name,
        messages=messages,
        temperature=0.3,
        max_tokens=2048,
    )
    latency = time.time() - start
    
    answer = response.choices[0].message.content
    
    # 5. 缓存高频问题
    QUERY_CACHE[cache_key] = answer
    
    return answer, latency, model_name


if __name__ == "__main__":
    test_questions = [
        ("Redis 怎么重启？", "简单问题 → 应路由到 7B"),
        ("数据库连接池打满同时 Redis 延迟升高，怎么排查？", "复杂问题 → 应路由到 14B"),
        ("怎么查看 Nginx 日志？", "简单问题 → 应路由到 7B"),
        ("K8s 集群中 Pod 反复重启，同时服务间调用超时，根因可能是什么？", "复杂问题 → 应路由到 14B"),
    ]
    
    for question, expectation in test_questions:
        print(f"\n问题: {question}")
        print(f"预期: {expectation}")
        answer, latency, model = route_and_answer(question)
        print(f"耗时: {latency:.2f}s | 模型: {model}")
        print(f"回答: {answer[:80]}...")
```