ZhouYI/06_agentic_rag/VPN_agentic_rag/app/decision_engine.py

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)