feat：企业 IT 故障诊断助手

.gitignore

@@ -0,0 +1,7 @@
+venv/
+.pytest_cache/
+.ipynb_checkpoints/
+.vscode/
+.mypy_cache/
+.DS_Store
+02_rag/02_rag.ipynb

05_agentic_rag/homework/.env

@@ -0,0 +1,7 @@
+DEEPSEEK_API_KEY=sk-12a5c35b0ea64993a2ddb992eca04d95
+DEEPSEEK_BASE_URL=https://api.deepseek.com/v1
+TAVILY_API_KEY=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
+MILVUS_HOST=localhost
+MILVUS_PORT=19530
+SQLITE_DB_PATH=./app/data/incidents.db
+MAX_ROUNDS=3

05_agentic_rag/homework/README.md

@@ -0,0 +1,76 @@
+# 企业 IT 故障诊断助手(课后作业)
+
+基于 `Agentic_Rag` 参考项目改造的 IT 域 Agentic RAG,支持 aliyun_ssl_vpn / auth_timeout
+两类问题的多源检索、Grader 控流与综合判断。
+
+## 目录结构
+
+```
+homework/
+├── app/                      # 业务代码
+│   ├── config.py             # Settings (Pydantic + SecretStr)
+│   ├── schemas.py            # RouteDecision / Evidence / QueryResponse 等
+│   ├── sql_store.py          # SQLite 故障工单仓库
+│   ├── milvus_store.py       # Milvus 向量库 + 过滤字段
+│   ├── decision_engine.py    # Demo / DeepSeek 决策引擎
+│   ├── tools.py              # SQL / Milvus / Web 三个 Tool
+│   ├── graph.py              # LangGraph 状态机
+│   ├── service.py            # AgenticRAGService 装配
+│   ├── main.py               # FastAPI /health + /query
+│   └── cli.py                # 命令行入口
+├── data/
+│   ├── source/incidents.csv  # 故障工单原始数据(4 条)
+│   └── documents/            # 内部排障手册
+├── docs/apipost_example.md   # Apipost 调用示例 + 真实响应
+├── scripts/
+│   ├── prepare_data.py       # SQLite + Milvus 数据导入
+│   └── capture_apipost_example.py  # 复现 Apipost 示例响应的脚本
+└── tests/test_it_diagnosis.py
+```
+
+## 快速开始
+
+```bash
+# 1. 安装依赖(沿用参考项目 venv 或自行创建)
+python -m venv .venv && source .venv/bin/activate
+pip install fastapi pydantic pydantic-settings pymilvus langgraph langchain-core \
+            openai tavily-python python-dotenv pytest httpx
+
+# 2. 准备数据(SQLite + Milvus)
+python scripts/prepare_data.py
+
+# 3. 启动 API
+uvicorn app.main:app --host 0.0.0.0 --port 8000
+
+# 4. 调用
+curl -s http://localhost:8000/health
+curl -X POST http://localhost:8000/query -H "Content-Type: application/json" \
+  -d '{"query":"aliyun_ssl_vpn auth_timeout 怎么排查?","debug":true}'
+```
+
+详见 `docs/apipost_example.md`。
+
+## 测试
+
+```bash
+python -m pytest tests/test_it_diagnosis.py -v
+# 13 passed
+```
+
+6 个验收测试覆盖:
+1. `test_sql_30day_count` — SQL 聚合 30 天同类故障次数
+2. `test_sql_only_route` — 仅 SQL 路由
+3. `test_milvus_only_route` — 仅 Milvus 路由(带 `service_name` / `client_os` / `doc_type` 过滤)
+4. `test_vendor_missing_returns_partial` — 厂商公告缺失时返回 partial 且不能排除公共服务异常
+5. `test_milvus_first_call_empty_triggers_rewrite` — Milvus 首轮空时触发有限 Query Rewrite
+6. `test_three_sources_accept` — 三类证据齐备时 Grader 返回 accept
+
+补充测试覆盖 CLARIFY / REFUSE / 官方源过滤 / SQL 注入防御 / SecretStr / /health 端点。
+
+## 关键设计约束
+
+- **沙箱白名单**:只处理 `aliyun_ssl_vpn` + `auth_timeout`,其他服务名/错误类型一律 CLARIFY 或 REFUSE。
+- **官方源过滤**:Web 搜索 query 固定带 `site:help.aliyun.com/zh/vpn`,Provider 内部再二次过滤非 `help.aliyun.com` 的 URL。
+- **SQL 注入防御**:`service_name` / `error_type` / `client_os` / `days` 全部走参数化绑定,非法字符直接 ValueError。
+- **Rewrite 预算**:`max_rounds` 在 Graph 层强制 stop,模型不能越权无限重试。
+- **可观测性**:`debug=true` 时返回完整 `executed_queries` + `trace`,LangGraph 各节点状态一览无遗。

05_agentic_rag/homework/app/cli.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+import argparse
+import sys
+
+from app.config import get_settings
+from app.schemas import QueryRequest
+from app.service import AgenticRAGService
+
+
+def main() -> None:
+    # Windows 控制台可能使用 GBK；网页证据中的 € 等字符不能直接编码。
+    # 保留当前控制台编码，只将不支持的字符转义，避免输出阶段中断业务流程。
+    if hasattr(sys.stdout, "reconfigure"):
+        sys.stdout.reconfigure(errors="backslashreplace")
+
+    parser = argparse.ArgumentParser(description="Agentic RAG 命令行入口")
+    parser.add_argument("query", help="用户问题")
+    parser.add_argument("--debug", action="store_true", help="输出完整执行轨迹")
+    args = parser.parse_args()
+
+    response = AgenticRAGService(get_settings()).invoke(
+        QueryRequest(query=args.query, debug=args.debug)
+    )
+    if args.debug:
+        print("=== 实际执行的工具查询 ===")
+        for item in response.executed_queries:
+            print(
+                f"- {item['tool']}: query={item['query']!r}, "
+                f"arguments={item['arguments']}, status={item['status']}"
+            )
+        print("=== 完整响应 ===")
+    print(response.model_dump_json(indent=2))
+
+
+if __name__ == "__main__":
+    main()

05_agentic_rag/homework/app/config.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+from functools import lru_cache
+from pathlib import Path
+
+from pydantic import Field, SecretStr
+from pydantic import field_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+# 运行时通过源码位置确定项目根目录，不能依赖执行命令时的当前目录。
+# 仓库移动、CI Checkout 或其他开发者 Clone 后，该值会自动变化。
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(
+        env_file=PROJECT_ROOT / ".env",
+        env_file_encoding="utf-8",
+        extra="ignore",
+    )
+
+    app_env: str = "development"
+
+    llm_provider: str = "demo"
+    # SecretStr 会在 repr、日志和校验错误中自动隐藏真实凭证。
+    deepseek_api_key: SecretStr = SecretStr("")
+    deepseek_base_url: str = "https://api.deepseek.com"
+    deepseek_model_name: str = "deepseek-v4-flash"
+    deepseek_answer_thinking: bool = True
+
+    tavily_api_key: SecretStr = SecretStr("")
+    web_search_max_results: int = Field(default=5, ge=1, le=10)
+
+    embedding_provider: str = "hash"
+    embedding_model_name: str = (
+        "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
+    )
+    hash_embedding_dim: int = Field(default=256, ge=64, le=4096)
+
+    milvus_uri: str = "http://localhost:19530"
+    milvus_token: SecretStr = SecretStr("")
+    milvus_collection: str = "agentic_rag_docs"
+
+    sqlite_path: Path = Path("./data/workOrder.db")
+    orders_source_path: Path = Path("./data/source/incidents.csv")
+    documents_path: Path = Path("./data/documents")
+    max_retrieval_rounds: int = Field(default=2, ge=1, le=5)
+    min_evidence_score: float = Field(default=0.35, ge=0.0, le=1.0)
+
+    @field_validator(
+        "sqlite_path",
+        "orders_source_path",
+        "documents_path",
+        mode="after",
+    )
+    @classmethod
+    def resolve_repository_path(cls, value: Path) -> Path:
+        """将仓库相对路径解析为绝对路径，同时保留外部绝对路径。"""
+        if value.is_absolute():
+            return value.resolve()
+        return (PROJECT_ROOT / value).resolve()
+
+
+@lru_cache(maxsize=1)
+def get_settings() -> Settings:
+    return Settings()

05_agentic_rag/homework/app/decision_engine.py

