第六次作业代码:Agentic RAG

06_agentic_rag/VPN_agentic_rag/app/decision_engine.py

@@ -0,0 +1,234 @@
+from app.config import Settings
+from langchain_openai import ChatOpenAI
+import json
+from app.schemas import (
+    Evidence,
+    PlanStep,
+    QualityGrade,
+    RetrievalPlan,
+    RouteDecision,
+    RouteName,
+)
+SERVICE_NAME = "aliyun_ssl_vpn"
+FAULT_TYPE = "auth_timeout"
+CLIENT_OS="windows"
+def extract_service(query:str)->str:
+    """只识别明确提到的阿里云 SSL-VPN。"""
+    normalized = query.lower().replace(" ", "").replace("-", "")
+    keywords = (
+        "阿里云sslvpn",
+        "aliyunsslvpn",
+        "ssl-vpn",
+    )
+    if any(keyword.replace("-", "") in normalized for keyword in keywords):
+        return SERVICE_NAME
+
+    return ""
+
+def extract_fault_type(query: str) -> str:
+    """只识别认证超时或认证阶段导致的频繁断线。"""
+    normalized = query.lower().replace(" ", "")
+
+    auth_timeout_keywords = (
+        "认证超时",
+        "登录超时",
+        "身份验证超时",
+        "认证失败超时",
+        "auth_timeout",
+        "authenticationtimeout",
+    )
+
+    if any(keyword in normalized for keyword in auth_timeout_keywords):
+        return FAULT_TYPE
+
+    return ""
+
+def extract_client_os(query: str) -> str:
+    """返回标准化的 windows 或 macos；无法明确识别时返回空字符串。"""
+    normalized = query.lower()
+    windows_keywords = (
+        "windows", 
+        "win",
+    )
+    if any(keyword in normalized for keyword in windows_keywords):
+        return CLIENT_OS
+    return ""
+
+class DeepSeekDecisionEngine:
+    def __init__(self,settings:Settings):
+        #获取apikey
+        api_key = settings.deepseek_api_key.get_secret_value()
+        if not api_key:
+            raise RuntimeError("LLM_PROVIDER=deepseek 时必须配置 DEEPSEEK_API_KEY")
+        #非机构化llm
+        common_kwargs = {
+            "model": settings.deepseek_model_name,
+            "api_key": api_key,
+            "base_url": settings.deepseek_base_url,
+            "max_retries": 2,
+        }
+        
+        answer_thinking = "enabled" if settings.deepseek_answer_thinking else "disabled"
+        self.llm = ChatOpenAI(
+            **common_kwargs,
+            extra_body={"thinking": {"type": answer_thinking}},
+        )
+        #结构化llm
+        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:
+        service = extract_service(query)
+        fault_type = extract_fault_type(query)
+        if not service or not fault_type:
+            return RouteDecision(
+                needs_retrieval=False,
+                intent="missing_or_unsupported_vpn_fault",
+                routes=[RouteName.CLARIFY],
+                confidence="high",
+                reason_code="SERVICE_OR_AUTH_TIMEOUT_NOT_CONFIRMED",
+                max_rounds=max_rounds,
+            )
+        prompt = f"""
+你是企业 IT 服务台的故障诊断路由器。只输出符合 Schema 的结果。
+可选路径：direct_answer、milvus_search、sql_query、web_search、clarify、refuse。
+本系统仅支持阿里云 SSL-VPN（service=aliyun_ssl_vpn）的认证超时/
+频繁断线故障（fault_type=auth_timeout）。
+数据源职责：
+- 内部 VPN 排障手册、适用操作系统、客户端版本和标准排查步骤：
+  使用 milvus_search。
+- 最近 30 天同类工单数量、受影响终端、操作系统/客户端版本分布、
+  历史解决方式：使用 sql_query。
+- 阿里云官网的最新运维事件、故障公告、版本通知：
+  使用 web_search。
+
+milvus_search、sql_query、web_search；返回多条路径，并设置
+requires_decomposition=true。
+路由限制：
+- 用户只要求解释、总结、改写已有文本，且不需要查询外部资料时，
+  使用 direct_answer。
+- 用户问题没有明确阿里云 SSL-VPN，或没有明确认证超时、频繁断线、
+  登录超时等现象时，使用 clarify。
+- 用户要求修改 VPN 配置、关闭 MFA、重置账号、执行网络变更，
+  或请求密码、验证码、密钥等敏感信息时，使用 refuse。
+- 不要为不支持的 IT 服务选择检索路径。
+filters 中只填写用户问题中明确出现且可确定的字段：
+service、fault_type、client_os、client_version、region。
+不要猜测、补全或编造这些字段。
+若问题明确属于支持范围，service 固定为 aliyun_ssl_vpn，
+fault_type 固定为 auth_timeout。
+最大检索轮数：{max_rounds}
+用户问题：{query}
+""".strip()
+        result=self.router.invoke(prompt)
+        if result is None:
+            raise RuntimeError("路由器未返回结果，请检查 LLM 配置和网络连接。")
+        deterministic_filters = {
+               key: value
+               for key, value in {
+                   "service": service,
+                   "fault_type": fault_type,
+                   "client_os": extract_client_os(query),
+               }.items()
+               if value
+               }
+        result.filters = {**result.filters, **deterministic_filters}
+        result.max_rounds = max_rounds
+        return result
+
+    def plan(self,query:str,decision:RouteDecision)->RetrievalPlan:
+        filters=decision.filters
+        steps:list[PlanStep] = []
+        service = str(filters.get("service") or "")
+        fault_type = str(filters.get("fault_type") or "")
+        client_os = str(filters.get("client_os") or "")
+
+        if RouteName.MILVUS_SEARCH in decision.routes:
+            steps.append(
+                PlanStep(
+                    id="milvus_search",
+                    tool=RouteName.MILVUS_SEARCH,
+                    query=query,
+                    arguments={
+                        "service": service,
+                        "fault_type":fault_type,
+                        "client_os":client_os,
+                        "top_k": 4,
+                    },
+
+                )
+            )
+
+        if RouteName.SQL_QUERY in decision.routes:
+            steps.append(
+                PlanStep(
+                    id="sql_query",
+                    tool=RouteName.SQL_QUERY,
+                    query=query,
+                    arguments={
+                        "service": service,
+                        "fault_type": fault_type,
+                        "client_os": client_os,
+                        "days": 30,
+                    },
+
+                )
+            )
+
+        if RouteName.WEB_SEARCH in decision.routes:
+            steps.append(
+                PlanStep(
+                    id="web_search",
+                    tool=RouteName.WEB_SEARCH,
+                    query=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:
+        prompt = f"""
+判断证据是否足以回答问题。recommended_action只能是accept、rewrite_query 或 stop。
+当前轮次：{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)
+        return result
+
+    def rewrite(self,query:str,grade:QualityGrade)->str:
+        missing=grade.missing_aspects
+        return f"{query};补充条件:{missing}"
+
+    def answer(self,query:str,evidence:list[Evidence],partial:bool)->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)->DeepSeekDecisionEngine:
+    return DeepSeekDecisionEngine(settings)

06_agentic_rag/VPN_agentic_rag/app/graph.py

@@ -0,0 +1,291 @@
+from app.tools import SQLQueryTool,WebSearchTool,VectorSearchTool
+from app.decision_engine import DeepSeekDecisionEngine
+from app.config import Settings
+from dataclasses import dataclass
+
+from concurrent.futures import ThreadPoolExecutor
+from langgraph.graph import END, START, StateGraph
+from app.state import AgentState
+from app.schemas import RouteName, ToolResult,PlanStep
+
+@dataclass
+class GraphBuilder:
+    settings:Settings
+    engine:DeepSeekDecisionEngine
+    sql_tool:SQLQueryTool
+    vector_tool:VectorSearchTool
+    web_tool:WebSearchTool
+
+def trace_event(state:AgentState,node:str,detail:dict)->list[dict]:
+    return[
+        *state.get("trace",[]),
+        {
+            "node":node,
+            "detail":detail
+        }
+    ]
+
+def build_graph(builder:GraphBuilder):
+    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=builder.engine.route(state["current_query"],builder.settings.max_retrieval_rounds)
+        return{
+            "route_decision":decision,
+            "trace":trace_event(
+                state,
+                node="route_query",
+                detail={
+                    "routes": decision.routes
+                }
+            )
+        }
+
+    def after_route(state:AgentState)->str:
+        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=builder.engine.plan(state["current_query"],state["route_decision"])
+        return{
+            "retrieval_plan":plan,
+            "trace":trace_event(
+                state,
+                node="plan_query",
+                detail={
+                    "steps":plan.steps
+                }
+            )
+        }
+
+    def execute_step(step:PlanStep)->ToolResult:
+        if step.tool==RouteName.MILVUS_SEARCH:
+            return builder.vector_tool.invoke(
+                query=step.query,
+                service=step.arguments.get("service",""),
+                fault_type=step.arguments.get("fault_type",""),
+                client_os=step.arguments.get("client_os",""),
+                top_k=step.arguments.get("top_k",5)
+                )
+        if step.tool==RouteName.SQL_QUERY:
+            return builder.sql_tool.invoke(
+                service=step.arguments.get("service",""),
+                fault_type=step.arguments.get("fault_type",""),
+                days=step.arguments.get("days",30),
+                client_os=step.arguments.get("client_os","")
+            )
+        if step.tool==RouteName.WEB_SEARCH:
+            return builder.web_tool.invoke(
+                query=step.query,
+                max_results=step.arguments.get("max_results",5)
+            )
+        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]=[]
+        else:
+            with ThreadPoolExecutor(max_workers=min(len(steps),4)) as executor:
+                results=list(executor.map(execute_step, steps))
+        evidence=[]
+        for result in results:
+            evidence.extend(result.evidence)
+        errors = [
+            {
+                "tool": result.tool,
+                "code": result.error_code,
+                "message": result.error_message,
+            }
+            for result in results
+            if result.status == "error"
+        ]
+        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=builder.engine.grade(
+            query=state["current_query"],
+            decision=state["route_decision"],
+            evidence=state["evidence"],
+            current_round=current_round,
+            max_rounds=max_rounds,
+            min_score=builder.settings.min_evidence_score
+        )
+        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=state,
+                node="grade_evidence",
+                detail=grade.model_dump(mode="json"),
+            ),
+        }
+
+    def after_grade(state:AgentState)->str:
+        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=builder.engine.rewrite(
+            query=state["current_query"],
+            grade=state["quality_grade"])
+        next_round=state.get("retrieval_round",0)+1
+        return{
+            "current_query":rewritten,
+            "retrieval_round":next_round,
+            "trace":trace_event(state=state,
+                                node="rewrite_query",
+                                detail={
+                                    "round":next_round,
+                                    "rewritten":rewritten
+                                })
+        }
+
+    def generate_answer(state: AgentState) -> dict:
+        answer = builder.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 = builder.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": "当前流程仅处理阿里云 SSL-VPN 的认证超时故障；请确认是否出现认证超时提示，并补充客户端系统或版本信息。",
+            "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", {}),
+        }
+
+    graph=StateGraph(AgentState)
+    graph.add_node("normalize_query",normalize_query)
+    graph.add_node("route_query",route_query)
+    graph.add_node("plan_query",plan_query)
+    graph.add_node("execute_plan",execute_plan)
+    graph.add_node("grade_evidence",grade_evidence)
+    graph.add_node("rewrite_query",rewrite_query)
+    graph.add_node("generate_answer",generate_answer)
+    graph.add_node("generate_partial_answer",generate_partial_answer)
+    graph.add_node("clarify",clarify)
+    graph.add_node("refuse",refuse)
+
+    graph.add_edge(START,"normalize_query")
+    graph.add_edge("normalize_query","route_query")
+    graph.add_conditional_edges(
+        "route_query",
+        after_route,
+        {
+            "clarify": "clarify",
+            "refuse": "refuse",
+            "generate": "generate_answer",
+            "plan": "plan_query",
+        }
+    )
+    graph.add_edge("plan_query","execute_plan")
+    graph.add_edge("execute_plan","grade_evidence")
+    graph.add_conditional_edges(
+        "grade_evidence",
+        after_grade,
+        {
+            "generate":"generate_answer",
+            "rewrite":"rewrite_query",
+            "generate_partial":"generate_partial_answer"
+        }
+    )
+
+    graph.add_edge("rewrite_query","plan_query")
+    graph.add_edge("generate_answer",END)
+    graph.add_edge("generate_partial_answer",END)
+    graph.add_edge("clarify", END)
+    graph.add_edge("refuse", END)
+
+    return graph.compile()
+
+
+
+
+    
+            
+
+
+

06_agentic_rag/VPN_agentic_rag/app/main.py

@@ -0,0 +1,31 @@
+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="Agentic RAG Course Project",
+    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)
+

06_agentic_rag/VPN_agentic_rag/app/service.py

@@ -0,0 +1,85 @@
+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 GraphBuilder, 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,
+    WebSearchTool,
+)
+
+
+class AgenticRAGService:
+    """组装 VPN 故障诊断 Agent，并提供单次查询入口。"""
+
+    def __init__(self, settings: Settings) -> None:
+        self.settings = settings
+
+        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,
+        )
+        vector_tool = VectorSearchTool(vector_store)
+
+        incident_repository = IncidentRepository(settings.sqlite_path)
+        sql_tool = SQLQueryTool(incident_repository)
+
+        web_provider = TavilyWebSearchProvider(
+            settings.tavily_api_key.get_secret_value()
+        )
+        web_tool = WebSearchTool(web_provider)
+
+        graph_builder = GraphBuilder(
+            settings=settings,
+            engine=create_decision_engine(settings),
+            vector_tool=vector_tool,
+            sql_tool=sql_tool,
+            web_tool=web_tool,
+        )
+        self.graph = build_graph(graph_builder)
+
+    def invoke(self, request: QueryRequest) -> QueryResponse:
+        """为每次请求创建独立 State 并运行 LangGraph。"""
+        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"),
+        )

06_agentic_rag/VPN_agentic_rag/data/documents/vpn_auth_timeout_certificate_profile.md

@@ -0,0 +1,27 @@
+# SSL-VPN 证书与配置文件排障指引（模拟）
+
+服务：aliyun_ssl_vpn
+故障类型：auth_timeout
+适用系统：all
+客户端版本：all
+文档类别：troubleshooting
+
+> 本文档为课程项目的模拟内部手册，仅用于演示检索、证据合并和排障顺序生成。
+
+## 适用范围
+
+适用于 Windows 与 macOS 终端的 SSL-VPN 认证超时。尤其适用于历史工单中“刷新证书”或“重新导入 VPN 配置文件”被记录为解决方式的情况。
+
+## 诊断顺序
+
+1. 先确认故障是否局限于单一员工，还是近 30 天存在多起同类工单。
+2. 检查工单中的操作系统分布；若集中于某系统，应优先使用该系统对应的排障手册。
+3. 检查证书有效期、证书更新记录和配置文件版本。
+4. 在 IT 服务台提供的受控渠道中刷新证书或重新导入配置文件。
+5. 重新认证后验证连接稳定性，并记录结果以更新工单。
+6. 若多终端同时受影响，或阿里云官方公告提示服务异常，暂停将问题简单归因为单个终端配置，并升级为平台侧风险排查。
+
+## 安全要求
+
+- 不在聊天、工单正文或日志中记录密码、验证码、私钥或完整证书内容。
+- 不自动执行网关配置修改、账号重置、MFA 关闭或网络策略变更。

06_agentic_rag/VPN_agentic_rag/data/documents/vpn_auth_timeout_macos.md

@@ -0,0 +1,27 @@
+# macOS SSL-VPN 认证超时排障手册（模拟）
+
+服务：aliyun_ssl_vpn
+故障类型：auth_timeout
+适用系统：macos
+客户端版本：all
+文档类别：troubleshooting
+
+> 本文档为课程项目的模拟内部手册，仅用于演示检索、证据合并和排障顺序生成。
+
+## 适用现象
+
+macOS 终端在启动 SSL-VPN 客户端、完成身份验证或网络切换后出现认证超时，导致无法建立或维持 VPN 连接。
+
+## 排查步骤
+
+1. 记录 macOS 版本、VPN 客户端版本、故障时间和完整错误信息。
+2. 确认系统日期、时间和时区设置正确，并确认网络未被企业代理或访客网络拦截。
+3. 核对客户端使用的证书或配置文件是否为当前版本；如配置已轮换，重新导入最新配置文件。
+4. 退出后重新打开客户端并再次认证；不得通过关闭 MFA 等方式绕过认证流程。
+5. 对照最近 30 天同类工单，判断是否集中于特定 macOS 版本或特定客户端版本。
+6. 若工单不集中于单一终端环境，查询阿里云官网是否发布相关运维事件、故障公告或版本通知。
+
+## 升级条件
+
+当重新导入配置后仍持续认证超时，且官方公告显示可能存在服务侧事件时，将故障时间、区域信息和日志摘要提交给网络平台主管人工跟进。
+

06_agentic_rag/VPN_agentic_rag/data/documents/vpn_auth_timeout_windows.md

@@ -0,0 +1,28 @@
+# Windows SSL-VPN 认证超时排障手册（模拟）
+
+服务：aliyun_ssl_vpn
+故障类型：auth_timeout
+适用系统：windows
+客户端版本：all
+文档类别：troubleshooting
+
+> 本文档为课程项目的模拟内部手册，仅用于演示检索、证据合并和排障顺序生成。
+
+## 适用现象
+
+员工在 Windows 终端使用阿里云 SSL-VPN 时，连接或重新认证阶段出现“认证超时”提示，或在重新认证后反复断线。
+
+## 排查步骤
+
+1. 记录故障发生时间、客户端版本、系统版本和错误提示截图。
+2. 确认终端系统时间、时区和时间同步状态正确；认证令牌对时间偏差敏感。
+3. 检查客户端证书是否在有效期内；证书已更新时，刷新证书后重新连接。
+4. 若历史工单集中于某客户端版本，优先升级到 IT 服务台批准的客户端版本。
+5. 重新下载并导入经 IT 服务台确认的 VPN 配置文件，再执行连接测试。
+6. 若多个终端在相近时间持续失败，先查询阿里云官方状态与运维公告，再将工单升级给网络平台主管。
+
+## 处理边界
+
+- 本手册仅描述只读排查和由员工自行完成的客户端检查。
+- 不在此流程中关闭 MFA、修改网关配置或重置账号。
+

06_agentic_rag/VPN_agentic_rag/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

06_agentic_rag/VPN_agentic_rag/scripts/prepare_data.py

@@ -0,0 +1,123 @@
+from __future__ import annotations
+
+import re
+from pathlib import Path
+
+from app.config import PROJECT_ROOT, Settings, get_settings
+from app.embeddings import create_embedding_provider
+from app.milvus_store import IndexedDocument, MilvusVectorStore
+from app.sql_store import initialize_database
+
+
+INCIDENTS_SOURCE_PATH = PROJECT_ROOT / "data" / "source" / "incidents.csv"
+
+
+def split_document(path: Path) -> list[IndexedDocument]:
+    text = path.read_text(encoding="utf-8").strip()
+    if not text:
+        raise ValueError(f"Milvus 原始文档为空：{path}")
+
+    metadata_lines = {
+        line.split("：", 1)[0]: line.split("：", 1)[1].strip()
+        for line in text.splitlines()
+        if "：" in line
+    }
+    service = metadata_lines.get("服务", "")
+    fault_type = metadata_lines.get("故障类型", "")
+    client_os = metadata_lines.get("适用系统", "")
+    client_version = metadata_lines.get("客户端版本", "")
+    document_type = metadata_lines.get("文档类别", "")
+    required_metadata = {
+        "服务": service,
+        "故障类型": fault_type,
+        "适用系统": client_os,
+        "客户端版本": client_version,
+        "文档类别": document_type,
+    }
+    missing_metadata = [
+        field_name
+        for field_name, value in required_metadata.items()
+        if not value
+    ]
+    if missing_metadata:
+        raise ValueError(
+            f"文档缺少元数据 {', '.join(missing_metadata)}：{path.name}"
+        )
+
+    # 课程文档以二级标题作为稳定 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=document_type,
+            chunk_index=index,
+            service=service,
+            fault_type=fault_type,
+            client_os=client_os,
+            client_version=client_version,
+            metadata={
+                "title": title,
+                "document_type": document_type,
+            },
+        )
+        for index, section in enumerate(sections)
+    ]
+
+
+def prepare_sqlite(settings: Settings) -> int:
+    return initialize_database(
+        path=settings.sqlite_path,
+        source_path=INCIDENTS_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 原始工单：{INCIDENTS_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()