开源大模型热潮下，一线工程师该抓住哪些AI落地机会？

```python
import os
import re
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class ProjectContext:
    """项目上下文信息"""
    language: str = "python"
    framework: str = ""
    dir_structure: str = ""
    key_files: dict = field(default_factory=dict)
    coding_conventions: str = ""

class ContextAwareCodeAssistant:
    """基于项目上下文的智能代码助手"""

    def __init__(self, llm_client, project_root: str):
        self.llm = llm_client
        self.project_root = project_root
        self.context: Optional[ProjectContext] = None

    def init_project(self, language: str = "python"):
        """初始化项目上下文"""
        # 扫描目录结构
        dir_structure = self._scan_directory(self.project_root, max_depth=3)

        # 读取关键配置文件
        key_files = {}
        config_files = self._find_config_files(language)
        for cf in config_files:
            filepath = os.path.join(self.project_root, cf)
            if os.path.exists(filepath):
                with open(filepath, 'r', encoding='utf-8') as f:
                    content = f.read()[:2000]
                    key_files[cf] = content

        # 推断框架
        framework = self._detect_framework(key_files)

        # 提取编码规范
        conventions = self._extract_conventions()

        self.context = ProjectContext(
            language=language,
            framework=framework,
            dir_structure=dir_structure,
            key_files=key_files,
            coding_conventions=conventions
        )
        print(f"项目初始化完成: {framework}项目, 扫描到{len(dir_structure.split(chr(10)))}个目录")

    def generate_code(self, task: str, related_files: list = None) -> dict:
        """根据任务描述生成代码"""
        if not self.context:
            return {"error": "请先调用init_project初始化"}

        # 收集相关文件的代码作为参考
        ref_code = ""
        if related_files:
            ref_code = self._collect_related_code(related_files)

        # 构建带上下文的提示词
        prompt = self._build_prompt(task, ref_code)

        # 调用模型
        response = self.llm.chat.completions.create(
            model="qwen2.5-coder-7b",
            messages=[
                {"role": "system", "content": self._system_prompt()},
                {"role": "user", "content": prompt}
            ],
            temperature=0.2,
            max_tokens=3000
        )

        raw = response.choices[0].message.content
        result = self._parse_output(raw)
        result['task'] = task
        return result

    def review_code(self, code_diff: str) -> dict:
        """AI代码审查"""
        if not self.context:
            return {"error": "请先调用init_project初始化"}

        prompt = f"""请审查以下代码变更，结合项目上下文给出意见。

项目信息:
- 语言: {self.context.language}
- 框架: {self.context.framework}
- 编码规范: {self.context.coding_conventions[:500]}

代码变更(diff格式):
{code_diff}

请从以下角度审查:
1. 是否有明显的Bug或逻辑错误
2. 是否符合项目编码规范
3. 是否有安全风险
4. 是否有性能问题
5. 给出改进建议

输出格式:
- 严重程度: 高/中/低
- 问题描述
- 修改建议
"""
        response = self.llm.chat.completions.create(
            model="qwen2.5-coder-7b",
            messages=[
                {"role": "system", "content": "你是一个严格的代码审查者"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.3,
            max_tokens=2000
        )
        return {"review": response.choices[0].message.content}

    def debug_assist(self, error_log: str, source_file: str = "") -> dict:
        """基于报错日志辅助调试"""
        source_code = ""
        if source_file and os.path.exists(source_file):
            with open(source_file, 'r', encoding='utf-8') as f:
                source_code = f.read()[:3000]

        prompt = f"""请分析以下错误日志，找出可能的原因和解决方案。

错误日志:
{error_log}

相关源码:
{source_code if source_code else "无"}

项目上下文: {self.context.framework} / {self.context.language}

请输出:
1. 错误原因分析
2. 可能的修复方案（给代码）
3. 预防措施
"""
        response = self.llm.chat.completions.create(
            model="qwen2.5-coder-7b",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.3,
            max_tokens=2000
        )
        return {"analysis": response.choices[0].message.content}

    def _scan_directory(self, path: str, max_depth: int = 3, current_depth: int = 0) -> str:
        """扫描目录结构，输出树状文本"""
        if current_depth >= max_depth:
            return ""

        lines = []
        try:
            entries = sorted(os.listdir(path))
        except PermissionError:
            return ""

        for entry in entries:
            # 跳过常见忽略目录
            if entry.startswith('.') or entry in ('node_modules', '__pycache__', 'venv', 'dist', 'build'):
                continue

            full_path = os.path.join(path, entry)
            indent = "  " * current_depth

            if os.path.isdir(full_path):
                lines.append(f"{indent}{entry}/")
                sub = self._scan_directory(full_path, max_depth, current_depth + 1)
                if sub:
                    lines.append(sub)
            else:
                lines.append(f"{indent}{entry}")

        return "\n".join(lines)

    def _find_config_files(self, language: str) -> list:
        """根据语言找配置文件"""
        configs = {
            "python": ["requirements.txt", "pyproject.toml", "setup.py", ".flake8", "Makefile"],
            "javascript": ["package.json", "tsconfig.json", ".eslintrc.js", ".prettierrc"],
            "go": ["go.mod", "go.sum", "Makefile"],
            "java": ["pom.xml", "build.gradle", "settings.gradle"],
        }
        return configs.get(language, ["Makefile"])

    def _detect_framework(self, key_files: dict) -> str:
        """从配置文件推断框架"""
        for fname, content in key_files.items():
            if "requirements.txt" in fname or "pyproject.toml" in fname:
                if "django" in content.lower():
                    return "Django"
                if "fastapi" in content.lower():
                    return "FastAPI"
                if "flask" in content.lower():
                    return "Flask"
            if "package.json" in fname:
                if '"react"' in content:
                    return "React"
                if '"vue"' in content:
                    return "Vue"
        return "Unknown"

    def _extract_conventions(self) -> str:
        """提取编码规范（从已有代码模式中学习）"""
        # 简化版：从现有代码中提取命名风格
        conventions = []
        py_files = []
        for root, _, files in os.walk(self.project_root):
            for f in files:
                if f.endswith('.py'):
                    py_files.append(os.path.join(root, f))
            if len(py_files) > 10:
                break

        if py_files:
            sample_file = py_files[0]
            with open(sample_file, 'r', encoding='utf-8') as f:
                sample = f.read()[:1000]
            if re.search(r'def [a-z_]+\(', sample):
                conventions.append("函数命名: snake_case")
            if re.search(r'class [A-Z]\w+', sample):
                conventions.append("类命名: PascalCase")
            if '"""' in sample or "'''" in sample:
                conventions.append("文档字符串: 三引号")
            if 'import' in sample:
                conventions.append("使用标准import")

        return "; ".join(conventions) if conventions else "未检测到明确规范"

    def _collect_related_code(self, file_paths: list) -> str:
        """收集相关文件的代码片段"""
        snippets = []
        for fp in file_paths:
            full_path = os.path.join(self.project_root, fp) if not os.path.isabs(fp) else fp
            if os.path.exists(full_path):
                with open(full_path, 'r', encoding='utf-8') as f:
                    content = f.read()[:2000]
                    snippets.append(f"--- {fp} ---\n{content}")
        return "\n\n".join(snippets)

    def _system_prompt(self) -> str:
        return (
            "你是一个项目代码助手，基于项目上下文生成代码。"
            "生成的代码必须符合项目编码规范和框架风格。"
            "输出格式: 先给代码，再给简要说明。"
        )

    def _build_prompt(self, task: str, ref_code: str) -> str:
        prompt = f"""请根据以下项目上下文完成代码任务。

项目上下文:
- 语言: {self.context.language}
- 框架: {self.context.framework}
- 编码规范: {self.context.coding_conventions}

项目目录结构:
{self.context.dir_structure[:1500]}

参考代码:
{ref_code if ref_code else "无"}

任务: {task}
"""
        return prompt

    def _parse_output(self, raw: str) -> dict:
        """解析LLM输出"""
        code = ""
        explanation = ""

        # 提取代码块
        bt = chr(96) * 3
        code_pattern = re.escape(bt) + r'(\w+)?\n(.*?)' + re.escape(bt)
        matches = re.findall(code_pattern, raw, re.DOTALL)
        if matches:
            code = "\n\n".join(m[1] for m in matches)
            # 代码块之外的部分作为说明
            no_code = re.sub(code_pattern, '', raw, flags=re.DOTALL)
            explanation = no_code.strip()
        else:
            explanation = raw

        return {
            "code": code,
            "explanation": explanation
        }


# 使用示例
if __name__ == '__main__':
    from openai import OpenAI

    client = OpenAI(base_url="http://localhost:8000/v1", api_key="local")
    assistant = ContextAwareCodeAssistant(client, project_root="./my-project")

    # 第一步: 初始化项目上下文
    assistant.init_project(language="python")

    # 第二步: 生成代码（会带上项目上下文）
    result = assistant.generate_code(
        task="写一个用户注册接口，包含邮箱格式校验、密码强度检查、重复注册检测",
        related_files=["src/models/user.py", "src/utils/validators.py"]
    )
    print("生成的代码:")
    print(result.get('code', '')[:500])
    print("\n说明:")
    print(result.get('explanation', '')[:300])
```