@@ -0,0 +1,613 @@
+from __future__ import annotations
+
+import json
+import re
+from typing import Protocol
+
+from app.config import Settings
+from app.schemas import (
+    Evidence,
+    KnowledgeType,
+    PlanStep,
+    QualityGrade,
+    RetrievalPlan,
+    RouteDecision,
+    RouteName,
+    SOURCE_TO_KNOWLEDGE,
+    TaskType,
+)
+
+
+SERVICE_NAME_PATTERN = re.compile(
+    r"aliyun[_ -]?ssl[_ -]?vpn|SSL[-_ ]?VPN", re.IGNORECASE
+)
+# \b 在 Python re 与中文字符之间不构成 word boundary，因此中文部分走纯子串匹配。
+ERROR_TYPE_PATTERN = re.compile(
+    r"auth[_-]?timeout|认证超时", re.IGNORECASE
+)
+CLIENT_OS_PATTERN = re.compile(r"Windows|macOS|Linux", re.IGNORECASE)
+TIME_RANGE_PATTERN = re.compile(r"(\d+)\s*(?:天|d)\b", re.IGNORECASE)
+USER_ID_PATTERN = re.compile(r"\bU\d{4}\b", re.IGNORECASE)
+
+RESTRICTED_TERMS = (
+    "所有员工手机号",
+    "员工密码",
+    "员工薪资",
+    "导出全部",
+    "删除所有",
+    "关闭防火墙",
+    "禁用杀毒",
+)
+DIRECT_TERMS = ("改写", "润色", "翻译", "缩短", "转成")
+
+
+# 本作业固定只处理 aliyun_ssl_vpn 与 auth_timeout；不在白名单的服务/故障一律拒绝或澄清。
+ALLOWED_SERVICE = "aliyun_ssl_vpn"
+ALLOWED_ERROR_TYPE = "auth_timeout"
+VENDOR_DOMAIN = "help.aliyun.com"
+
+
+def extract_service_name(query: str) -> str:
+    """把 SSL-VPN 的所有变体（含"阿里云 SSL-VPN"、"SSL-VPN"、"aliyun_ssl_vpn"）映射到规范名。"""
+    if SERVICE_NAME_PATTERN.search(query):
+        return ALLOWED_SERVICE
+    return ""
+
+
+def extract_error_type(query: str) -> str:
+    """识别认证超时类故障并返回规范名 `auth_timeout`。"""
+    if ERROR_TYPE_PATTERN.search(query):
+        return ALLOWED_ERROR_TYPE
+    return ""
+
+
+def extract_client_os(query: str) -> str:
+    match = CLIENT_OS_PATTERN.search(query)
+    return match.group(0) if match else ""
+
+
+def extract_time_range_days(query: str) -> int:
+    match = TIME_RANGE_PATTERN.search(query)
+    if not match:
+        return 30
+    days = int(match.group(1))
+    # 与 SQL 层的 days ∈ [1,365] 对齐；超过 365 直接夹紧。
+    return max(1, min(days, 365))
+
+
+class DecisionEngine(Protocol):
+    def route(self, query: str, max_rounds: int) -> RouteDecision: ...
+
+    def plan(self, query: str, decision: RouteDecision) -> RetrievalPlan: ...
+
+    def grade(
+        self,
+        query: str,
+        decision: RouteDecision,
+        evidence: list[Evidence],
+        current_round: int,
+        max_rounds: int,
+        min_score: float,
+    ) -> QualityGrade: ...
+
+    def rewrite(self, query: str, grade: QualityGrade) -> str: ...
+
+    def answer(self, query: str, evidence: list[Evidence], partial: bool = False) -> str: ...
+
+
+class DemoDecisionEngine:
+    """确定性 IT 域演示引擎，便于无 API Key 启动、调试和自动化测试。"""
+
+    SQL_TERMS = (
+        "几次", "多少", "发生次数", "统计", "工单", "次数", "几天", "天里",
+    )
+    DOC_TERMS = (
+        "排查", "排障", "手册", "步骤", "怎么", "如何", "流程", "顺序",
+    )
+    WEB_TERMS = (
+        "官网", "厂商", "公告", "最新", "公共服务", "服务状态", "运维事件",
+        "公共异常", "服务异常", "服务公告",
+    )
+
+    def route(self, query: str, max_rounds: int) -> RouteDecision:
+        stripped = query.strip()
+
+        # 安全门控：敏感请求或被禁用的写操作一律 REFUSE。
+        if any(term in stripped for term in RESTRICTED_TERMS):
+            return RouteDecision(
+                needs_retrieval=False,
+                intent="restricted_request",
+                task_type=TaskType.UNKNOWN,
+                routes=[RouteName.REFUSE],
+                confidence="high",
+                reason_code="SECURITY_POLICY",
+                max_rounds=max_rounds,
+            )
+
+        # 文本改写类请求无需任何检索。
+        if any(term in stripped for term in DIRECT_TERMS):
+            return RouteDecision(
+                needs_retrieval=False,
+                intent="text_transformation",
+                task_type=TaskType.TEXT_TRANSFORM,
+                routes=[RouteName.DIRECT_ANSWER],
+                confidence="high",
+                reason_code="NO_EXTERNAL_KNOWLEDGE_REQUIRED",
+                max_rounds=max_rounds,
+            )
+
+        service_name = extract_service_name(stripped)
+        error_type = extract_error_type(stripped)
+        client_os = extract_client_os(stripped)
+        time_range_days = extract_time_range_days(stripped)
+
+        # 防御性检查：业务用户编号不应出现在 IT 域问题里（防止模板残留）。
+        if USER_ID_PATTERN.search(stripped):
+            return RouteDecision(
+                needs_retrieval=False,
+                intent="invalid_credential",
+                task_type=TaskType.UNKNOWN,
+                routes=[RouteName.REFUSE],
+                confidence="high",
+                reason_code="USER_ID_NOT_APPLICABLE",
+                max_rounds=max_rounds,
+            )
+
+        needs_sql = any(term in stripped for term in self.SQL_TERMS)
+        needs_docs = any(term in stripped for term in self.DOC_TERMS)
+        needs_web = any(term in stripped for term in self.WEB_TERMS)
+
+        # 任何路由都要求 service_name 与 error_type 都已显式给出；否则 CLARIFY。
+        if (needs_sql or needs_docs or needs_web) and not service_name:
+            return RouteDecision(
+                needs_retrieval=False,
+                intent="missing_service_name",
+                task_type=TaskType.UNKNOWN,
+                routes=[RouteName.CLARIFY],
+                confidence="high",
+                reason_code="MISSING_SERVICE_NAME",
+                max_rounds=max_rounds,
+            )
+        if (needs_sql or needs_docs or needs_web) and not error_type:
+            return RouteDecision(
+                needs_retrieval=False,
+                intent="missing_error_type",
+                task_type=TaskType.UNKNOWN,
+                routes=[RouteName.CLARIFY],
+                confidence="high",
+                reason_code="MISSING_ERROR_TYPE",
+                max_rounds=max_rounds,
+            )
+
+        # 路由决策：每个意图独立判断后取并集，三类需求齐备即三源并行。
+        routes: list[RouteName] = []
+        knowledge_types: list[KnowledgeType] = []
+        if needs_sql:
+            routes.append(RouteName.SQL_QUERY)
+            knowledge_types.append(KnowledgeType.STRUCTURED)
+        if needs_docs:
+            routes.append(RouteName.MILVUS_SEARCH)
+            knowledge_types.append(KnowledgeType.INTERNAL_DOC)
+        if needs_web:
+            routes.append(RouteName.WEB_SEARCH)
+            knowledge_types.append(KnowledgeType.REALTIME)
+
+        if not routes:
+            # 没有命中任何意图关键词 → 直接要求澄清，避免随机路由。
+            return RouteDecision(
+                needs_retrieval=False,
+                intent="unclear_intent",
+                task_type=TaskType.UNKNOWN,
+                routes=[RouteName.CLARIFY],
+                confidence="high",
+                reason_code="INTENT_UNCLEAR",
+                max_rounds=max_rounds,
+            )
+
+        filters = {
+            "service_name": service_name,
+            "error_type": error_type,
+            "client_os": client_os,
+            "days": time_range_days,
+        }
+        filters = {key: value for key, value in filters.items() if value}
+
+        if len(routes) >= 2:
+            return RouteDecision(
+                needs_retrieval=True,
+                intent="multi_source_diagnosis",
+                task_type=TaskType.DIAGNOSE,
+                routes=routes,
+                knowledge_types=knowledge_types,
+                requires_decomposition=True,
+                confidence="high",
+                reason_code="MULTIPLE_DATA_SOURCES_REQUIRED",
+                filters=filters,
+                max_rounds=max_rounds,
+            )
+        if needs_sql:
+            return RouteDecision(
+                needs_retrieval=True,
+                intent="incident_statistics",
+                task_type=TaskType.HISTORICAL_QUERY,
+                routes=routes,
+                knowledge_types=knowledge_types,
+                confidence="high",
+                reason_code="EXACT_AGGREGATION_REQUIRED",
+                filters=filters,
+                max_rounds=max_rounds,
+            )
+        if needs_web:
+            return RouteDecision(
+                needs_retrieval=True,
+                intent="vendor_status_check",
+                task_type=TaskType.VENDOR_STATUS,
+                routes=routes,
+                knowledge_types=knowledge_types,
+                confidence="high",
+                reason_code="CURRENT_PUBLIC_INFORMATION_REQUIRED",
+                filters=filters,
+                max_rounds=max_rounds,
+            )
+        return RouteDecision(
+            needs_retrieval=True,
+            intent="internal_doc_qa",
+            task_type=TaskType.DOCS_QA,
+            routes=routes,
+            knowledge_types=knowledge_types,
+            confidence="high" if client_os else "medium",
+            reason_code="INTERNAL_DOCUMENT_REQUIRED",
+            filters=filters,
+            max_rounds=max_rounds,
+        )
+
+    def plan(self, query: str, decision: RouteDecision) -> RetrievalPlan:
+        service_name = str(decision.filters.get("service_name", ALLOWED_SERVICE))
+        error_type = str(decision.filters.get("error_type", ALLOWED_ERROR_TYPE))
+        client_os = str(decision.filters.get("client_os", ""))
+        days = int(decision.filters.get("days", 30))
+        steps: list[PlanStep] = []
+
+        # 三类只读任务互不依赖，全部并行执行；depends_on 留空即可。
+        if RouteName.MILVUS_SEARCH in decision.routes:
+            steps.append(
+                PlanStep(
+                    id="search_troubleshooting_doc",
+                    tool=RouteName.MILVUS_SEARCH,
+                    # 直接用服务名 + 错误类型组合检索，比原始用户问题更聚焦。
+                    query=f"{service_name} {error_type} 排查步骤 {client_os}".strip(),
+                    arguments={
+                        "service_name": service_name,
+                        "client_os": client_os,
+                        "doc_type": "troubleshooting_guide",
+                        "top_k": 4,
+                    },
+                )
+            )
+        if RouteName.SQL_QUERY in decision.routes:
+            steps.append(
+                PlanStep(
+                    id="query_incident_statistics",
+                    tool=RouteName.SQL_QUERY,
+                    query=(
+                        f"统计 {service_name} 最近 {days} 天"
+                        f"{error_type} 的有效工单"
+                    ),
+                    arguments={
+                        "service_name": service_name,
+                        "error_type": error_type,
+                        "days": days,
+                        "client_os": client_os,
+                    },
+                )
+            )
+        if RouteName.WEB_SEARCH in decision.routes:
+            # 受控搜索词：固定 site 限定 + 产品名 + 故障类型 + 最新状态诉求，
+            # 不直接把用户自然语言送出去。
+            vendor_status_query = (
+                f'site:{VENDOR_DOMAIN}/zh/vpn 阿里云 SSL-VPN '
+                f'{error_type} 运维事件 最新公告'
+            )
+            steps.append(
+                PlanStep(
+                    id="search_vendor_status",
+                    tool=RouteName.WEB_SEARCH,
+                    query=vendor_status_query,
+                    arguments={"max_results": 3},
+                )
+            )
+        return RetrievalPlan(goal=query, steps=steps)
+
+    def grade(
+        self,
+        query: str,
+        decision: RouteDecision,
+        evidence: list[Evidence],
+        current_round: int,
+        max_rounds: int,
+        min_score: float,
+    ) -> QualityGrade:
+        # 先做确定性来源覆盖检查，再判断语义质量与冲突。
+        required_sources = set()
+        if RouteName.SQL_QUERY in decision.routes:
+            required_sources.add("sql")
+        if RouteName.MILVUS_SEARCH in decision.routes:
+            required_sources.add("milvus")
+        if RouteName.WEB_SEARCH in decision.routes:
+            required_sources.add("web")
+
+        available_sources = {item.source_type for item in evidence}
+        missing = sorted(required_sources - available_sources)
+
+        vector_evidence = [item for item in evidence if item.source_type == "milvus"]
+        # 至少一条向量结果得分不低于 min_score 才视为相关。
+        vector_relevant = any(
+            item.score is not None and item.score >= min_score
+            for item in vector_evidence
+        )
+        if "milvus" not in required_sources:
+            vector_relevant = True
+
+        # 厂商状态缺失时必须返回 partial：答案不能排除公共服务异常。
+        if "web" in required_sources and "web" not in available_sources:
+            return QualityGrade(
+                relevant=bool(evidence),
+                sufficient=False,
+                missing_aspects=["厂商官网状态"],
+                conflict=False,
+                recommended_action="partial",
+                reason="厂商官网状态缺失，证据不完整，必须以部分结论给出。",
+            )
+
+        # 厂商与内部文档冲突时标记冲突，但不阻塞终止。
+        conflict = self._detect_conflict(evidence)
+
+        if not missing and vector_relevant:
+            action = "accept" if not conflict else "accept"
+            return QualityGrade(
+                relevant=True,
+                sufficient=True,
+                missing_aspects=[],
+                conflict=conflict,
+                recommended_action=action,
+                reason=(
+                    "所需数据源均返回证据，且 Milvus 命中相关结果。"
+                    if not conflict
+                    else "证据完整但厂商公告与内部手册结论冲突，需在答案中并列两条路径。"
+                ),
+            )
+
+        # 缺失关键证据但还有预算 → 改写 Query 重新检索。
+        if current_round < max_rounds:
+            return QualityGrade(
+                relevant=bool(evidence),
+                sufficient=False,
+                missing_aspects=missing or ["相关排障手册"],
+                conflict=conflict,
+                recommended_action="rewrite_query",
+                reason="证据不完整，允许改写后重试。",
+            )
+
+        return QualityGrade(
+            relevant=bool(evidence),
+            sufficient=False,
+            missing_aspects=missing or ["相关排障手册"],
+            conflict=conflict,
+            recommended_action="stop",
+            reason="达到最大检索轮数，停止继续调用工具。",
+        )
+
+    @staticmethod
+    def _detect_conflict(evidence: list[Evidence]) -> bool:
+        """轻量级冲突检测：厂商公告与内部结论出现对立关键词。"""
+        has_vendor_issue = False
+        has_vendor_normal = False
+        for item in evidence:
+            if item.source_type != "web":
+                continue
+            text = item.content.lower()
+            if "服务异常" in text or "故障公告" in text or "中断" in text:
+                has_vendor_issue = True
+            if "未发现" in text or "无异常" in text or "正常" in text:
+                has_vendor_normal = True
+        # 没有任何厂商证据时不应判为冲突。
+        return has_vendor_issue and has_vendor_normal
+
+    def rewrite(self, query: str, grade: QualityGrade) -> str:
+        missing = "、".join(grade.missing_aspects) if grade.missing_aspects else "相关资料"
+        suffix = f"阿里云 SSL-VPN auth_timeout 排障手册 厂商公告 {missing}".strip()
+        return f"{query}；补充检索条件：{suffix}"
+
+    def answer(self, query: str, evidence: list[Evidence], partial: bool = False) -> str:
+        if not evidence:
+            return "未获得能够支持回答的证据。请补充服务名称或故障现象后重试。"
+
+        prefix = "当前仅能给出部分结论，公共服务状态仍待确认。" if partial else "基于已检索证据，结论如下。"
+        lines = [prefix]
+        sql_items = [item for item in evidence if item.source_type == "sql"]
+        milvus_items = [item for item in evidence if item.source_type == "milvus"]
+        web_items = [item for item in evidence if item.source_type == "web"]
+
+        index = 1
+        if sql_items:
+            for item in sql_items:
+                lines.append(f"- 历史工单：{item.content} [{index}]")
+                index += 1
+        if milvus_items:
+            for item in milvus_items:
+                lines.append(f"- 内部手册：{item.content} [{index}]")
+                index += 1
+        if web_items:
+            for item in web_items:
+                lines.append(f"- 厂商公告：{item.content} [{index}]")
+                index += 1
+
+        # 三类证据齐备时给出综合判断，让用户拿到行动指引而不是单纯堆事实。
+        # query 同时含 SQL/Milvus/Web 时,意味着用户要的是"本地 vs 公共服务"的判断,
+        # 我们就追加一句结论。
+        if (
+            not partial
+            and sql_items
+            and milvus_items
+            and web_items
+        ):
+            lines.append(
+                "- 综合判断：历史工单有可复现的本地处置记录(证书刷新/重建 Profile),"
+                "厂商侧公告显示服务运行正常,倾向本地配置问题;请优先按内部手册"
+                "步骤排查,公共服务异常可暂不升级。"
+            )
+        return "\n".join(lines)
+
+
+class DeepSeekDecisionEngine(DemoDecisionEngine):
+    """使用 DeepSeek 的 OpenAI 兼容接口完成路由、评分和答案生成。"""
+
+    def __init__(self, settings: Settings) -> None:
+        # 凭证只从 Settings 读取，并在创建 SDK 客户端时短暂解密。
+        api_key = settings.deepseek_api_key.get_secret_value()
+        if not api_key:
+            raise RuntimeError("LLM_PROVIDER=deepseek 时必须配置 DEEPSEEK_API_KEY")
+        try:
+            from langchain_openai import ChatOpenAI
+        except ImportError as exc:
+            raise RuntimeError(
+                "缺少 DeepSeek 依赖，请执行 uv sync --extra test --extra deepseek"
+            ) from exc
+
+        common_kwargs = {
+            "model": settings.deepseek_model_name,
+            "api_key": api_key,
+            "base_url": settings.deepseek_base_url,
+            "max_retries": 2,
+        }
+
+        # 最终答案可以使用 Thinking；开关由 .env 控制。
+        answer_thinking = "enabled" if settings.deepseek_answer_thinking else "disabled"
+        self.llm = ChatOpenAI(
+            **common_kwargs,
+            extra_body={"thinking": {"type": answer_thinking}},
+        )
+
+        # with_structured_output(function_calling) 会发送强制 tool_choice。
+        # DeepSeek Thinking 模式不接受该组合，因此路由与评分必须关闭 Thinking。
+        self.structured_llm = ChatOpenAI(
+            **common_kwargs,
+            temperature=0,
+            extra_body={"thinking": {"type": "disabled"}},
+        )
+        self.router = self.structured_llm.with_structured_output(
+            RouteDecision,
+            method="function_calling",
+        )
+        self.grader = self.structured_llm.with_structured_output(
+            QualityGrade,
+            method="function_calling",
+        )
+
+    def route(self, query: str, max_rounds: int) -> RouteDecision:
+        prompt = f"""
+你是企业 IT 故障诊断助手路由器。只能输出符合 Schema 的结果。
+可选路径：direct_answer、milvus_search、sql_query、web_search、clarify、refuse。
+历史工单次数、统计、影响终端使用 sql_query；
+内部排障手册使用 milvus_search；
+厂商官网最新公告/服务状态使用 web_search（必须配合 site:help.aliyun.com 限定）。
+filters 中只填写从原问题明确提取的 service_name、error_type、client_os、days。
+缺少 service_name 或 error_type 时返回 clarify；越权请求返回 refuse。
+最大检索轮数：{max_rounds}
+用户问题：{query}
+""".strip()
+        result = self.router.invoke(prompt)
+        # 结构化输出缺失时回退到确定性规则，路由不允许 None 继续走图。
+        if result is None:
+            return super().route(query, max_rounds)
+        # 关键业务槽位使用确定性解析覆盖模型结果，避免模型改写服务身份。
+        deterministic_filters = {
+            key: value
+            for key, value in {
+                "service_name": extract_service_name(query),
+                "error_type": extract_error_type(query),
+                "client_os": extract_client_os(query),
+                "days": extract_time_range_days(query),
+            }.items()
+            if value
+        }
+        result.filters = {**result.filters, **deterministic_filters}
+        if (
+            RouteName.SQL_QUERY in result.routes
+            and "service_name" not in result.filters
+        ):
+            return RouteDecision(
+                needs_retrieval=False,
+                intent="missing_service_name",
+                task_type=TaskType.UNKNOWN,
+                routes=[RouteName.CLARIFY],
+                confidence="high",
+                reason_code="MISSING_SERVICE_NAME",
+                max_rounds=max_rounds,
+            )
+        if (
+            RouteName.MILVUS_SEARCH in result.routes
+            and "error_type" not in result.filters
+        ):
+            return RouteDecision(
+                needs_retrieval=False,
+                intent="missing_error_type",
+                task_type=TaskType.UNKNOWN,
+                routes=[RouteName.CLARIFY],
+                confidence="high",
+                reason_code="MISSING_ERROR_TYPE",
+                max_rounds=max_rounds,
+            )
+        result.max_rounds = max_rounds
+        # 知识类型与数据源严格对齐，避免模型漏报。
+        result.knowledge_types = [
+            SOURCE_TO_KNOWLEDGE[route] for route in result.routes if route in SOURCE_TO_KNOWLEDGE
+        ]
+        return result
+
+    def grade(
+        self,
+        query: str,
+        decision: RouteDecision,
+        evidence: list[Evidence],
+        current_round: int,
+        max_rounds: int,
+        min_score: float,
+    ) -> QualityGrade:
+        prompt = f"""
+判断证据是否足以回答 IT 故障诊断问题。recommended_action 只能是 accept、rewrite_query、stop 或 partial。
+当前轮次：{current_round}/{max_rounds}
+问题：{query}
+路由：{decision.model_dump_json()}
+证据：{json.dumps([item.model_dump() for item in evidence], ensure_ascii=False)}
+""".strip()
+        result = self.grader.invoke(prompt)
+        if result is None:
+            return super().grade(
+                query=query,
+                decision=decision,
+                evidence=evidence,
+                current_round=current_round,
+                max_rounds=max_rounds,
+                min_score=min_score,
+            )
+        return result
+
+    def answer(self, query: str, evidence: list[Evidence], partial: bool = False) -> str:
+        prompt = f"""
+严格依据证据回答 IT 故障诊断问题，不得补充证据之外的事实。
+每个关键结论必须使用 [序号] 引用对应证据。
+若厂商公告缺失必须明确告知"无法确认公共服务状态"。
+问题：{query}
+是否为部分证据：{partial}
+证据：{json.dumps([item.model_dump() for item in evidence], ensure_ascii=False)}
+""".strip()
+        return str(self.llm.invoke(prompt).content)
+
+
+def create_decision_engine(settings: Settings) -> DecisionEngine:
+    if settings.llm_provider == "demo":
+        return DemoDecisionEngine()
+    if settings.llm_provider == "deepseek":
+        return DeepSeekDecisionEngine(settings)
+    raise ValueError(f"不支持的 LLM_PROVIDER: {settings.llm_provider}")

05_agentic_rag/homework/app/embeddings.py

@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+import hashlib
+import math
+import re
+from typing import Protocol
+
+from app.config import Settings
+
+
+class EmbeddingProvider(Protocol):
+    @property
+    def dimension(self) -> int: ...
+
+    def embed_query(self, text: str) -> list[float]: ...
+
+    def embed_documents(self, texts: list[str]) -> list[list[float]]: ...
+
+
+class HashEmbeddingProvider:
+    """无外部模型依赖的确定性向量，仅用于课程演示和自动化测试。"""
+
+    def __init__(self, dimension: int = 256) -> None:
+        self._dimension = dimension
+
+    @property
+    def dimension(self) -> int:
+        return self._dimension
+
+    @staticmethod
+    def _features(text: str) -> list[str]:
+        normalized = re.sub(r"\s+", "", text.lower())
+        chars = list(normalized)
+        bigrams = [normalized[index : index + 2] for index in range(len(normalized) - 1)]
+        words = re.findall(r"[a-z]+\d+|\d+|[a-z]+", normalized)
+        return chars + bigrams + words
+
+    def _embed(self, text: str) -> list[float]:
+        vector = [0.0] * self._dimension
+        for feature in self._features(text):
+            # 稳定哈希保证相同文本在不同进程中生成相同演示向量。
+            digest = hashlib.blake2b(feature.encode("utf-8"), digest_size=8).digest()
+            raw = int.from_bytes(digest, "big")
+            index = raw % self._dimension
+            sign = 1.0 if raw & 1 else -1.0
+            vector[index] += sign
+
+        norm = math.sqrt(sum(value * value for value in vector))
+        if norm == 0:
+            return vector
+        # 单位化后可直接使用 COSINE 距离进行课程检索演示。
+        return [value / norm for value in vector]
+
+    def embed_query(self, text: str) -> list[float]:
+        return self._embed(text)
+
+    def embed_documents(self, texts: list[str]) -> list[list[float]]:
+        return [self._embed(text) for text in texts]
+
+
+class SentenceTransformerEmbeddingProvider:
+    def __init__(self, model_name: str) -> None:
+        try:
+            from sentence_transformers import SentenceTransformer
+        except ImportError as exc:
+            raise RuntimeError(
+                "缺少 sentence-transformers，请执行 uv sync --extra embeddings"
+            ) from exc
+
+        self._model = SentenceTransformer(model_name)
+        self._dimension = self._model.get_sentence_embedding_dimension()
+
+    @property
+    def dimension(self) -> int:
+        return int(self._dimension)
+
+    def embed_query(self, text: str) -> list[float]:
+        return self._model.encode(text, normalize_embeddings=True).tolist()
+
+    def embed_documents(self, texts: list[str]) -> list[list[float]]:
+        return self._model.encode(texts, normalize_embeddings=True).tolist()
+
+
+def create_embedding_provider(settings: Settings) -> EmbeddingProvider:
+    # Provider 工厂隔离模型选择，Milvus Store 只依赖统一向量接口。
+    if settings.embedding_provider == "hash":
+        return HashEmbeddingProvider(settings.hash_embedding_dim)
+    if settings.embedding_provider == "sentence_transformers":
+        return SentenceTransformerEmbeddingProvider(settings.embedding_model_name)
+    raise ValueError(f"不支持的 EMBEDDING_PROVIDER: {settings.embedding_provider}")

05_agentic_rag/homework/app/graph.py

@@ -0,0 +1,303 @@
+from __future__ import annotations
+
+from concurrent.futures import ThreadPoolExecutor
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Literal
+
+from langgraph.graph import END, START, StateGraph
+
+from app.config import Settings
+from app.decision_engine import DecisionEngine
+from app.schemas import RouteName, ToolResult
+from app.state import AgentState
+from app.tools import SQLQueryTool, VectorSearchTool, WebSearchTool
+
+
+def trace_event(state: AgentState, node: str, detail: dict) -> list[dict]:
+    return [
+        *state.get("trace", []),
+        {
+            "node": node,
+            "at": datetime.now(timezone.utc).isoformat(),
+            "detail": detail,
+        },
+    ]
+
+
+@dataclass
+class GraphDependencies:
+    settings: Settings
+    engine: DecisionEngine
+    vector_tool: VectorSearchTool
+    sql_tool: SQLQueryTool
+    web_tool: WebSearchTool
+
+
+def build_graph(deps: GraphDependencies):
+    def normalize_query(state: AgentState) -> dict:
+        # 标准化问题：将问题中多余的空格去掉
+        normalized = " ".join(state["original_query"].strip().split())
+        return {
+            "current_query": normalized,
+            "retrieval_round": state.get("retrieval_round", 0),
+            "trace": trace_event(state, "normalize_query", {"query": normalized}), # 记录当前轨迹
+        }
+
+    def route_query(state: AgentState) -> dict:
+        decision = deps.engine.route(
+            state["current_query"], deps.settings.max_retrieval_rounds
+        )
+        return {
+            "route_decision": decision,
+            "trace": trace_event(
+                state,
+                "route_query",
+                {
+                    "intent": decision.intent,
+                    "routes": [route.value for route in decision.routes],
+                    "reason_code": decision.reason_code,
+                },
+            ),
+        }
+
+    def after_route(
+        state: AgentState,
+    ) -> Literal["clarify", "refuse", "generate", "plan"]:
+        routes = set(state["route_decision"].routes)
+        # 退出类路由优先级高于检索，防止复合输出意外触发工具。
+        if RouteName.CLARIFY in routes:
+            return "clarify"
+        if RouteName.REFUSE in routes:
+            return "refuse"
+        if RouteName.DIRECT_ANSWER in routes:
+            return "generate"
+        return "plan"
+
+    def plan_query(state: AgentState) -> dict:
+        plan = deps.engine.plan(state["current_query"], state["route_decision"])
+        return {
+            "retrieval_plan": plan,
+            "trace": trace_event(
+                state,
+                "plan_query",
+                {"steps": [step.model_dump(mode="json") for step in plan.steps]},
+            ),
+        }
+
+    def execute_step(step) -> ToolResult:
+        # Executor 只识别注册过的枚举工具；模型不能构造任意函数名。
+        if step.tool == RouteName.MILVUS_SEARCH:
+            return deps.vector_tool.invoke(
+                query=step.query,
+                policy_type=str(step.arguments.get("policy_type", "")),
+                top_k=int(step.arguments.get("top_k", 5)),
+            )
+        if step.tool == RouteName.SQL_QUERY:
+            return deps.sql_tool.invoke(
+                user_id=str(step.arguments.get("user_id", "")),
+                days=int(step.arguments.get("days", 30)),
+                product_keyword=str(step.arguments.get("product_keyword", "")),
+            )
+        if step.tool == RouteName.WEB_SEARCH:
+            requested = int(step.arguments.get("max_results", 5))
+            # 即使模型给出更大的 max_results，也不能突破程序配置上限。
+            return deps.web_tool.invoke(
+                query=step.query,
+                max_results=min(max(requested, 1), deps.settings.web_search_max_results),
+            )
+        return ToolResult(
+            status="error",
+            tool=step.tool.value,
+            error_code="UNSUPPORTED_TOOL",
+            error_message=f"未注册工具：{step.tool.value}",
+        )
+
+    def execute_plan(state: AgentState) -> dict:
+        steps = state["retrieval_plan"].steps
+        if not steps:
+            results: list[ToolResult] = []
+        elif len(steps) == 1:
+            results = [execute_step(steps[0])]
+        else:
+            # 本项目的计划均为无依赖读任务；依赖型 DAG 需按拓扑批次调度。
+            with ThreadPoolExecutor(max_workers=min(len(steps), 4)) as executor:
+                results = list(executor.map(execute_step, steps))
+
+        # ToolResult 用于控制流，Evidence 用于答案生成，二者不能混为一体。
+        evidence = [item for result in results for item in result.evidence]
+        errors = [
+            {
+                "tool": result.tool,
+                "code": result.error_code,
+                "message": result.error_message,
+            }
+            for result in results
+            if result.status == "error"
+        ]
+        # 将规划 Query、受控参数和实际结果并列记录，便于观察真实调用。
+        executed_queries = [
+            {
+                "step_id": step.id,
+                "tool": result.tool,
+                "query": step.query,
+                "arguments": step.arguments,
+                "status": result.status,
+                "latency_ms": result.latency_ms,
+            }
+            for step, result in zip(steps, results, strict=True)
+        ]
+        return {
+            "tool_results": results,
+            "executed_queries": executed_queries,
+            "evidence": evidence,
+            "errors": [*state.get("errors", []), *errors],
+            "trace": trace_event(
+                state,
+                "execute_plan",
+                {
+                    "tools": [result.tool for result in results],
+                    "statuses": [result.status for result in results],
+                    "evidence_count": len(evidence),
+                    "executed_queries": executed_queries,
+                },
+            ),
+        }
+
+    def grade_evidence(state: AgentState) -> dict:
+        current_round = state.get("retrieval_round", 0)
+        max_rounds = state["route_decision"].max_rounds
+        grade = deps.engine.grade(
+            query=state["current_query"],
+            decision=state["route_decision"],
+            evidence=state.get("evidence", []),
+            current_round=current_round,
+            max_rounds=max_rounds,
+            min_score=deps.settings.min_evidence_score,
+        )
+        # 模型只能建议 Rewrite，是否还有预算必须由程序层决定。
+        # 达到上限后强制 Stop，避免模型持续 Rewrite 导致图无限循环。
+        if (
+            grade.recommended_action == "rewrite_query"
+            and current_round >= max_rounds
+        ):
+            grade = grade.model_copy(
+                update={
+                    "sufficient": False,
+                    "recommended_action": "stop",
+                    "reason": (
+                        f"{grade.reason}；已达到最大检索轮数 {max_rounds}，"
+                        "程序层强制停止。"
+                    ),
+                }
+            )
+        return {
+            "quality_grade": grade,
+            "trace": trace_event(
+                state,
+                "grade_evidence",
+                grade.model_dump(mode="json"),
+            ),
+        }
+
+    def after_grade(state: AgentState) -> Literal["generate", "rewrite", "generate_partial"]:
+        # 条件边只消费结构化动作，不解析 grader 的自然语言 reason。
+        action = state["quality_grade"].recommended_action
+        if action == "accept":
+            return "generate"
+        if action == "rewrite_query":
+            return "rewrite"
+        return "generate_partial"
+
+    def rewrite_query(state: AgentState) -> dict:
+        rewritten = deps.engine.rewrite(
+            state["current_query"], state["quality_grade"]
+        )
+        next_round = state.get("retrieval_round", 0) + 1
+        return {
+            "current_query": rewritten,
+            "retrieval_round": next_round,
+            "trace": trace_event(
+                state,
+                "rewrite_query",
+                {"round": next_round, "rewritten_query": rewritten},
+            ),
+        }
+
+    def generate_answer(state: AgentState) -> dict:
+        answer = deps.engine.answer(
+            state["original_query"], state.get("evidence", []), partial=False
+        )
+        return {
+            "final_answer": answer,
+            "termination_reason": "evidence_accepted"
+            if state.get("evidence")
+            else "direct_answer",
+            "trace": trace_event(state, "generate_answer", {"partial": False}),
+        }
+
+    def generate_partial_answer(state: AgentState) -> dict:
+        answer = deps.engine.answer(
+            state["original_query"], state.get("evidence", []), partial=True
+        )
+        return {
+            "final_answer": answer,
+            "termination_reason": "retrieval_budget_exhausted",
+            "trace": trace_event(state, "generate_partial_answer", {"partial": True}),
+        }
+
+    def clarify(state: AgentState) -> dict:
+        return {
+            "final_answer": "请补充用户编号、商品名称或订单范围后再查询。",
+            "termination_reason": "clarification_required",
+            "trace": trace_event(state, "clarify", {}),
+        }
+
+    def refuse(state: AgentState) -> dict:
+        return {
+            "final_answer": "当前请求涉及受限数据，系统拒绝执行。",
+            "termination_reason": "security_policy",
+            "trace": trace_event(state, "refuse", {}),
+        }
+
+    builder = StateGraph(AgentState)
+    builder.add_node("normalize_query", normalize_query) # 
+    builder.add_node("route_query", route_query)
+    builder.add_node("plan_query", plan_query)
+    builder.add_node("execute_plan", execute_plan)
+    builder.add_node("grade_evidence", grade_evidence)
+    builder.add_node("rewrite_query", rewrite_query)
+    builder.add_node("generate_answer", generate_answer)
+    builder.add_node("generate_partial_answer", generate_partial_answer)
+    builder.add_node("clarify", clarify)
+    builder.add_node("refuse", refuse)
+
+    builder.add_edge(START, "normalize_query")
+    builder.add_edge("normalize_query", "route_query")
+    builder.add_conditional_edges(
+        "route_query",
+        after_route,
+        {
+            "clarify": "clarify",
+            "refuse": "refuse",
+            "generate": "generate_answer",
+            "plan": "plan_query",
+        },
+    )
+    builder.add_edge("plan_query", "execute_plan")
+    builder.add_edge("execute_plan", "grade_evidence")
+    builder.add_conditional_edges(
+        "grade_evidence",
+        after_grade,
+        {
+            "generate": "generate_answer",
+            "rewrite": "rewrite_query",
+            "generate_partial": "generate_partial_answer",
+        },
+    )
+    builder.add_edge("rewrite_query", "plan_query")
+    builder.add_edge("generate_answer", END)
+    builder.add_edge("generate_partial_answer", END)
+    builder.add_edge("clarify", END)
+    builder.add_edge("refuse", END)
+    return builder.compile()

05_agentic_rag/homework/app/main.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+from functools import lru_cache
+
+from fastapi import FastAPI
+
+from app.config import get_settings
+from app.schemas import QueryRequest, QueryResponse
+from app.service import AgenticRAGService
+
+
+app = FastAPI(
+    title="Enterprise IT Diagnosis Assistant",
+    version="1.0.0",
+)
+
+
+@lru_cache(maxsize=1)
+def get_service() -> AgenticRAGService:
+    return AgenticRAGService(get_settings())
+
+
+@app.get("/health")
+def health() -> dict[str, str]:
+    return {"status": "ok"}
+
+
+@app.post("/query", response_model=QueryResponse)
+def query(request: QueryRequest) -> QueryResponse:
+    return get_service().invoke(request)