```python
import os
import re
import json
from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class Document:
    """文档数据结构"""
    doc_id: str
    title: str
    content: str
    source: str = ""
    category: str = ""
    created_at: str = ""

@dataclass
class SearchResult:
    """检索结果"""
    doc_id: str
    title: str
    content: str
    score: float
    source: str

class KnowledgeBaseQA:
    """企业知识库问答系统"""

    def __init__(self, llm_client, embedding_client=None):
        self.llm = llm_client
        self.embedding = embedding_client
        self.documents: list[Document] = []
        self.vectors: dict[str, list[float]] = {}
        # 关键词索引（降级方案：没有向量模型时用关键词匹配）
        self.keyword_index: dict[str, list[str]] = {}

    def add_documents(self, docs: list[dict]):
        """批量导入文档"""
        for d in docs:
            doc = Document(
                doc_id=d.get('id', f"doc_{len(self.documents)}"),
                title=d.get('title', ''),
                content=d.get('content', ''),
                source=d.get('source', ''),
                category=d.get('category', ''),
                created_at=d.get('created_at', datetime.now().strftime('%Y-%m-%d'))
            )
            self.documents.append(doc)

            # 建立关键词索引
            keywords = self._extract_keywords(doc.title + " " + doc.content)
            for kw in keywords:
                if kw not in self.keyword_index:
                    self.keyword_index[kw] = []
                self.keyword_index[kw].append(doc.doc_id)

            # 生成向量（如果有embedding模型）
            if self.embedding:
                text = doc.title + " " + doc.content[:500]
                vec = self._get_embedding(text)
                if vec:
                    self.vectors[doc.doc_id] = vec

        print(f"已导入{len(docs)}篇文档, 总计{len(self.documents)}篇")

    def ask(self, question: str, top_k: int = 3) -> dict:
        """问答主流程"""
        # 第一步: 检索相关文档
        results = self._search(question, top_k)

        if not results:
            return {
                "question": question,
                "answer": "抱歉，知识库中没有找到相关内容。",
                "sources": [],
                "confidence": 0
            }

        # 第二步: 构建上下文
        context = self._build_context(results)

        # 第三步: 调用LLM生成回答
        prompt = self._build_qa_prompt(question, context)

        response = self.llm.chat.completions.create(
            model="qwen2.5-7b",
            messages=[
                {"role": "system", "content": "你是企业知识库助手。只基于提供的上下文回答，不要编造。如果上下文中没有相关信息，明确说不知道。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.3,
            max_tokens=1500
        )

        answer = response.choices[0].message.content

        # 第四步: 提取引用来源
        sources = [{"title": r.title, "source": r.source} for r in results]

        # 置信度评估
        avg_score = sum(r.score for r in results) / len(results)
        confidence = "高" if avg_score > 0.75 else ("中" if avg_score > 0.5 else "低")

        return {
            "question": question,
            "answer": answer,
            "sources": sources,
            "confidence": confidence,
            "retrieved_docs": len(results)
        }

    def _search(self, query: str, top_k: int = 3) -> list[SearchResult]:
        """混合检索: 向量检索 + 关键词检索"""
        results = []

        # 向量检索（如果有embedding模型）
        vec_results = {}
        if self.embedding and self.vectors:
            query_vec = self._get_embedding(query)
            if query_vec:
                for doc_id, doc_vec in self.vectors.items():
                    score = self._cosine_sim(query_vec, doc_vec)
                    vec_results[doc_id] = score

        # 关键词检索
        kw_results = {}
        query_keywords = self._extract_keywords(query)
        for kw in query_keywords:
            if kw in self.keyword_index:
                for doc_id in self.keyword_index[kw]:
                    kw_results[doc_id] = kw_results.get(doc_id, 0) + 1

        # 融合排序（简单的加权融合）
        all_doc_ids = set(list(vec_results.keys()) + list(kw_results.keys()))
        scored = []
        for doc_id in all_doc_ids:
            vec_score = vec_results.get(doc_id, 0)
            kw_score = min(kw_results.get(doc_id, 0) / max(len(query_keywords), 1), 1.0)
            # 向量权重0.7，关键词权重0.3
            final_score = vec_score * 0.7 + kw_score * 0.3
            scored.append((doc_id, final_score))

        scored.sort(key=lambda x: x[1], reverse=True)

        # 取top_k
        doc_map = {d.doc_id: d for d in self.documents}
        for doc_id, score in scored[:top_k]:
            if doc_id in doc_map:
                doc = doc_map[doc_id]
                results.append(SearchResult(
                    doc_id=doc.doc_id,
                    title=doc.title,
                    content=doc.content[:500],
                    score=score,
                    source=doc.source
                ))

        return results

    def _extract_keywords(self, text: str) -> list[str]:
        """简易关键词提取"""
        # 去标点和特殊字符
        cleaned = re.sub(r'[^\w\u4e00-\u9fff]', ' ', text)
        words = cleaned.split()

        # 过滤停用词
        stop_words = {'的', '了', '在', '是', '我', '有', '和', '就', '不', '人', '都', '一', '上', '也', '很', '到', '说', '要', '去', 'the', 'a', 'an', 'is', 'are', 'what', 'how', 'to', 'for', 'of', 'in', 'on'}
        keywords = [w.lower() for w in words if len(w) > 1 and w.lower() not in stop_words]

        return keywords[:10]

    def _get_embedding(self, text: str) -> list[float]:
        """获取文本向量"""
        try:
            response = self.embedding.embeddings.create(
                model="bge-large-zh",
                input=text[:500]
            )
            return response.data[0].embedding
        except Exception:
            return []

    def _cosine_sim(self, vec1: list[float], vec2: list[float]) -> float:
        """计算余弦相似度"""
        if len(vec1) != len(vec2) or not vec1:
            return 0.0
        dot = sum(a * b for a, b in zip(vec1, vec2))
        norm1 = sum(a * a for a in vec1) ** 0.5
        norm2 = sum(b * b for b in vec2) ** 0.5
        if norm1 == 0 or norm2 == 0:
            return 0.0
        return dot / (norm1 * norm2)

    def _build_context(self, results: list[SearchResult]) -> str:
        """构建提供给LLM的上下文"""
        parts = []
        for i, r in enumerate(results, 1):
            parts.append(f"[文档{i}] 标题: {r.title}\n来源: {r.source}\n内容: {r.content}")
        return "\n\n".join(parts)

    def _build_qa_prompt(self, question: str, context: str) -> str:
        return f"""请根据以下文档内容回答问题。

{context}

问题: {question}

要求:
1. 只基于上面提供的文档内容回答，不要编造
2. 如果文档中没有相关信息，明确说"知识库中暂无相关内容"
3. 回答时标注信息来源（如"根据文档1"）
4. 回答要简洁明了，不要啰嗦
"""


# 使用示例
if __name__ == '__main__':
    from openai import OpenAI

    client = OpenAI(base_url="http://localhost:8000/v1", api_key="local")
    qa = KnowledgeBaseQA(llm_client=client, embedding_client=client)

    # 导入文档（实际使用时从wiki/API等读取）
    docs = [
        {"id": "d1", "title": "请假流程说明", "content": "员工请假需要提前在OA系统提交申请，直属领导审批后生效。事假需提前3天申请，病假需提供医院证明。年假天数根据工龄计算：满1年5天，满3年10天，满5年15天。", "source": "HR-Wiki", "category": "人事"},
        {"id": "d2", "title": "报销规范", "content": "差旅报销需附发票原件，餐饮报销上限每人每餐50元。报销单在每月25号前提交财务，超出时间延至下月。交通费报销需标注出发地和目的地。", "source": "Finance-Wiki", "category": "财务"},
        {"id": "d3", "title": "服务器申请流程", "content": "新申请服务器需要先在工单系统提交申请，注明用途、配置要求、预计使用时间。运维组审批后1-2个工作日交付。测试环境服务器最多保留30天。", "source": "Ops-Wiki", "category": "运维"},
    ]
    qa.add_documents(docs)

    # 提问
    result = qa.ask("年假怎么算的？")
    print(f"问题: {result['question']}")
    print(f"回答: {result['answer']}")
    print(f"置信度: {result['confidence']}")
    print(f"来源: {result['sources']}")
```