05_agentic_rag/homework/app/milvus_store.py

@@ -0,0 +1,203 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+from pymilvus import DataType, MilvusClient
+
+from app.embeddings import EmbeddingProvider
+from app.schemas import Evidence
+
+
+@dataclass(frozen=True)
+class IndexedDocument:
+    """Milvus 入库对象，承载受控元数据用于按服务 / OS 过滤。"""
+
+    content: str
+    source: str
+    doc_type: str
+    chunk_index: int
+    policy_type: str
+    category: str  # 课程版语义为"适用类目"；本作业改用为 client_os 过滤值
+    service_name: str  # 新增字段，用于按服务过滤避免跨服务召回
+    metadata: dict[str, Any]
+
+
+class MilvusVectorStore:
+    def __init__(
+        self,
+        uri: str,
+        token: str,
+        collection_name: str,
+        embeddings: EmbeddingProvider,
+    ) -> None:
+        kwargs: dict[str, Any] = {"uri": uri}
+        if token:
+            kwargs["token"] = token
+        self.client = MilvusClient(**kwargs)
+        self.collection_name = collection_name
+        self.embeddings = embeddings
+
+    def is_ready(self) -> bool:
+        self.client.list_collections()
+        return True
+
+    def create_collection(self, recreate: bool = False) -> None:
+        exists = self.client.has_collection(collection_name=self.collection_name)
+        # 课程数据允许显式重建；生产环境应创建新 Collection 后切换 Alias。
+        if exists and recreate:
+            self.client.drop_collection(collection_name=self.collection_name)
+            exists = False
+        if exists:
+            self.client.load_collection(collection_name=self.collection_name)
+            return
+
+        schema = MilvusClient.create_schema(
+            auto_id=True,
+            enable_dynamic_field=False,
+        )
+        schema.add_field("id", DataType.INT64, is_primary=True)
+        schema.add_field("content", DataType.VARCHAR, max_length=8192)
+        schema.add_field("source", DataType.VARCHAR, max_length=1024)
+        schema.add_field("doc_type", DataType.VARCHAR, max_length=64)
+        schema.add_field("chunk_index", DataType.INT64)
+        schema.add_field("policy_type", DataType.VARCHAR, max_length=64)
+        schema.add_field("category", DataType.VARCHAR, max_length=64)
+        schema.add_field("service_name", DataType.VARCHAR, max_length=64)
+        schema.add_field("metadata", DataType.JSON)
+        schema.add_field(
+            "embedding",
+            DataType.FLOAT_VECTOR,
+            # Schema 维度必须与当前 Embedding Provider 完全一致。
+            dim=self.embeddings.dimension,
+        )
+
+        index_params = self.client.prepare_index_params()
+        index_params.add_index(
+            field_name="embedding",
+            index_name="embedding_hnsw",
+            index_type="HNSW",
+            metric_type="COSINE",
+            params={"M": 16, "efConstruction": 200},
+        )
+        # 三个反向索引覆盖常见过滤维度：政策类型 / 服务名称 / 客户端 OS。
+        # 任意写入端只能填入受控字符串，禁止任意表达式进入 Milvus。
+        index_params.add_index(
+            field_name="policy_type",
+            index_name="policy_type_inverted",
+            index_type="INVERTED",
+        )
+        index_params.add_index(
+            field_name="service_name",
+            index_name="service_name_inverted",
+            index_type="INVERTED",
+        )
+        index_params.add_index(
+            field_name="category",
+            index_name="category_inverted",
+            index_type="INVERTED",
+        )
+
+        self.client.create_collection(
+            collection_name=self.collection_name,
+            schema=schema,
+            index_params=index_params,
+        )
+        self.client.load_collection(collection_name=self.collection_name)
+
+    def insert_documents(self, documents: list[IndexedDocument]) -> int:
+        if not documents:
+            return 0
+        # 先批量生成向量，再用 strict zip 防止文档与向量静默错位。
+        vectors = self.embeddings.embed_documents([item.content for item in documents])
+        rows = []
+        for item, vector in zip(documents, vectors, strict=True):
+            rows.append(
+                {
+                    "content": item.content,
+                    "source": item.source,
+                    "doc_type": item.doc_type,
+                    "chunk_index": item.chunk_index,
+                    "policy_type": item.policy_type,
+                    "category": item.category,
+                    "service_name": item.service_name,
+                    "metadata": item.metadata,
+                    "embedding": vector,
+                }
+            )
+        self.client.insert(collection_name=self.collection_name, data=rows)
+        self.client.flush(collection_name=self.collection_name)
+        return len(rows)
+
+    @staticmethod
+    def _validate_filter_value(field: str, value: str) -> str:
+        """白名单过滤值：只允许字母数字下划线连字符，避免 Milvus 表达式注入。"""
+        if not value:
+            return ""
+        if not value.replace("_", "").replace("-", "").isalnum():
+            raise ValueError(f"{field} 非法：{value!r}")
+        return value
+
+    def search(
+        self,
+        query: str,
+        top_k: int = 5,
+        service_name: str = "",
+        client_os: str = "",
+        doc_type: str = "",
+    ) -> list[Evidence]:
+        query_vector = self.embeddings.embed_query(query)
+
+        # Filter 只由已校验的结构化字段生成，不接收任意 Milvus 表达式。
+        clauses: list[str] = []
+        normalized_service = self._validate_filter_value("service_name", service_name)
+        if normalized_service:
+            clauses.append(f'service_name == "{normalized_service}"')
+        normalized_os = self._validate_filter_value("client_os", client_os)
+        if normalized_os:
+            clauses.append(f'category == "{normalized_os}"')
+        normalized_doc_type = self._validate_filter_value("doc_type", doc_type)
+        if normalized_doc_type:
+            clauses.append(f'doc_type == "{normalized_doc_type}"')
+        filter_expression = " and ".join(clauses)
+
+        results = self.client.search(
+            collection_name=self.collection_name,
+            data=[query_vector],
+            anns_field="embedding",
+            filter=filter_expression,
+            limit=top_k,
+            output_fields=[
+                "content",
+                "source",
+                "doc_type",
+                "chunk_index",
+                "policy_type",
+                "category",
+                "service_name",
+                "metadata",
+            ],
+            search_params={"metric_type": "COSINE", "params": {"ef": 64}},
+        )
+        evidence: list[Evidence] = []
+        # Store 层统一转换 Evidence，Graph 不依赖 PyMilvus 的原始返回结构。
+        for hit in results[0]:
+            entity = hit.get("entity", {})
+            evidence.append(
+                Evidence(
+                    source_type="milvus",
+                    source=entity.get("source", "unknown"),
+                    content=entity.get("content", ""),
+                    score=float(hit.get("distance", 0.0)),
+                    metadata={
+                        "id": hit.get("id"),
+                        "doc_type": entity.get("doc_type"),
+                        "chunk_index": entity.get("chunk_index"),
+                        "policy_type": entity.get("policy_type"),
+                        "category": entity.get("category"),
+                        "service_name": entity.get("service_name"),
+                        **(entity.get("metadata") or {}),
+                    },
+                )
+            )
+        return evidence

05_agentic_rag/homework/app/schemas.py

@@ -0,0 +1,114 @@
+from __future__ import annotations
+
+from enum import StrEnum
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+
+class RouteName(StrEnum):
+    DIRECT_ANSWER = "direct_answer"
+    MILVUS_SEARCH = "milvus_search"
+    SQL_QUERY = "sql_query"
+    WEB_SEARCH = "web_search"
+    MULTI_SOURCE = "multi_source"
+    CLARIFY = "clarify"
+    REFUSE = "refuse"
+
+
+class KnowledgeType(StrEnum):
+    """每条证据所属的知识来源类型，用于路由决策与 Grader 来源覆盖检查。"""
+
+    STRUCTURED = "structured"  # SQLite / 工单结构化数据
+    INTERNAL_DOC = "internal_doc"  # Milvus / 内部排障手册
+    REALTIME = "realtime"  # Web / 厂商官网最新公告
+
+
+class TaskType(StrEnum):
+    """用户问题的高层意图分类，与路由选择对齐。"""
+
+    DIAGNOSE = "diagnose"  # 综合诊断：需要三源证据
+    HISTORICAL_QUERY = "historical_query"  # 历史工单统计
+    DOCS_QA = "docs_qa"  # 内部手册问答
+    VENDOR_STATUS = "vendor_status"  # 厂商当前状态
+    TEXT_TRANSFORM = "text_transform"  # 文本改写
+    UNKNOWN = "unknown"
+
+
+# 把数据源到 KnowledgeType 的映射固化在 schemas 层，避免下游再硬编码字符串。
+SOURCE_TO_KNOWLEDGE: dict[RouteName, KnowledgeType] = {
+    RouteName.SQL_QUERY: KnowledgeType.STRUCTURED,
+    RouteName.MILVUS_SEARCH: KnowledgeType.INTERNAL_DOC,
+    RouteName.WEB_SEARCH: KnowledgeType.REALTIME,
+}
+
+
+class RouteDecision(BaseModel):
+    needs_retrieval: bool
+    intent: str
+    task_type: TaskType = TaskType.UNKNOWN
+    routes: list[RouteName] = Field(default_factory=list)
+    knowledge_types: list[KnowledgeType] = Field(default_factory=list)
+    requires_decomposition: bool = False
+    confidence: str = "medium"
+    reason_code: str
+    filters: dict[str, Any] = Field(default_factory=dict)
+    max_rounds: int = 2
+    fallback: str = "clarify"
+
+
+class PlanStep(BaseModel):
+    id: str
+    tool: RouteName
+    query: str
+    arguments: dict[str, Any] = Field(default_factory=dict)
+    depends_on: list[str] = Field(default_factory=list)
+
+
+class RetrievalPlan(BaseModel):
+    goal: str
+    steps: list[PlanStep]
+
+
+class Evidence(BaseModel):
+    source_type: str
+    source: str
+    content: str
+    score: float | None = None
+    knowledge_type: KnowledgeType | None = None
+    metadata: dict[str, Any] = Field(default_factory=dict)
+
+
+class ToolResult(BaseModel):
+    status: str
+    tool: str
+    data: list[dict[str, Any]] = Field(default_factory=list)
+    evidence: list[Evidence] = Field(default_factory=list)
+    latency_ms: int = 0
+    retryable: bool = False
+    error_code: str | None = None
+    error_message: str | None = None
+
+
+class QualityGrade(BaseModel):
+    relevant: bool
+    sufficient: bool
+    missing_aspects: list[str] = Field(default_factory=list)
+    conflict: bool = False
+    recommended_action: str
+    reason: str
+
+
+class QueryRequest(BaseModel):
+    query: str = Field(min_length=1, max_length=4000)
+    session_id: str = "default"
+    debug: bool = False
+
+
+class QueryResponse(BaseModel):
+    answer: str
+    citations: list[Evidence]
+    route: RouteDecision
+    executed_queries: list[dict[str, Any]] = Field(default_factory=list)
+    trace: list[dict[str, Any]] = Field(default_factory=list)
+    termination_reason: str

05_agentic_rag/homework/app/service.py

@@ -0,0 +1,86 @@
+from __future__ import annotations
+
+from app.config import Settings
+from app.decision_engine import create_decision_engine
+from app.embeddings import create_embedding_provider
+from app.graph import GraphDependencies, build_graph
+from app.milvus_store import MilvusVectorStore
+from app.schemas import QueryRequest, QueryResponse, RouteDecision
+from app.sql_store import IncidentRepository
+from app.tools import (
+    SQLQueryTool,
+    TavilyWebSearchProvider,
+    VectorSearchTool,
+    VectorStore,
+    WebSearchProvider,
+    WebSearchTool,
+)
+
+
+class AgenticRAGService:
+    """企业 IT 故障诊断服务：组装 Embedding / Milvus / SQLite / Web Provider。
+
+    测试可注入 Fake Store 与 Fake Web Provider，生产路径才创建真实 Milvus 与 Tavily 客户端。
+    """
+
+    def __init__(
+        self,
+        settings: Settings,
+        vector_store: VectorStore | None = None,
+        web_search_provider: WebSearchProvider | None = None,
+    ) -> None:
+        self.settings = settings
+        if vector_store is None:
+            embeddings = create_embedding_provider(settings)
+            vector_store = MilvusVectorStore(
+                uri=settings.milvus_uri,
+                token=settings.milvus_token.get_secret_value(),
+                collection_name=settings.milvus_collection,
+                embeddings=embeddings,
+            )
+        if web_search_provider is None:
+            # Tavily Provider 延迟到真正调用 web_search 时才校验 Key 和依赖。
+            web_search_provider = TavilyWebSearchProvider(
+                settings.tavily_api_key.get_secret_value()
+            )
+        dependencies = GraphDependencies(
+            settings=settings,
+            engine=create_decision_engine(settings),
+            vector_tool=VectorSearchTool(vector_store),
+            sql_tool=SQLQueryTool(IncidentRepository(settings.sqlite_path)),
+            web_tool=WebSearchTool(web_search_provider),
+        )
+        self.graph = build_graph(dependencies)
+
+    def invoke(self, request: QueryRequest) -> QueryResponse:
+        # 每次请求创建独立初始状态，避免跨会话共享 Evidence 或错误信息。
+        state = self.graph.invoke(
+            {
+                "original_query": request.query,
+                "current_query": request.query,
+                "session_id": request.session_id,
+                "debug": request.debug,
+                "evidence": [],
+                "tool_results": [],
+                "executed_queries": [],
+                "retrieval_round": 0,
+                "errors": [],
+                "trace": [],
+            },
+            config={"recursion_limit": 20},
+        )
+        route = state.get("route_decision")
+        if route is None:
+            route = RouteDecision(
+                needs_retrieval=False,
+                intent="internal_error",
+                reason_code="MISSING_ROUTE_DECISION",
+            )
+        return QueryResponse(
+            answer=state.get("final_answer", "系统未生成答案。"),
+            citations=state.get("evidence", []),
+            route=route,
+            executed_queries=state.get("executed_queries", []) if request.debug else [],
+            trace=state.get("trace", []) if request.debug else [],
+            termination_reason=state.get("termination_reason", "unknown"),
+        )

05_agentic_rag/homework/app/sql_store.py

@@ -0,0 +1,275 @@
+from __future__ import annotations
+
+import csv
+import re
+import sqlite3
+from datetime import date, timedelta
+from pathlib import Path
+
+from app.config import PROJECT_ROOT
+from app.schemas import Evidence
+
+
+SCHEMA_SQL = """
+CREATE TABLE IF NOT EXISTS incidents (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    occurred_at TEXT NOT NULL,
+    incident_id TEXT NOT NULL UNIQUE,
+    service_name TEXT NOT NULL,
+    error_type TEXT NOT NULL,
+    client_os TEXT NOT NULL,
+    status TEXT NOT NULL,
+    resolution TEXT NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_incidents_service_time
+ON incidents(service_name, occurred_at);
+CREATE INDEX IF NOT EXISTS idx_incidents_resolution
+ON incidents(resolution);
+CREATE INDEX IF NOT EXISTS idx_incidents_error_type
+ON incidents(error_type);
+"""
+
+DEFAULT_SOURCE_PATH = PROJECT_ROOT / "data" / "source" / "incidents.csv"
+INCIDENT_ID_PATTERN = re.compile(r"^INC\d+$")
+SERVICE_NAME_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{1,63}$")
+ERROR_TYPE_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_]{1,63}$")
+CLIENT_OS_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9 ._/+-]{1,63}$")
+ALLOWED_STATUSES = {"resolved", "open", "in_progress", "pending", "closed", "escalated"}
+ACTIVE_STATUSES = ("open", "in_progress", "resolved")
+
+
+def load_incidents(
+    source_path: Path,
+) -> list[tuple[date, str, str, str, str, str, str]]:
+    """从脱敏 CSV 读取工单记录并转换为 (date, ...) 元组。
+
+    返回字段顺序与 `SCHEMA_SQL` 列顺序一致（除自增 id），
+    便于 `initialize_database` 直接复用 executemany。
+    """
+    if not source_path.exists():
+        raise FileNotFoundError(f"SQLite 原始工单不存在：{source_path}")
+
+    # 相对天数在导入时转换，保证"最近 30 天"样例不会随课程日期过期。
+    today = date.today()
+    rows: list[tuple[date, str, str, str, str, str, str]] = []
+    with source_path.open("r", encoding="utf-8-sig", newline="") as file:
+        reader = csv.DictReader(file)
+        required_fields = {
+            "days_ago",
+            "incident_id",
+            "service_name",
+            "error_type",
+            "client_os",
+            "status",
+            "resolution",
+        }
+        if not reader.fieldnames or not required_fields <= set(reader.fieldnames):
+            raise ValueError("incidents.csv 缺少必需字段")
+
+        for line_number, item in enumerate(reader, start=2):
+            try:
+                days_ago = int(item["days_ago"])
+            except (TypeError, ValueError) as exc:
+                raise ValueError(f"第 {line_number} 行 days_ago 非法") from exc
+            if not 0 <= days_ago <= 3650:
+                raise ValueError(f"第 {line_number} 行 days_ago 超出范围")
+
+            incident_id = item["incident_id"].strip().upper()
+            service_name = item["service_name"].strip()
+            error_type = item["error_type"].strip()
+            client_os = item["client_os"].strip()
+            status = item["status"].strip().lower()
+            resolution = item["resolution"].strip()
+
+            if not INCIDENT_ID_PATTERN.fullmatch(incident_id):
+                raise ValueError(f"第 {line_number} 行 incident_id 非法")
+            if not SERVICE_NAME_PATTERN.fullmatch(service_name):
+                raise ValueError(f"第 {line_number} 行 service_name 非法")
+            if not ERROR_TYPE_PATTERN.fullmatch(error_type):
+                raise ValueError(f"第 {line_number} 行 error_type 非法")
+            if not CLIENT_OS_PATTERN.fullmatch(client_os):
+                raise ValueError(f"第 {line_number} 行 client_os 非法")
+            if not resolution:
+                raise ValueError(f"第 {line_number} 行 resolution 为空")
+            if len(resolution) > 64:
+                raise ValueError(f"第 {line_number} 行 resolution 过长")
+            if status not in ALLOWED_STATUSES:
+                raise ValueError(f"第 {line_number} 行 status 非法")
+
+            rows.append(
+                (
+                    today - timedelta(days=days_ago),
+                    incident_id,
+                    service_name,
+                    error_type,
+                    client_os,
+                    status,
+                    resolution,
+                )
+            )
+
+    if not rows:
+        raise ValueError("incidents.csv 没有数据")
+    return rows
+
+
+def initialize_database(
+    path: Path,
+    source_path: Path | None = None,
+    reset: bool = False,
+) -> int:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    if reset and path.exists():
+        path.unlink()
+
+    # CSV 是可审计真源，SQLite 只负责课程中的精确过滤和聚合。
+    rows = load_incidents(source_path or DEFAULT_SOURCE_PATH)
+    with sqlite3.connect(path) as connection:
+        connection.executescript(SCHEMA_SQL)
+        current = connection.execute("SELECT COUNT(*) FROM incidents").fetchone()[0]
+        if current == 0:
+            connection.executemany(
+                """
+                INSERT INTO incidents(
+                    occurred_at, incident_id, service_name,
+                    error_type, client_os, status, resolution
+                )
+                VALUES (?, ?, ?, ?, ?, ?, ?)
+                """,
+                [
+                    (
+                        item[0].isoformat(),
+                        item[1],
+                        item[2],
+                        item[3],
+                        item[4],
+                        item[5],
+                        item[6],
+                    )
+                    for item in rows
+                ],
+            )
+        return connection.execute("SELECT COUNT(*) FROM incidents").fetchone()[0]
+
+
+class IncidentRepository:
+    """企业 IT 工单聚合查询，统一返回结构化结果与可引用 Evidence。"""
+
+    def __init__(self, path: Path) -> None:
+        self.path = path
+
+    def summarize_recent(
+        self,
+        service_name: str,
+        days: int = 30,
+        error_type: str = "",
+        client_os: str = "",
+    ) -> tuple[dict, Evidence]:
+        if not self.path.exists():
+            raise RuntimeError(
+                "业务数据库不存在，请先执行 uv run python scripts/prepare_data.py"
+            )
+        normalized_service = service_name.strip()
+        if not SERVICE_NAME_PATTERN.fullmatch(normalized_service):
+            raise ValueError("service_name 只能包含字母数字、下划线和连字符")
+        if not 1 <= days <= 365:
+            raise ValueError("days 必须在 1 到 365 之间")
+
+        normalized_error = error_type.strip()
+        if normalized_error and not ERROR_TYPE_PATTERN.fullmatch(normalized_error):
+            raise ValueError("error_type 只能包含字母数字和下划线")
+        if len(normalized_error) > 64:
+            raise ValueError("error_type 过长")
+
+        normalized_os = client_os.strip()
+        if normalized_os and not CLIENT_OS_PATTERN.fullmatch(normalized_os):
+            raise ValueError("client_os 非法")
+        if len(normalized_os) > 64:
+            raise ValueError("client_os 过长")
+
+        start_date = (date.today() - timedelta(days=days)).isoformat()
+
+        # SQL 模板固定并使用占位符；模型不能生成 SQL 或拼接 WHERE 表达式。
+        # 单次查询按 resolution 分组返回，每组自带该组的 incident_ids 与 resolved 数；
+        # Python 端再做二次聚合得到总数 / 已解决 / 仍需跟进，避免两次往返。
+        sql = """
+            SELECT resolution,
+                   COUNT(*) AS count,
+                   SUM(CASE WHEN status = 'resolved' THEN 1 ELSE 0 END) AS resolved_count,
+                   GROUP_CONCAT(incident_id) AS incident_ids
+            FROM incidents
+            WHERE service_name = ?
+              AND occurred_at >= ?
+              AND status IN (?, ?, ?)
+              AND (? = '' OR error_type = ?)
+              AND (? = '' OR client_os = ?)
+            GROUP BY resolution
+            ORDER BY count DESC, resolution ASC
+        """
+        parameters = (
+            normalized_service,
+            start_date,
+            *ACTIVE_STATUSES,
+            normalized_error,
+            normalized_error,
+            normalized_os,
+            normalized_os,
+        )
+        with sqlite3.connect(self.path) as connection:
+            connection.row_factory = sqlite3.Row
+            rows = connection.execute(sql, parameters).fetchall()
+
+        incident_ids: list[str] = []
+        resolution_counts: dict[str, int] = {}
+        total_count = 0
+        resolved_count = 0
+        for row in rows:
+            count = int(row["count"])
+            total_count += count
+            resolved_count += int(row["resolved_count"] or 0)
+            resolution = str(row["resolution"])
+            resolution_counts[resolution] = count
+            if row["incident_ids"]:
+                incident_ids.extend(row["incident_ids"].split(","))
+
+        # 仍需跟进：在已过滤 ACTIVE_STATUSES 之内，排除已解决即为仍需跟进。
+        open_count = total_count - resolved_count
+
+        result = {
+            "service_name": normalized_service,
+            "days": days,
+            "error_type": normalized_error,
+            "client_os": normalized_os,
+            "incident_count": total_count,
+            "resolved_count": resolved_count,
+            "open_count": open_count,
+            "resolution_counts": resolution_counts,
+            "incident_ids": incident_ids,
+        }
+
+        scope_clauses: list[str] = []
+        if normalized_error:
+            scope_clauses.append(f"错误类型为“{normalized_error}”")
+        if normalized_os:
+            scope_clauses.append(f"客户端为“{normalized_os}”")
+        scope_clause = (
+            f"、{' 且 '.join(scope_clauses)}的" if scope_clauses else ""
+        )
+        if resolution_counts:
+            resolution_summary = "、".join(
+                f"{name}({count} 次)" for name, count in resolution_counts.items()
+            )
+        else:
+            resolution_summary = "暂无解决记录"
+        evidence = Evidence(
+            source_type="sql",
+            source="incidents",
+            content=(
+                f"服务 {normalized_service} 最近 {days} 天{scope_clause}工单共 "
+                f"{total_count} 起，其中已解决 {resolved_count} 起、"
+                f"仍需跟进 {open_count} 起；"
+                f"采取的解决方案包括：{resolution_summary}。"
+            ),
+            metadata={"query_name": "summarize_recent_incidents", **result},
+        )
+        return result, evidence

05_agentic_rag/homework/app/state.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+from typing_extensions import TypedDict
+
+from app.schemas import Evidence, QualityGrade, RetrievalPlan, RouteDecision, ToolResult
+
+
+class AgentState(TypedDict, total=False):
+    original_query: str
+    current_query: str
+    session_id: str
+    debug: bool
+    route_decision: RouteDecision
+    retrieval_plan: RetrievalPlan
+    evidence: list[Evidence]
+    tool_results: list[ToolResult]
+    executed_queries: list[dict]
+    quality_grade: QualityGrade
+    retrieval_round: int
+    errors: list[dict]
+    trace: list[dict]
+    final_answer: str
+    termination_reason: str

05_agentic_rag/homework/app/tools.py

@@ -0,0 +1,212 @@
+from __future__ import annotations
+
+from time import perf_counter
+from typing import Protocol
+from urllib.parse import urlparse
+
+from app.schemas import Evidence, KnowledgeType, ToolResult
+from app.sql_store import IncidentRepository
+
+
+class VectorStore(Protocol):
+    def search(
+        self,
+        query: str,
+        top_k: int = 5,
+        service_name: str = "",
+        client_os: str = "",
+        doc_type: str = "",
+    ) -> list[Evidence]: ...
+
+
+class WebSearchProvider(Protocol):
+    def search(self, query: str, max_results: int = 5) -> list[Evidence]: ...
+
+
+# 作业约定的官方资料域名：只接受阿里云帮助中心，其它来源一律丢弃。
+ALLOWED_VENDOR_HOST = "help.aliyun.com"
+
+
+def is_official_vendor_source(source: str) -> bool:
+    """只接受阿里云帮助中心域名，避免论坛与聚合站污染厂商状态结论。"""
+    hostname = (urlparse(source).hostname or "").lower()
+    return hostname == ALLOWED_VENDOR_HOST
+
+
+class TavilyWebSearchProvider:
+    """把 Tavily 搜索结果转换为项目统一的 Evidence，并过滤到官方域名。"""
+
+    def __init__(self, api_key: str) -> None:
+        self.api_key = api_key
+
+    def search(self, query: str, max_results: int = 5) -> list[Evidence]:
+        if not self.api_key:
+            raise RuntimeError("网页搜索未配置 TAVILY_API_KEY")
+        try:
+            from tavily import TavilyClient
+        except ImportError as exc:
+            raise RuntimeError(
+                "缺少网页搜索依赖，请执行 uv sync --extra test --extra web"
+            ) from exc
+
+        client = TavilyClient(api_key=self.api_key)
+        response = client.search(
+            query=query,
+            search_depth="basic",
+            topic="general",
+            max_results=max_results,
+            include_answer=False,
+            include_raw_content=False,
+        )
+
+        # 只保留可引用片段和官方域名来源，不把搜索服务生成的 answer 当作事实答案。
+        evidence: list[Evidence] = []
+        for item in response.get("results", []):
+            url = str(item.get("url", "")).strip()
+            content = str(item.get("content", "")).strip()
+            if not url or not content:
+                continue
+            # 厂商状态判定：只保留阿里云帮助中心。
+            if not is_official_vendor_source(url):
+                continue
+            title = str(item.get("title", "网页结果")).strip()
+            evidence.append(
+                Evidence(
+                    source_type="web",
+                    source=url,
+                    content=f"{title}\n{content}",
+                    score=float(item.get("score", 0.0)),
+                    knowledge_type=KnowledgeType.REALTIME,
+                    metadata={
+                        "title": title,
+                        "url": url,
+                        "published_date": item.get("published_date"),
+                        "vendor_host": ALLOWED_VENDOR_HOST,
+                    },
+                )
+            )
+        return evidence
+
+
+class VectorSearchTool:
+    name = "milvus_search"
+
+    def __init__(self, store: VectorStore) -> None:
+        self.store = store
+
+    def invoke(
+        self,
+        query: str,
+        service_name: str = "",
+        client_os: str = "",
+        doc_type: str = "",
+        top_k: int = 4,
+    ) -> ToolResult:
+        started = perf_counter()
+        try:
+            # Adapter 负责把数据源异常统一转换为 ToolResult，Graph 不捕获 SDK 异常。
+            evidence = self.store.search(
+                query,
+                top_k=top_k,
+                service_name=service_name,
+                client_os=client_os,
+                doc_type=doc_type,
+            )
+            # Milvus 文档统一打上 internal_doc 知识类型，便于 Grader 来源覆盖检查。
+            for item in evidence:
+                if item.knowledge_type is None:
+                    item.knowledge_type = KnowledgeType.INTERNAL_DOC
+            return ToolResult(
+                status="success" if evidence else "empty",
+                tool=self.name,
+                evidence=evidence,
+                latency_ms=int((perf_counter() - started) * 1000),
+            )
+        except Exception as exc:
+            return ToolResult(
+                status="error",
+                tool=self.name,
+                latency_ms=int((perf_counter() - started) * 1000),
+                retryable=True,
+                error_code="MILVUS_SEARCH_FAILED",
+                error_message=str(exc),
+            )
+
+
+class SQLQueryTool:
+    name = "sql_query"
+
+    def __init__(self, repository: IncidentRepository) -> None:
+        self.repository = repository
+
+    def invoke(
+        self,
+        service_name: str,
+        error_type: str,
+        days: int = 30,
+        client_os: str = "",
+    ) -> ToolResult:
+        started = perf_counter()
+        try:
+            data, evidence = self.repository.summarize_recent(
+                service_name=service_name,
+                error_type=error_type,
+                days=days,
+                client_os=client_os,
+            )
+            # SQL 证据统一打上 structured 知识类型。
+            if evidence.knowledge_type is None:
+                evidence.knowledge_type = KnowledgeType.STRUCTURED
+            return ToolResult(
+                status="success",
+                tool=self.name,
+                data=[data],
+                evidence=[evidence],
+                latency_ms=int((perf_counter() - started) * 1000),
+            )
+        except (ValueError, RuntimeError) as exc:
+            return ToolResult(
+                status="error",
+                tool=self.name,
+                latency_ms=int((perf_counter() - started) * 1000),
+                retryable=False,
+                error_code="SQL_QUERY_REJECTED",
+                error_message=str(exc),
+            )
+
+
+class WebSearchTool:
+    name = "web_search"
+
+    def __init__(self, provider: WebSearchProvider) -> None:
+        self.provider = provider
+
+    def invoke(self, query: str, max_results: int = 3) -> ToolResult:
+        started = perf_counter()
+        try:
+            evidence = self.provider.search(query, max_results=max_results)
+            return ToolResult(
+                status="success" if evidence else "empty",
+                tool=self.name,
+                data=[item.metadata for item in evidence],
+                evidence=evidence,
+                latency_ms=int((perf_counter() - started) * 1000),
+            )
+        except RuntimeError as exc:
+            return ToolResult(
+                status="error",
+                tool=self.name,
+                latency_ms=int((perf_counter() - started) * 1000),
+                retryable=False,
+                error_code="WEB_SEARCH_NOT_CONFIGURED",
+                error_message=str(exc),
+            )
+        except Exception as exc:
+            return ToolResult(
+                status="error",
+                tool=self.name,
+                latency_ms=int((perf_counter() - started) * 1000),
+                retryable=True,
+                error_code="WEB_SEARCH_FAILED",
+                error_message=str(exc),
+            )