```python
import re
from dataclasses import dataclass

@dataclass
class TableSchema:
    """表结构定义"""
    table_name: str
    description: str
    columns: list  # [{"name": "xxx", "type": "xxx", "desc": "xxx"}]

class NLQueryEngine:
    """自然语言转SQL查询引擎"""

    def __init__(self, llm_client, db_client=None):
        self.llm = llm_client
        self.db = db_client
        self.schemas: list[TableSchema] = []
        self.query_history: list[dict] = []

    def register_table(self, table_name: str, description: str, columns: list[dict]):
        """注册表结构"""
        schema = TableSchema(table_name, description, columns)
        self.schemas.append(schema)

    def query(self, natural_language: str, limit: int = 100) -> dict:
        """自然语言查询主流程"""
        # 第一步: 生成SQL
        sql = self._generate_sql(natural_language)

        # 第二步: 安全校验
        safety = self._safety_check(sql)
        if not safety['passed']:
            return {
                "question": natural_language,
                "sql": sql,
                "error": f"SQL安全校验未通过: {safety['issues']}",
                "results": []
            }

        # 第三步: 执行查询
        results = []
        if self.db:
            try:
                results = self.db.execute(sql, limit=limit)
            except Exception as e:
                return {
                    "question": natural_language,
                    "sql": sql,
                    "error": f"查询执行失败: {str(e)}",
                    "results": []
                }

        # 第四步: 用自然语言总结结果
        summary = self._summarize_results(natural_language, sql, results)

        # 记录历史
        self.query_history.append({
            "question": natural_language,
            "sql": sql,
            "row_count": len(results),
            "timestamp": None
        })

        return {
            "question": natural_language,
            "sql": sql,
            "results": results[:limit],
            "row_count": len(results),
            "summary": summary
        }

    def _generate_sql(self, question: str) -> str:
        """用LLM把自然语言转成SQL"""
        schema_desc = self._format_schema()

        prompt = f"""请将以下自然语言问题转换为SQL查询语句。

数据库表结构:
{schema_desc}

问题: {question}

要求:
1. 只生成SELECT语句，禁止INSERT/UPDATE/DELETE/DROP/ALTER
2. 使用标准SQL语法
3. 如果问题不明确，做出最合理的假设
4. 默认加LIMIT 100
5. 只输出SQL语句，不要加其他说明

SQL:
"""
        response = self.llm.chat.completions.create(
            model="qwen2.5-7b",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.1,
            max_tokens=500
        )

        sql = response.choices[0].message.content.strip()
        # 清理可能的多余标记
        bt = chr(96) * 3
        sql = sql.replace(bt + "sql", "").replace(bt, "").strip()
        return sql

    def _format_schema(self) -> str:
        """格式化表结构描述"""
        parts = []
        for s in self.schemas:
            cols = ", ".join(f"{c['name']}({c['type']}) {c.get('desc', '')}" for c in s.columns)
            parts.append(f"表: {s.table_name} - {s.description}\n  字段: {cols}")
        return "\n\n".join(parts)

    def _safety_check(self, sql: str) -> dict:
        """SQL安全校验"""
        sql_upper = sql.upper().strip()

        forbidden = ['INSERT', 'UPDATE', 'DELETE', 'DROP', 'ALTER', 'TRUNCATE', 'CREATE', 'GRANT', 'REVOKE']
        issues = []

        for kw in forbidden:
            if kw in sql_upper:
                issues.append(f"包含危险操作: {kw}")

        # 检查是否有LIMIT
        if 'LIMIT' not in sql_upper:
            issues.append("缺少LIMIT子句，可能返回过多数据")

        # 检查是否有WHERE（全表扫描风险）
        if 'WHERE' not in sql_upper and 'COUNT' not in sql_upper and 'GROUP BY' not in sql_upper:
            issues.append("缺少WHERE条件，可能全表扫描")

        return {
            "passed": len(issues) == 0,
            "issues": issues
        }

    def _summarize_results(self, question: str, sql: str, results: list) -> str:
        """用自然语言总结查询结果"""
        if not results:
            return "查询结果为空"

        # 取前5行做摘要
        sample = results[:5]
        prompt = f"""请用一句话总结以下查询结果。

原始问题: {question}
SQL: {sql}
结果行数: {len(results)}
前5行数据: {sample}

请用口语化的方式总结结果，直接说结论，不要描述过程。比如"上周总销售额为12.3万元，比上周增长15%"。
"""
        response = self.llm.chat.completions.create(
            model="qwen2.5-7b",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.3,
            max_tokens=200
        )
        return response.choices[0].message.content


# 使用示例
if __name__ == '__main__':
    from openai import OpenAI

    client = OpenAI(base_url="http://localhost:8000/v1", api_key="local")
    engine = NLQueryEngine(llm_client=client)

    # 注册表结构
    engine.register_table(
        table_name="orders",
        description="订单表",
        columns=[
            {"name": "order_id", "type": "INTEGER", "desc": "订单ID"},
            {"name": "user_id", "type": "INTEGER", "desc": "用户ID"},
            {"name": "amount", "type": "DECIMAL", "desc": "订单金额"},
            {"name": "status", "type": "VARCHAR", "desc": "订单状态: pending/paid/shipped/done"},
            {"name": "created_at", "type": "DATETIME", "desc": "创建时间"},
        ]
    )
    engine.register_table(
        table_name="users",
        description="用户表",
        columns=[
            {"name": "user_id", "type": "INTEGER", "desc": "用户ID"},
            {"name": "name", "type": "VARCHAR", "desc": "用户名"},
            {"name": "city", "type": "VARCHAR", "desc": "所在城市"},
            {"name": "register_date", "type": "DATE", "desc": "注册日期"},
        ]
    )

    # 自然语言查询
    result = engine.query("上周销售额最高的前5个城市")
    print(f"问题: {result['question']}")
    print(f"SQL: {result['sql']}")
    print(f"结果: {result['row_count']}行")
    print(f"摘要: {result['summary']}")
```