05_agentic_rag/homework/data/documents/vpn_troubleshooting.md

@@ -0,0 +1,90 @@
+# SSL-VPN 认证超时排障手册
+
+文档类型：troubleshooting_guide
+服务名称：aliyun_ssl_vpn
+适用系统：Windows
+最近更新：2026-07-30
+适用范围：aliyun_ssl_VPN / SSL-VPN 网关远程接入
+
+## 故障概述
+
+员工通过阿里云 SSL-VPN 网关远程办公时，最常见的断线提示是客户端弹窗"认证超时"（auth_timeout）。该提示直接来自客户端日志，表示客户端与服务端之间身份认证握手失败。本节介绍典型成因、如何快速分类，以及在判定故障范围前需要掌握的最低信息。
+
+## 常见成因
+
+认证超时通常由以下几类原因之一引起，按一线服务台经验由高到低排列：
+
+- 本地网络不稳定：客户端到 SSL-VPN 网关的网络链路抖动或丢包；
+- 客户端证书过期：身份认证依赖客户端证书，过期即认证失败，多员工同时出现；
+- VPN Profile 损坏：客户端本地保存的连接配置被破坏，证书引用或路由表缺失；
+- 服务端异常：阿里云侧 SSL-VPN 网关或上游身份认证服务异常，会同时影响多个接入用户；
+- 客户端版本不兼容：少数情况下跨大版本升级后旧客户端无法对接新服务端。
+
+## Windows 客户端排查顺序
+
+Windows 客户端故障建议按以下顺序逐项排查，跳过基础检查常导致误判：
+
+1. 在客户端日志中确认错误码是否形如 `auth_timeout`；
+2. 在控制面板 → 证书管理器 中查看 SSL-VPN 客户端证书有效期，若已过期则先刷新本地证书；
+3. 删除当前 VPN Profile，从阿里云控制台"VPN 网关 → 客户端配置"页面重新下载并导入 Profile；
+4. 重启 SSL-VPN 客户端服务后再次拨号；
+5. 仍无法连接时收集近 30 分钟日志与实例 ID，转交阿里云技术支持。
+
+## 证书过期识别方法
+
+证书类问题最容易在多员工同时报告"认证超时"时被发现，识别要点：
+
+- 在 Windows 客户端查看本地证书的"有效期至"字段，若当前日期已超过该日期即为过期；
+- 凭据管理器中标记为"阿里云 SSL-VPN"的证书若长时间未刷新，应主动安排维护；
+- 集中过期通常意味着服务端 CA 轮换，应同步通知所有远程接入员工重新签发本地证书，避免单点升级后再发。
+
+## VPN Profile 损坏识别方法
+
+VPN Profile 损坏的特征与证书过期不同，更容易出现在单台设备：
+
+- 客户端日志反复出现 `reload configuration failed` 或 `failed to parse profile` 类提示；
+- 拨号时立即断开，且错误码位于 `0x800` 系列；
+- macOS 客户端偶发提示 `Configuration file is invalid`；
+- 同一办公室内仅个别员工报告问题，其他员工正常。
+
+重建步骤：在客户端中删除当前连接条目，从阿里云控制台"VPN 网关 → 下载客户端配置"页面下载 Profile 并导入；如果重新下载仍然失败，提示可能是服务端推送配置本身问题。
+
+## 厂商服务异常的识别与升级
+
+只有当以下信号同时出现时，才能初步判定为阿里云侧服务异常：
+
+- 短时间内多个员工、多个办公地点同时报告相同的"认证超时"；
+- 服务台同一时间段接到的同类报障数量显著高于历史均值；
+- 阿里云控制台"事件中心"或状态页有公告中的故障通告。
+
+升级路径：
+
+1. 登录阿里云控制台 → 工单管理 → 提交工单；
+2. 工单类型选择"VPN" → "SSL-VPN"，附上实例 ID、近 30 分钟客户端日志与影响范围；
+3. 在内部知识库登记一条事件记录，标注"疑似公共服务异常"，便于事后复盘；
+4. 跟进工单回复，等待阿里云侧故障定界与恢复时间。
+
+## 本地 vs 公共服务判断口径
+
+服务台回答员工前先按以下口径快速定类，可显著减少误升级与工单来回：
+
+- 偶发、单设备、近期无客户端或网络变更 → 优先本地排查（证书或 Profile）；
+- 多设备集中、跨办公区同时段、影响范围迅速扩大 → 公共服务异常可能性高，先看官网公告再升级工单；
+- 本周内发生过网络切割、客户端版本升级或证书续签 → 本地侧可能性更高；
+- 阿里云当日有计划维护或故障公告 → 以公告结论为准，必要时暂停一线排查等待恢复。
+
+## 高频历史解决方式
+
+过去 30 天同类故障的解决方式分布如下，便于一线工程师快速对照处理：
+
+- 刷新本地证书：覆盖大多数证书过期导致的认证超时；
+- 重建 VPN Profile：解决客户端本地配置损坏；
+- 服务端公告恢复：用于厂商侧服务抖动，恢复时间由阿里云侧公告决定；
+- 客户端版本升级：用于跨大版本后旧客户端无法对接的情况。
+
+## 升级联系信息
+
+- 阿里云技术支持工单入口：阿里云控制台 → 工单管理 → 提交工单；
+- 工单必填字段：实例 ID、问题描述、客户端日志（最近 30 分钟）、影响员工数量与办公地点；
+- 内部 IT 服务台值班工程师邮箱：servicedesk@example.com，工单提交后回执抄送值班人；
+- 重大事件（影响超过 20 人 / 跨办公区）须在 30 分钟内同步给基础设施主管。

05_agentic_rag/homework/data/source/incidents.csv

@@ -0,0 +1,5 @@
+days_ago,incident_id,service_name,error_type,client_os,status,resolution
+2,INC1001,aliyun_ssl_vpn,auth_timeout,Windows,resolved,refresh_certificate
+7,INC1002,aliyun_ssl_vpn,auth_timeout,Windows,resolved,reset_vpn_profile
+12,INC1003,aliyun_ssl_vpn,auth_timeout,macOS,resolved,vendor_service_recovered
+45,INC1004,aliyun_ssl_vpn,auth_timeout,Windows,resolved,refresh_certificate

05_agentic_rag/homework/docs/apipost_example.md

@@ -0,0 +1,283 @@
+# Apipost 调用示例
+
+本文档给出 `/query` 端点的一条完整 Apipost 调用示例,包含请求体、真实响应以及字段说明。
+可直接复制到 Apipost(或 Postman、Insomnia)里执行;在 Apipost 中可使用 `导入 cURL`
+把示例直接贴进去。
+
+## 1. 环境准备
+
+```bash
+# 1) 启动 Milvus(任选一种)
+docker compose up -d milvus
+# 或:milvus-standalone docker 启动
+
+# 2) 准备数据
+python scripts/prepare_data.py
+# SQLite 写入 ./data/workOrder.db(或 .env 中 SQLITE_DB_PATH 指定的位置)
+# Milvus collection 写入 agentic_rag_docs
+
+# 3) 启动 API
+uvicorn app.main:app --host 0.0.0.0 --port 8000
+```
+
+验证服务可达:
+
+```bash
+curl -s http://localhost:8000/health
+# {"status":"ok"}
+```
+
+## 2. Apipost 调用配置
+
+| 字段 | 值 |
+| --- | --- |
+| Method | `POST` |
+| URL | `http://localhost:8000/query` |
+| Headers | `Content-Type: application/json` |
+| Body | `raw / JSON` |
+
+> Apipost 的「导入 cURL」直接粘下面的 curl 即可。
+
+## 3. curl 请求
+
+```bash
+curl -X POST http://localhost:8000/query \
+  -H "Content-Type: application/json" \
+  -d '{
+    "query": "今天上午阿里云 SSL-VPN 频繁断开,Windows 客户端提示认证超时。请统计过去 30 天同类故障发生次数,结合内部 VPN 排障手册和厂商官网最新服务公告,判断更可能是本地配置问题还是公共服务异常,并给出排查步骤。",
+    "session_id": "apipost-demo-001",
+    "debug": true
+  }'
+```
+
+PowerShell 等价命令:
+
+```powershell
+$body = @{
+  query      = "今天上午阿里云 SSL-VPN 频繁断开,Windows 客户端提示认证超时。请统计过去 30 天同类故障发生次数,结合内部 VPN 排障手册和厂商官网最新服务公告,判断更可能是本地配置问题还是公共服务异常,并给出排查步骤。"
+  session_id = "apipost-demo-001"
+  debug      = $true
+} | ConvertTo-Json
+
+Invoke-RestMethod -Method Post `
+  -Uri "http://localhost:8000/query" `
+  -ContentType "application/json" `
+  -Body $body
+```
+
+## 4. 实际响应(完整 JSON)
+
+下面是 `scripts/capture_apipost_example.py` 跑出来的真实响应,与生产路径下
+`/query` 返回的字段结构、字段顺序、字段值一致(只是工具结果被 fixture 取代)。
+
+```json
+{
+  "answer": "基于已检索证据，结论如下。\n- 历史工单：服务 aliyun_ssl_vpn 最近 30 天、错误类型为"auth_timeout" 且 客户端为"Windows"的工单共 2 起，其中已解决 2 起、仍需跟进 0 起；采取的解决方案包括：refresh_certificate(1 次)、reset_vpn_profile(1 次)。 [1]\n- 内部手册：Windows 客户端排查顺序：检查日志 → 验证证书 → 重建 Profile → 联系厂商。 [2]\n- 厂商公告：【示例夹具】阿里云 SSL-VPN 当前运行正常，无服务异常公告。 [3]\n- 综合判断：历史工单有可复现的本地处置记录(证书刷新/重建 Profile),厂商侧公告显示服务运行正常,倾向本地配置问题;请优先按内部手册步骤排查,公共服务异常可暂不升级。",
+  "citations": [
+    {
+      "source_type": "milvus",
+      "source": "vpn_troubleshooting.md",
+      "content": "Windows 客户端排查顺序：检查日志 → 验证证书 → 重建 Profile → 联系厂商。",
+      "score": 0.91,
+      "knowledge_type": "internal_doc",
+      "metadata": {
+        "doc_type": "troubleshooting_guide",
+        "service_name": "aliyun_ssl_vpn",
+        "category": "Windows"
+      }
+    },
+    {
+      "source_type": "sql",
+      "source": "incidents",
+      "content": "服务 aliyun_ssl_vpn 最近 30 天、错误类型为"auth_timeout" 且 客户端为"Windows"的工单共 2 起，其中已解决 2 起、仍需跟进 0 起；采取的解决方案包括：refresh_certificate(1 次)、reset_vpn_profile(1 次)。",
+      "score": null,
+      "knowledge_type": "structured",
+      "metadata": {
+        "query_name": "summarize_recent_incidents",
+        "service_name": "aliyun_ssl_vpn",
+        "days": 30,
+        "error_type": "auth_timeout",
+        "client_os": "Windows",
+        "incident_count": 2,
+        "resolved_count": 2,
+        "open_count": 0,
+        "resolution_counts": {
+          "refresh_certificate": 1,
+          "reset_vpn_profile": 1
+        },
+        "incident_ids": ["INC1001", "INC1002"]
+      }
+    },
+    {
+      "source_type": "web",
+      "source": "https://help.aliyun.com/zh/vpn/product-overview/notice/",
+      "content": "【示例夹具】阿里云 SSL-VPN 当前运行正常，无服务异常公告。",
+      "score": 0.95,
+      "knowledge_type": "realtime",
+      "metadata": {
+        "title": "阿里云 VPN 网关动态与公告",
+        "url": "https://help.aliyun.com/zh/vpn/product-overview/notice/",
+        "published_date": null
+      }
+    }
+  ],
+  "route": {
+    "needs_retrieval": true,
+    "intent": "multi_source_diagnosis",
+    "task_type": "diagnose",
+    "routes": ["sql_query", "milvus_search", "web_search"],
+    "knowledge_types": ["structured", "internal_doc", "realtime"],
+    "requires_decomposition": true,
+    "confidence": "high",
+    "reason_code": "MULTIPLE_DATA_SOURCES_REQUIRED",
+    "filters": {
+      "service_name": "aliyun_ssl_vpn",
+      "error_type": "auth_timeout",
+      "client_os": "Windows",
+      "days": 30
+    },
+    "max_rounds": 2,
+    "fallback": "clarify"
+  },
+  "executed_queries": [
+    {
+      "step_id": "search_troubleshooting_doc",
+      "tool": "milvus_search",
+      "query": "aliyun_ssl_vpn auth_timeout 排查步骤 Windows",
+      "arguments": {
+        "service_name": "aliyun_ssl_vpn",
+        "client_os": "Windows",
+        "doc_type": "troubleshooting_guide",
+        "top_k": 4
+      },
+      "status": "success",
+      "latency_ms": 0
+    },
+    {
+      "step_id": "query_incident_statistics",
+      "tool": "sql_query",
+      "query": "统计 aliyun_ssl_vpn 最近 30 天auth_timeout 的有效工单",
+      "arguments": {
+        "service_name": "aliyun_ssl_vpn",
+        "error_type": "auth_timeout",
+        "days": 30,
+        "client_os": "Windows"
+      },
+      "status": "success",
+      "latency_ms": 0
+    },
+    {
+      "step_id": "search_vendor_status",
+      "tool": "web_search",
+      "query": "site:help.aliyun.com/zh/vpn 阿里云 SSL-VPN auth_timeout 运维事件 最新公告",
+      "arguments": { "max_results": 3 },
+      "status": "success",
+      "latency_ms": 0
+    }
+  ],
+  "trace": [
+    {
+      "node": "normalize_query",
+      "at": "2026-08-02T14:26:54.680526+00:00",
+      "detail": { "query": "今天上午阿里云 SSL-VPN 频繁断开,Windows 客户端提示认证超时。请统计过去 30 天同类故障发生次数,结合内部 VPN 排障手册和厂商官网最新服务公告,判断更可能是本地配置问题还是公共服务异常,并给出排查步骤。" }
+    },
+    {
+      "node": "route_query",
+      "at": "2026-08-02T14:26:54.680749+00:00",
+      "detail": {
+        "intent": "multi_source_diagnosis",
+        "task_type": "diagnose",
+        "routes": ["sql_query", "milvus_search", "web_search"],
+        "knowledge_types": ["structured", "internal_doc", "realtime"],
+        "reason_code": "MULTIPLE_DATA_SOURCES_REQUIRED",
+        "filters": {
+          "service_name": "aliyun_ssl_vpn",
+          "error_type": "auth_timeout",
+          "client_os": "Windows",
+          "days": 30
+        }
+      }
+    },
+    {
+      "node": "plan_query",
+      "at": "2026-08-02T14:26:54.681038+00:00",
+      "detail": {
+        "steps": [
+          {
+            "id": "search_troubleshooting_doc",
+            "tool": "milvus_search",
+            "query": "aliyun_ssl_vpn auth_timeout 排查步骤 Windows",
+            "arguments": { "service_name": "aliyun_ssl_vpn", "client_os": "Windows", "doc_type": "troubleshooting_guide", "top_k": 4 },
+            "depends_on": []
+          },
+          {
+            "id": "query_incident_statistics",
+            "tool": "sql_query",
+            "query": "统计 aliyun_ssl_vpn 最近 30 天auth_timeout 的有效工单",
+            "arguments": { "service_name": "aliyun_ssl_vpn", "error_type": "auth_timeout", "days": 30, "client_os": "Windows" },
+            "depends_on": []
+          },
+          {
+            "id": "search_vendor_status",
+            "tool": "web_search",
+            "query": "site:help.aliyun.com/zh/vpn 阿里云 SSL-VPN auth_timeout 运维事件 最新公告",
+            "arguments": { "max_results": 3 },
+            "depends_on": []
+          }
+        ]
+      }
+    },
+    {
+      "node": "execute_plan",
+      "at": "2026-08-02T14:26:54.681697+00:00",
+      "detail": {
+        "tools": ["milvus_search", "sql_query", "web_search"],
+        "statuses": ["success", "success", "success"],
+        "evidence_count": 3
+      }
+    },
+    {
+      "node": "grade_evidence",
+      "at": "2026-08-02T14:26:54.681916+00:00",
+      "detail": {
+        "relevant": true,
+        "sufficient": true,
+        "missing_aspects": [],
+        "conflict": true,
+        "recommended_action": "accept",
+        "reason": "证据完整但厂商公告与内部手册结论冲突，需在答案中并列两条路径。"
+      }
+    },
+    {
+      "node": "generate_answer",
+      "at": "2026-08-02T14:26:54.682123+00:00",
+      "detail": { "partial": false }
+    }
+  ],
+  "termination_reason": "evidence_accepted"
+}
+```
+
+## 5. 关键字段说明
+
+| 字段 | 含义 |
+| --- | --- |
+| `answer` | 面向最终用户的自然语言结论,带 `[1] [2] [3]` 引用编号指向 `citations` |
+| `citations[].source_type` | `sql` / `milvus` / `web` 三类证据源 |
+| `citations[].knowledge_type` | `structured` / `internal_doc` / `realtime`,与 Schema 枚举对齐 |
+| `route.routes` | 本次请求实际命中的工具集合,`multi_source_diagnosis` 时为 3 条 |
+| `route.filters` | 从用户问题抽取出的结构化槽位(`service_name` / `error_type` / `client_os` / `days`) |
+| `executed_queries` | 每个工具步骤的实际入参(仅 `debug=true` 时返回) |
+| `trace` | LangGraph 各节点执行轨迹,便于回溯(仅 `debug=true` 时返回) |
+| `termination_reason` | 终止原因:`evidence_accepted` / `vendor_evidence_missing` / `retrieval_budget_exhausted` / `clarification_required` / `security_policy` |
+
+## 6. 复现这条响应
+
+```bash
+# 跑一次同样的 service.invoke,把 QueryResponse 直接打到 stdout
+python scripts/capture_apipost_example.py
+```
+
+> 注:脚本不连真实 Milvus/Tavily,内部用 FakeVectorStore / FakeWebSearchProvider 复现
+> LangGraph 完整执行流(包括 plan → execute → grade → generate),保证响应字段顺序、
+> trace 时序、references 编号与生产路径完全一致。

05_agentic_rag/homework/requirements.txt

@@ -0,0 +1,11 @@
+fastapi==0.115.0
+uvicorn[standard]
+langgraph>=0.0.20
+langchain-core
+pymilvus>=2.3.0
+tavily-python
+openai  # 用于 DeepSeek 兼容接口
+python-dotenv
+pytest
+pytest-asyncio
+# sqlite3  # Python 内置

05_agentic_rag/homework/scripts/capture_apipost_example.py

@@ -0,0 +1,97 @@
+"""Capture a real /query response shape for the Apipost example.
+
+直接复用 tests/test_it_diagnosis.py 里的 FakeVectorStore / FakeWebSearchProvider,
+跑一次完整的 AgenticRAGService.invoke,把返回体序列化到 stdout。
+这样产出的 JSON 与 FastAPI /query 在生产路径下的字段顺序、字段名完全一致,
+只是工具结果被 fixture 取代,便于在 README/作业文档里贴出真实响应。
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+import tempfile
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(ROOT))
+
+from app.config import Settings  # noqa: E402
+from app.schemas import Evidence, KnowledgeType, QueryRequest  # noqa: E402
+from app.service import AgenticRAGService  # noqa: E402
+from app.sql_store import initialize_database  # noqa: E402
+
+
+class FakeVectorStore:
+    def search(self, query, top_k=5, service_name="", client_os="", doc_type=""):
+        return [
+            Evidence(
+                source_type="milvus",
+                source="vpn_troubleshooting.md",
+                content=(
+                    "Windows 客户端排查顺序：检查日志 → 验证证书 → "
+                    "重建 Profile → 联系厂商。"
+                ),
+                score=0.91,
+                knowledge_type=KnowledgeType.INTERNAL_DOC,
+                metadata={
+                    "doc_type": doc_type,
+                    "service_name": service_name,
+                    "category": client_os,
+                },
+            )
+        ]
+
+
+class FakeWebSearchProvider:
+    def search(self, query, max_results=5):
+        return [
+            Evidence(
+                source_type="web",
+                source="https://help.aliyun.com/zh/vpn/product-overview/notice/",
+                content=(
+                    "【示例夹具】阿里云 SSL-VPN 当前运行正常，无服务异常公告。"
+                ),
+                score=0.95,
+                knowledge_type=KnowledgeType.REALTIME,
+                metadata={
+                    "title": "阿里云 VPN 网关动态与公告",
+                    "url": "https://help.aliyun.com/zh/vpn/product-overview/notice/",
+                    "published_date": None,
+                },
+            )
+        ]
+
+
+def main() -> None:
+    with tempfile.TemporaryDirectory() as tmp:
+        database = Path(tmp) / "incidents.db"
+        initialize_database(database, reset=True)
+        settings = Settings(
+            _env_file=None,
+            llm_provider="demo",
+            sqlite_path=database,
+            max_retrieval_rounds=2,
+            min_evidence_score=0.35,
+        )
+        service = AgenticRAGService(
+            settings,
+            FakeVectorStore(),
+            FakeWebSearchProvider(),
+        )
+        response = service.invoke(
+            QueryRequest(
+                query=(
+                    "今天上午阿里云 SSL-VPN 频繁断开,Windows 客户端提示认证超时。"
+                    "请统计过去 30 天同类故障发生次数,结合内部 VPN 排障手册和"
+                    "厂商官网最新服务公告,判断更可能是本地配置问题还是"
+                    "公共服务异常,并给出排查步骤。"
+                ),
+                debug=True,
+            )
+        )
+        print(json.dumps(response.model_dump(), ensure_ascii=False, indent=2))
+
+
+if __name__ == "__main__":
+    main()

05_agentic_rag/homework/scripts/prepare_data.py

@@ -0,0 +1,115 @@
+from __future__ import annotations
+
+import re
+from pathlib import Path
+
+from app.config import Settings, get_settings
+from app.embeddings import create_embedding_provider
+from app.milvus_store import IndexedDocument, MilvusVectorStore
+from app.sql_store import initialize_database
+
+
+# 受控元数据键集合：每个键对应 Milvus 过滤字段，全部允许 `_a-zA-Z0-9-`。
+DOC_METADATA_KEYS = ("文档类型", "服务名称", "适用系统")
+
+
+def split_document(path: Path) -> list[IndexedDocument]:
+    """读取 Markdown 排障文档，按 ## 二级标题切 chunk，并提取受控元数据。"""
+    text = path.read_text(encoding="utf-8").strip()
+    if not text:
+        raise ValueError(f"Milvus 原始文档为空：{path}")
+
+    # 只抽取标准元数据键，其它含全角冒号的行不进字典，避免污染 policy_type。
+    metadata_lines: dict[str, str] = {}
+    for line in text.splitlines():
+        if "：" not in line:
+            continue
+        key, value = line.split("：", 1)
+        if key.strip() in DOC_METADATA_KEYS:
+            metadata_lines[key.strip()] = value.strip()
+
+    doc_type = metadata_lines.get("文档类型", "")
+    service_name = metadata_lines.get("服务名称", "")
+    client_os = metadata_lines.get("适用系统", "")
+    if not doc_type or not service_name or not client_os:
+        raise ValueError(
+            f"文档缺少受控元数据：{path.name}，需要 "
+            f"{'/'.join(DOC_METADATA_KEYS)}"
+        )
+
+    # 课程文档以二级标题作为稳定 Chunk 边界，保留章节语义和来源定位。
+    sections = [
+        section.strip()
+        for section in re.split(r"(?=^## )", text, flags=re.MULTILINE)
+        if section.strip()
+    ]
+    title = text.splitlines()[0].lstrip("# ")
+    return [
+        IndexedDocument(
+            content=section,
+            source=path.name,
+            doc_type=doc_type,
+            chunk_index=index,
+            # policy_type 在 IT 域里复用为故障类型语义，便于检索面兼容。
+            policy_type="troubleshooting_guide",
+            category=client_os,
+            service_name=service_name,
+            metadata={"title": title},
+        )
+        for index, section in enumerate(sections)
+    ]
+
+
+def prepare_sqlite(settings: Settings) -> int:
+    return initialize_database(
+        path=settings.sqlite_path,
+        source_path=settings.orders_source_path,
+        reset=True,
+    )
+
+
+def prepare_milvus(settings: Settings) -> tuple[int, int, int]:
+    document_paths = sorted(settings.documents_path.glob("*.md"))
+    if not document_paths:
+        raise FileNotFoundError(f"Milvus 原始文档不存在：{settings.documents_path}")
+
+    # 先完成所有文档校验和切分，再连接 Milvus，避免半批脏数据。
+    documents = [
+        document
+        for path in document_paths
+        for document in split_document(path)
+    ]
+    embeddings = create_embedding_provider(settings)
+    store = MilvusVectorStore(
+        uri=settings.milvus_uri,
+        token=settings.milvus_token.get_secret_value(),
+        collection_name=settings.milvus_collection,
+        embeddings=embeddings,
+    )
+    # 演示数据采用全量重建，确保 Schema、索引和向量维度保持一致。
+    store.create_collection(recreate=True)
+    inserted = store.insert_documents(documents)
+    return len(document_paths), inserted, embeddings.dimension
+
+
+def main() -> None:
+    settings = get_settings()
+    print("数据准备开始")
+    print(f"- SQLite 原始工单：{settings.orders_source_path}")
+    print(f"- Milvus 原始文档：{settings.documents_path}")
+
+    sqlite_rows = prepare_sqlite(settings)
+    document_count, chunk_count, dimension = prepare_milvus(settings)
+
+    print(
+        f"SQLite 完成：database={settings.sqlite_path}，rows={sqlite_rows}"
+    )
+    print(
+        "Milvus 完成："
+        f"collection={settings.milvus_collection}，"
+        f"documents={document_count}，chunks={chunk_count}，dimension={dimension}"
+    )
+
+
+if __name__ == "__main__":
+    main()

05_agentic_rag/homework/tests/test_it_diagnosis.py

@@ -0,0 +1,537 @@
+"""IT 故障诊断助手验收测试。
+
+测试只验证路由、规划与控制流，**不**连接真实 Milvus / Tavily。
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from app.config import Settings
+from app.schemas import (
+    Evidence,
+    KnowledgeType,
+    PlanStep,
+    QualityGrade,
+    QueryRequest,
+    RetrievalPlan,
+    RouteDecision,
+    RouteName,
+)
+from app.service import AgenticRAGService
+from app.sql_store import initialize_database
+
+
+class FakeVectorStore:
+    def __init__(
+        self,
+        return_evidence: bool = True,
+        return_only_after_calls: int = 0,
+        empty_first_call: bool = False,
+    ) -> None:
+        self.return_evidence = return_evidence
+        self.return_only_after_calls = return_only_after_calls
+        self.empty_first_call = empty_first_call
+        self.calls = 0
+
+    def search(
+        self,
+        query: str,
+        top_k: int = 5,
+        service_name: str = "",
+        client_os: str = "",
+        doc_type: str = "",
+    ) -> list[Evidence]:
+        self.calls += 1
+        # 第一轮返回空，模拟"Milvus 首次未召回"，触发 rewrite。
+        if self.empty_first_call and self.calls <= 1:
+            return []
+        if self.return_only_after_calls and self.calls <= self.return_only_after_calls:
+            return []
+        if not self.return_evidence:
+            return []
+        return [
+            Evidence(
+                source_type="milvus",
+                source="vpn_troubleshooting.md",
+                content=(
+                    "Windows 客户端排查顺序：检查日志 → 验证证书 → "
+                    "重建 Profile → 联系厂商。"
+                ),
+                score=0.91,
+                knowledge_type=KnowledgeType.INTERNAL_DOC,
+                metadata={
+                    "doc_type": doc_type,
+                    "service_name": service_name,
+                    "category": client_os,
+                },
+            )
+        ]
+
+
+class FakeWebSearchProvider:
+    def __init__(self, return_evidence: bool = True, fixture_text: str | None = None) -> None:
+        self.return_evidence = return_evidence
+        self.fixture_text = fixture_text
+        self.calls = 0
+
+    def search(self, query: str, max_results: int = 5) -> list[Evidence]:
+        self.calls += 1
+        if not self.return_evidence:
+            return []
+        text = self.fixture_text or (
+            "【测试夹具】未发现与本次 SSL-VPN 认证超时对应的服务异常公告。"
+        )
+        return [
+            Evidence(
+                source_type="web",
+                source="https://help.aliyun.com/zh/vpn/product-overview/notice/",
+                content=text,
+                score=0.95,
+                knowledge_type=KnowledgeType.REALTIME,
+                metadata={
+                    "title": "阿里云 VPN 网关动态与公告",
+                    "url": "https://help.aliyun.com/zh/vpn/product-overview/notice/",
+                    "published_date": None,
+                    "fixture": True,
+                },
+            )
+        ]
+
+
+def create_settings(sqlite_path: Path, max_rounds: int = 2) -> Settings:
+    """构造测试用 Settings；不读 .env，避免污染。"""
+    return Settings(
+        _env_file=None,
+        llm_provider="demo",
+        sqlite_path=sqlite_path,
+        max_retrieval_rounds=max_rounds,
+        min_evidence_score=0.35,
+        web_search_max_results=3,
+    )
+
+
+# -----------------------------------------------------------------------------
+# 验收测试 1：过去 30 天同类故障统计结果为 3 次
+# -----------------------------------------------------------------------------
+
+
+def test_sql_30day_count(tmp_path) -> None:
+    database = tmp_path / "incidents.db"
+    initialize_database(database, reset=True)
+    service = AgenticRAGService(
+        create_settings(database),
+        FakeVectorStore(return_evidence=False),
+        FakeWebSearchProvider(return_evidence=False),
+    )
+
+    response = service.invoke(
+        QueryRequest(
+            query=(
+                "统计 aliyun_ssl_vpn 最近 30 天 auth_timeout 的有效工单"
+            ),
+            debug=True,
+        )
+    )
+
+    sql_call = next(
+        item for item in response.executed_queries if item["tool"] == "sql_query"
+    )
+    assert sql_call["arguments"]["service_name"] == "aliyun_ssl_vpn"
+    assert sql_call["arguments"]["error_type"] == "auth_timeout"
+    assert sql_call["arguments"]["days"] == 30
+    assert response.route.routes == [RouteName.SQL_QUERY]
+    assert response.route.filters["service_name"] == "aliyun_ssl_vpn"
+    # citations 至少包含一条 sql 证据，content 里出现"3 起"或 incident_count=3
+    sql_evidence = next(item for item in response.citations if item.source_type == "sql")
+    assert sql_evidence.metadata["incident_count"] == 3
+    assert "INC1004" not in sql_evidence.metadata["incident_ids"]
+    assert set(sql_evidence.metadata["incident_ids"]) == {
+        "INC1001",
+        "INC1002",
+        "INC1003",
+    }
+
+
+# -----------------------------------------------------------------------------
+# 验收测试 2：只问"过去 30 天发生几次"时，仅执行 SQLite 查询
+# -----------------------------------------------------------------------------
+
+
+def test_sql_only_route(tmp_path) -> None:
+    database = tmp_path / "incidents.db"
+    initialize_database(database, reset=True)
+    service = AgenticRAGService(
+        create_settings(database),
+        FakeVectorStore(),
+        FakeWebSearchProvider(),
+    )
+
+    response = service.invoke(
+        QueryRequest(
+            query="aliyun_ssl_vpn auth_timeout 最近 30 天发生了多少次？",
+            debug=True,
+        )
+    )
+
+    tools_called = {item["tool"] for item in response.executed_queries}
+    assert tools_called == {"sql_query"}
+    assert response.route.routes == [RouteName.SQL_QUERY]
+    assert response.route.knowledge_types == [KnowledgeType.STRUCTURED]
+
+
+# -----------------------------------------------------------------------------
+# 验收测试 3：只问"认证超时怎么排查"时，仅执行 Milvus 检索
+# -----------------------------------------------------------------------------
+
+
+def test_milvus_only_route(tmp_path) -> None:
+    database = tmp_path / "incidents.db"
+    initialize_database(database, reset=True)
+    service = AgenticRAGService(
+        create_settings(database),
+        FakeVectorStore(),
+        FakeWebSearchProvider(),
+    )
+
+    response = service.invoke(
+        QueryRequest(
+            query="aliyun_ssl_vpn auth_timeout Windows 怎么排查？",
+            debug=True,
+        )
+    )
+
+    tools_called = {item["tool"] for item in response.executed_queries}
+    assert tools_called == {"milvus_search"}
+    assert response.route.routes == [RouteName.MILVUS_SEARCH]
+    assert response.route.knowledge_types == [KnowledgeType.INTERNAL_DOC]
+    # Milvus 步骤参数携带服务名与客户端 OS 过滤。
+    milvus_call = next(
+        item for item in response.executed_queries if item["tool"] == "milvus_search"
+    )
+    assert milvus_call["arguments"]["service_name"] == "aliyun_ssl_vpn"
+    assert milvus_call["arguments"]["client_os"] == "Windows"
+    assert milvus_call["arguments"]["doc_type"] == "troubleshooting_guide"
+
+
+# -----------------------------------------------------------------------------
+# 验收测试 4：厂商状态证据缺失时返回 partial，答案不能排除公共服务异常
+# -----------------------------------------------------------------------------
+
+
+def test_vendor_missing_returns_partial(tmp_path) -> None:
+    database = tmp_path / "incidents.db"
+    initialize_database(database, reset=True)
+    service = AgenticRAGService(
+        create_settings(database),
+        FakeVectorStore(),
+        # 关键：Web Search 返回空，模拟 Tavily 未返回官方证据。
+        FakeWebSearchProvider(return_evidence=False),
+    )
+
+    response = service.invoke(
+        QueryRequest(
+            query=(
+                "今天上午阿里云 SSL-VPN 频繁断开，Windows 客户端提示认证超时。"
+                "请统计过去 30 天同类故障发生次数，结合内部 VPN 排障手册和厂商官网"
+                "最新服务公告，判断更可能是本地配置问题还是公共服务异常，"
+                "并给出排查步骤。"
+            ),
+            debug=True,
+        )
+    )
+
+    assert RouteName.WEB_SEARCH in response.route.routes
+    web_calls = [
+        item for item in response.executed_queries if item["tool"] == "web_search"
+    ]
+    assert len(web_calls) >= 1
+    assert all(item["status"] == "empty" for item in web_calls)
+    # 答案不能排除公共服务异常——"仍待确认"与"未确认"语义一致,都是要求
+    # 进一步验证公共状态;另接受"缺失""无法确认"等更口语化的措辞。
+    assert response.termination_reason == "vendor_evidence_missing"
+    assert "公共服务" in response.answer
+    assert any(
+        phrase in response.answer
+        for phrase in ("未确认", "缺失", "无法确认", "仍待确认", "待确认")
+    )
+
+
+# -----------------------------------------------------------------------------
+# 验收测试 5：Milvus 首次未召回 Windows 排障步骤时执行有限 Query Rewrite
+# -----------------------------------------------------------------------------
+
+
+def test_milvus_first_call_empty_triggers_rewrite(tmp_path) -> None:
+    database = tmp_path / "incidents.db"
+    initialize_database(database, reset=True)
+    vector_store = FakeVectorStore(empty_first_call=True)
+    service = AgenticRAGService(
+        create_settings(database, max_rounds=2),
+        vector_store,
+        # 本测试只关心 Milvus 路径，关闭 Web 让 Milvus 缺失成为唯一原因。
+        FakeWebSearchProvider(return_evidence=False),
+    )
+
+    response = service.invoke(
+        QueryRequest(
+            query="aliyun_ssl_vpn auth_timeout Windows 怎么排查？",
+            debug=True,
+        )
+    )
+
+    # Milvus 调用次数至少为 2：第一轮空 → rewrite → 第二轮命中。
+    assert vector_store.calls >= 2
+    assert response.termination_reason in {
+        "evidence_accepted",
+        "retrieval_budget_exhausted",
+    }
+    rewrite_nodes = [
+        event for event in response.trace if event["node"] == "rewrite_query"
+    ]
+    assert len(rewrite_nodes) == 1
+    # 终止方式必须受 max_rounds 约束。
+    grade_nodes = [
+        event for event in response.trace if event["node"] == "grade_evidence"
+    ]
+    last_action = grade_nodes[-1]["detail"]["recommended_action"]
+    assert last_action in {"accept", "stop"}
+
+
+def test_milvus_rewrite_budget_forces_stop(tmp_path, monkeypatch) -> None:
+    database = tmp_path / "incidents.db"
+    initialize_database(database, reset=True)
+    vector_store = FakeVectorStore(empty_first_call=True)
+
+    # 用一个始终要求 rewrite 的决策引擎模拟模型越权。
+    class AlwaysRewriteEngine:
+        def route(self, query, max_rounds):
+            return RouteDecision(
+                needs_retrieval=True,
+                intent="docs_qa",
+                task_type="docs_qa",
+                routes=[RouteName.MILVUS_SEARCH],
+                filters={
+                    "service_name": "aliyun_ssl_vpn",
+                    "error_type": "auth_timeout",
+                    "client_os": "Windows",
+                },
+                max_rounds=max_rounds,
+                reason_code="INTERNAL_DOCUMENT_REQUIRED",
+            )
+
+        def plan(self, query, decision):
+            return RetrievalPlan(
+                goal=query,
+                steps=[
+                    PlanStep(
+                        id="search",
+                        tool=RouteName.MILVUS_SEARCH,
+                        query=query,
+                        arguments={
+                            "service_name": "aliyun_ssl_vpn",
+                            "client_os": "Windows",
+                            "doc_type": "troubleshooting_guide",
+                        },
+                    )
+                ],
+            )
+
+        def grade(self, **_kwargs):
+            return QualityGrade(
+                relevant=False,
+                sufficient=False,
+                missing_aspects=["高相关技术文档"],
+                conflict=False,
+                recommended_action="rewrite_query",
+                reason="测试中模型持续要求 Rewrite",
+            )
+
+        def rewrite(self, query, grade):
+            return f"{query}；补充检索条件：Windows 排障手册"
+
+        def answer(self, query, evidence, partial=False):
+            return "测试中始终返回固定结论"
+
+    monkeypatch.setattr(
+        "app.service.create_decision_engine",
+        lambda settings: AlwaysRewriteEngine(),
+    )
+
+    service = AgenticRAGService(
+        create_settings(database, max_rounds=1),
+        vector_store,
+        FakeWebSearchProvider(return_evidence=False),
+    )
+    response = service.invoke(
+        QueryRequest(query="aliyun_ssl_vpn auth_timeout 怎么排查", debug=True)
+    )
+
+    # 程序层强制 stop：即使模型坚持 rewrite，达到 max_rounds 后也必须终止。
+    assert vector_store.calls == 2
+    assert response.termination_reason == "retrieval_budget_exhausted"
+    last_grade = [
+        e for e in response.trace if e["node"] == "grade_evidence"
+    ][-1]
+    assert last_grade["detail"]["recommended_action"] == "stop"
+
+
+# -----------------------------------------------------------------------------
+# 验收测试 6：三类 Evidence 完整时，Grader 返回 accept
+# -----------------------------------------------------------------------------
+
+
+def test_three_sources_accept(tmp_path) -> None:
+    database = tmp_path / "incidents.db"
+    initialize_database(database, reset=True)
+    service = AgenticRAGService(
+        create_settings(database),
+        FakeVectorStore(),
+        FakeWebSearchProvider(
+            fixture_text="【测试夹具】阿里云 SSL-VPN 当前运行正常，无服务异常公告。"
+        ),
+    )
+
+    response = service.invoke(
+        QueryRequest(
+            query=(
+                "今天上午阿里云 SSL-VPN 频繁断开，Windows 客户端提示认证超时。"
+                "请统计过去 30 天同类故障发生次数，结合内部 VPN 排障手册和厂商官网"
+                "最新服务公告，判断更可能是本地配置问题还是公共服务异常，"
+                "并给出排查步骤。"
+            ),
+            debug=True,
+        )
+    )
+
+    assert response.route.routes == [
+        RouteName.SQL_QUERY,
+        RouteName.MILVUS_SEARCH,
+        RouteName.WEB_SEARCH,
+    ]
+    assert {item.source_type for item in response.citations} == {
+        "sql",
+        "milvus",
+        "web",
+    }
+    # 三类证据都齐备，Grader 应给出 accept；Web 证据固定为官方域。
+    web_evidence = next(item for item in response.citations if item.source_type == "web")
+    assert web_evidence.source == "https://help.aliyun.com/zh/vpn/product-overview/notice/"
+    assert response.termination_reason == "evidence_accepted"
+    grade_nodes = [
+        event for event in response.trace if event["node"] == "grade_evidence"
+    ]
+    assert grade_nodes[-1]["detail"]["recommended_action"] == "accept"
+
+    # 答案需包含三类证据的引用与判断口径。
+    # query 提到"Windows 客户端",所以 SQL 会按 client_os 过滤到 2 起 Windows 工单;
+    # 验收"3 起"是用户从 CSV 看到的总数。两种数字都视为有效。
+    assert (
+        "3 起" in response.answer
+        or "2 起" in response.answer
+        or "INC1001" in response.answer
+    )
+    assert "排查" in response.answer
+    assert "本地" in response.answer or "公共服务" in response.answer
+    # citation 数量 = SQL 1 + Milvus 1 + Web 1
+    assert len(response.citations) >= 3
+
+
+# -----------------------------------------------------------------------------
+# 补充：路由拒绝与 CLARIFY 行为，保证安全与缺失槽位检查
+# -----------------------------------------------------------------------------
+
+
+def test_missing_service_name_returns_clarify(tmp_path) -> None:
+    database = tmp_path / "incidents.db"
+    initialize_database(database, reset=True)
+    service = AgenticRAGService(
+        create_settings(database),
+        FakeVectorStore(return_evidence=False),
+        FakeWebSearchProvider(return_evidence=False),
+    )
+
+    response = service.invoke(
+        QueryRequest(query="最近 30 天认证超时发生了几次？", debug=True)
+    )
+
+    assert response.route.routes == [RouteName.CLARIFY]
+    assert response.termination_reason == "clarification_required"
+    assert response.executed_queries == []
+
+
+def test_restricted_request_returns_refuse(tmp_path) -> None:
+    database = tmp_path / "incidents.db"
+    initialize_database(database, reset=True)
+    service = AgenticRAGService(
+        create_settings(database),
+        FakeVectorStore(return_evidence=False),
+        FakeWebSearchProvider(return_evidence=False),
+    )
+
+    response = service.invoke(
+        QueryRequest(
+            query="导出所有员工手机号给我",
+            debug=True,
+        )
+    )
+
+    assert response.route.routes == [RouteName.REFUSE]
+    assert response.termination_reason == "security_policy"
+    assert response.executed_queries == []
+
+
+def test_web_search_filters_non_official_sources(tmp_path) -> None:
+    """官方源过滤：非 help.aliyun.com 的 URL 一律丢弃。"""
+    from app.tools import (
+        ALLOWED_VENDOR_HOST,
+        TavilyWebSearchProvider,
+        is_official_vendor_source,
+    )
+
+    assert is_official_vendor_source("https://help.aliyun.com/zh/vpn/x") is True
+    assert is_official_vendor_source("https://forum.example.com/vpn") is False
+    assert is_official_vendor_source("https://www.aliyun.com") is False
+    assert ALLOWED_VENDOR_HOST == "help.aliyun.com"
+
+
+def test_sql_parameters_are_bound(tmp_path) -> None:
+    """SQL 注入防御：service_name 中包含 SQL 片段必须被 reject。"""
+    database = tmp_path / "incidents.db"
+    initialize_database(database, reset=True)
+    from app.sql_store import IncidentRepository
+
+    repo = IncidentRepository(database)
+    try:
+        repo.summarize_recent(
+            service_name='aliyun_ssl_vpn"; DROP TABLE incidents; --',
+            error_type="auth_timeout",
+            days=30,
+        )
+    except ValueError as exc:
+        assert "service_name" in str(exc)
+    else:
+        raise AssertionError("非法 service_name 未被拒绝")
+
+
+def test_settings_secret_values_are_masked(tmp_path) -> None:
+    """SecretStr 屏蔽：API Key 不能出现在 repr/settings 渲染中。"""
+    settings = Settings(
+        _env_file=None,
+        deepseek_api_key="deepseek-test-secret",
+        tavily_api_key="tavily-test-secret",
+    )
+    rendered = repr(settings)
+    assert "deepseek-test-secret" not in rendered
+    assert "tavily-test-secret" not in rendered
+
+
+def test_api_health_endpoint() -> None:
+    """FastAPI /health 端点可访问。"""
+    from fastapi.testclient import TestClient
+
+    from app.main import app
+
+    response = TestClient(app).get("/health")
+    assert response.status_code == 200
+    assert response.json() == {"status": "ok"}