```python
import re
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from collections import defaultdict

@dataclass
class Alert:
    """告警数据结构"""
    alert_id: str
    service: str
    severity: str  # critical / warning / info
    message: str
    timestamp: datetime
    host: str = ""
    metric_value: float = 0
    threshold: float = 0
    tags: dict = field(default_factory=dict)

class IntelligentAlertSystem:
    """智能告警处理系统"""

    def __init__(self, llm_client=None):
        self.llm = llm_client
        self.alert_history: list[Alert] = []
        self.resolved_patterns: list[dict] = []  # 已知的自愈模式
        self.service_deps: dict[str, list[str]] = {}  # 服务依赖关系

    def set_dependencies(self, deps: dict):
        """设置服务依赖关系"""
        self.service_deps = deps

    def process_alert(self, alert: Alert) -> dict:
        """处理单条告警"""
        # 第一步: 重复检测
        duplicate = self._check_duplicate(alert)

        # 第二步: 自愈检测
        auto_resolved = self._check_auto_resolved(alert)

        # 第三步: 关联分析
        related = self._find_related_alerts(alert)

        if duplicate:
            return {
                "action": "merged",
                "reason": "与已有告警重复，已合并",
                "original_alert": duplicate.alert_id,
                "need_attention": False
            }

        if auto_resolved:
            return {
                "action": "auto_resolved",
                "reason": "匹配自愈模式，系统已自动恢复",
                "pattern_matched": auto_resolved,
                "need_attention": False
            }

        # 第四步: 对需要关注的告警做AI诊断
        diagnosis = {}
        if self.llm and alert.severity in ('critical', 'warning'):
            diagnosis = self._ai_diagnose(alert, related)

        self.alert_history.append(alert)

        return {
            "action": "escalate",
            "alert": {
                "id": alert.alert_id,
                "service": alert.service,
                "severity": alert.severity,
                "message": alert.message,
                "timestamp": alert.timestamp.isoformat()
            },
            "related_alerts": [a.alert_id for a in related],
            "diagnosis": diagnosis,
            "need_attention": True
        }

    def _check_duplicate(self, alert: Alert) -> Alert:
        """检查是否有重复告警（5分钟内相同服务相同信息）"""
        time_window = timedelta(minutes=5)
        for hist in reversed(self.alert_history[-50:]):
            if (alert.service == hist.service and
                alert.message == hist.message and
                abs((alert.timestamp - hist.timestamp)) < time_window):
                return hist
        return None

    def _check_auto_resolved(self, alert: Alert) -> str:
        """检查是否匹配已知的自愈模式"""
        for pattern in self.resolved_patterns:
            if (alert.service == pattern.get('service') and
                pattern.get('keyword') in alert.message):
                return pattern.get('name', 'unknown_pattern')
        return None

    def _find_related_alerts(self, alert: Alert) -> list[Alert]:
        """查找关联告警"""
        related = []
        time_window = timedelta(minutes=10)

        # 基于服务依赖关系查找
        deps = self.service_deps.get(alert.service, [])
        for hist in self.alert_history[-100:]:
            if abs(alert.timestamp - hist.timestamp) < time_window:
                if hist.service in deps or hist.service == alert.service:
                    if hist.alert_id != alert.alert_id:
                        related.append(hist)

        return related

    def _ai_diagnose(self, alert: Alert, related: list[Alert]) -> dict:
        """AI辅助故障诊断"""
        related_desc = ""
        if related:
            related_desc = "\n".join(
                f"- [{a.severity}] {a.service}: {a.message} ({a.timestamp.strftime('%H:%M')})"
                for a in related
            )

        deps_desc = ""
        if alert.service in self.service_deps:
            deps_desc = f"依赖的下游服务: {', '.join(self.service_deps[alert.service])}"

        prompt = f"""请分析以下告警，给出诊断意见。

告警信息:
- 服务: {alert.service}
- 级别: {alert.severity}
- 内容: {alert.message}
- 主机: {alert.host}
- 指标值: {alert.metric_value} (阈值: {alert.threshold})
- 时间: {alert.timestamp}

{deps_desc}

关联告警:
{related_desc if related_desc else "无"}

请输出:
1. 可能的根因分析（列出2-3个最可能的原因）
2. 建议的排查步骤
3. 是否需要立即升级处理
4. 历史上类似故障的处置参考（如果有的话）
"""
        response = self.llm.chat.completions.create(
            model="qwen2.5-7b",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.3,
            max_tokens=1000
        )
        return {"analysis": response.choices[0].message.content}


# 使用示例
if __name__ == '__main__':
    from openai import OpenAI

    client = OpenAI(base_url="http://localhost:8000/v1", api_key="local")
    system = IntelligentAlertSystem(llm_client=client)

    # 设置服务依赖关系
    system.set_dependencies({
        "order-service": ["mysql", "redis", "inventory-service"],
        "payment-service": ["mysql", "redis", "mq"],
        "user-service": ["mysql", "redis"],
    })

    # 设置已知的自愈模式
    system.resolved_patterns = [
        {"name": "redis主从切换", "service": "redis", "keyword": "connection reset"},
        {"name": "JVM GC暂停", "service": "order-service", "keyword": "GC pause"},
    ]

    # 模拟告警
    now = datetime.now()
    alert = Alert(
        alert_id="alert_001",
        service="order-service",
        severity="critical",
        message="订单创建接口P99延迟超过5秒",
        timestamp=now,
        host="prod-order-01",
        metric_value=5.2,
        threshold=2.0,
        tags={"env": "production", "region": "cn-east-1"}
    )

    result = system.process_alert(alert)
    print(f"处理动作: {result['action']}")
    print(f"需要关注: {result['need_attention']}")
    if result.get('diagnosis'):
        print(f"诊断分析:\n{result['diagnosis']['analysis'][:300]}")
```