DLOS AI操作系统：基于双环验证架构的AI输出治理系统

DLOS AI操作系统：基于双环验证架构的AI输出治理系统

技术支持：拓世网络技术开发部

完整技术规范与实现

---

摘要

大规模语言模型（LLM）的快速普及在人工智能基础设施中创造了一个关键缺口：缺乏能够以确定性可靠性验证、控制和执行LLM输出的系统化治理层。本文提出了DLOS（深度学习输出系统），一个基于双环验证架构构建的AI输出治理操作系统。与传统方法将输出验证视为事后处理的附属步骤不同，DLOS作为一个系统级操作层，拦截LLM生成内容，通过多维验证（网络事实检查FCS、逻辑推理验证RCS、时序状态预测推理SAS）对其进行处理，计算人因可靠性指数（HRI），并通过闭环反馈机制执行治理决策（放行/重写/拦截）。该系统在保持通过规则引擎和反馈机制实现自适应学习能力的同时，实现了可验证的输出控制。本文呈现了DLOS作为AI栈基础设施软件的完整架构规范、工程实现、部署架构和商业定位。

---

第一章 引言与问题陈述

1.1 AI治理鸿沟

大规模语言模型的出现从根本上改变了人类与计算系统的交互方式。GPT-4、Claude、Llama及其变体等模型在自然语言理解、生成、推理和任务完成方面展现出卓越能力。然而，这些模型共享一个关键漏洞：它们的输出是概率性的而非确定性的，是统计优化的而非逻辑保证的，是上下文可变的而非系统可验证的。

这种概率性质造成了我们称之为"AI治理鸿沟"的问题——在LLM生成与现实世界执行之间，缺乏能够确保输出可靠性、安全性和可控性的系统化层。当前解决该问题的方法可分为以下三类不充分的类别：

提示工程方法试图通过精心设计的指令来约束输出。虽然有用，但这种方法无法提供硬性保证，并且在提示被越狱或模型遇到边缘情况时会系统性地失效。

微调方法修改模型权重以使输出偏向期望行为。这些方法需要大量计算资源，创建无法适应新威胁的静态解决方案，并且仍然产生概率性输出。

后处理过滤器在生成后应用关键词屏蔽或情感分析。这些方法是被动而非主动的，无法验证事实声明或逻辑一致性，并且不提供系统级学习机制。

1.2 操作系统类比

AI输出操作系统的论证基于一个历史类比：早期计算在不可靠硬件和不一致软件执行方面面临类似挑战。解决方案不是更好的应用程序或更谨慎的编程，而是操作系统——一个抽象硬件复杂性、提供一致接口、强制执行安全边界并支持受控执行的系统级层。

DLOS将同样的洞见应用于人工智能。正如传统操作系统位于硬件和应用之间，DLOS位于LLM和执行环境之间。它提供：

· 进程隔离：不同AI输出与其执行上下文之间的隔离

· 内存管理：跨验证周期维护状态

· 调度：决定哪些输出继续执行

· 安全执行：通过基于HRI的决策引擎实现

· 系统调用：通过VALIDATOR API实现受控的LLM交互

1.3 贡献

本文做出以下贡献：

1. 架构规范：用于AI输出治理的双环验证操作系统的完整设计

2. 验证分类法：覆盖事实准确性（FCS）、逻辑一致性（RCS）和时序状态对齐（SAS）的三维验证

3. HRI评分系统：用于在0-1尺度上量化输出可靠性的数学框架

4. 决策引擎：具有自适应阈值的三态治理模型（放行/重写/拦截）

5. 完整实现：后端、前端和部署基础设施的生产级代码

6. 商业框架：SaaS商业模式、API货币化策略和市场定位

---

第二章 系统架构

2.1 架构概述

DLOS实现了双环验证架构，其中外环治理内环的学习和适应。本节描述两个环及其交互。

2.1.1 内环：验证-执行周期

内环表示单个LLM输出通过系统的实时路径：

```

用户输入 → LLM生成 → 验证器 → 决策引擎 → 执行/重写/拦截

↓

反馈收集

```

该环在推理时运行，独立处理每个输出，同时通过规则引擎维护状态。

2.1.2 外环：学习-适应周期

外环跨验证周期运行，基于累积反馈更新系统行为：

```

验证历史 → 规则引擎更新 → 反馈分析 → TSPR状态精化

↓

规则传播至验证器

```

该环使系统能够随时间改进，从成功和失败的验证中学习。

2.1.3 双环集成

两个环通过规则引擎集成，规则引擎维护：

· 静态规则：永不改变的硬编码约束（例如"拦截SQL注入模式"）

· 动态规则：通过反馈演化的学习模式（例如"当上下文包含金融术语时，重写HRI>0.4的输出"）

· TSPR状态：捕获系统历史的时序状态表示

2.2 组件架构

2.2.1 LLM引擎层

LLM引擎抽象模型异质性，无论底层模型如何（GPT-4、Claude、Llama或自定义模型），都提供统一的生成接口。主要职责包括：

· 基于任务需求的模型路由

· 温度和采样参数管理

· 保留上下文的提示模板注入

· 输出流式和缓冲

2.2.2 验证器子系统

验证器是核心验证引擎，实现三个并行验证通道：

网络验证器（FCS - 事实检查分数）

根据可验证来源验证事实声明。对于输出中的每个事实声明，验证器：

1. 通过命名实体识别和关系提取提取声明

2. 为声明验证生成搜索查询

3. 与可信知识库交叉引用（维基数据、DBpedia、自定义知识图谱）

4. 基于来源权威性和佐证计数计算声明置信度分数

5. 将单个声明分数聚合为总体FCS（0到1尺度，0=完全验证，1=完全未验证）

逻辑验证器（RCS - 推理一致性分数）

评估逻辑连贯性和推理有效性：

1. 使用自定义语义解析器将输出解析为逻辑命题

2. 构建以命题为节点、关系为边的逻辑图

3. 通过识别具有真值冲突的节点检测矛盾

4. 评估推理链的三段论有效性

5. 检查常见谬误（循环推理、错误二分法、草率概括）

6. 将RCS计算为逻辑有效语句的比例（0=完全有效，1=完全无效）

TSPR验证器（SAS - 状态对齐分数）

验证与系统状态和时序约束的对齐：

1. 维护表示系统对世界理解的状态向量

2. 将输出声明与当前状态进行比较（对于现在时声明）

3. 基于输出预测投射未来状态并检查可行性

4. 针对记录的状态历史验证过去时声明的历史一致性

5. 将SAS计算为与期望状态的偏离（0=完全对齐，1=完全偏离）

2.2.3 HRI评分引擎

人因可靠性指数（HRI）在0到1尺度上量化总体输出可靠性，其中0表示完全可靠，1表示完全不可靠。该公式结合了每个验证维度的加权贡献：

```

HRI = 1 - (W_fcs × FCS + W_rcs × RCS + W_sas × SAS)

```

权重可配置但默认为：

· W_fcs = 0.4（事实准确性权重最高）

· W_rcs = 0.3（逻辑一致性）

· W_sas = 0.3（状态对齐）

HRI分数支持：

· 定量比较：不同输出或模型之间的比较

· 基于阈值的决策：具有清晰数值边界的决策

· 趋势分析：监控系统随时间变化的性能

· A/B测试：验证配置变更的测试

2.2.4 决策引擎

决策引擎基于HRI阈值实现三态治理模型：

```

如果 HRI < 0.2： 决策 = "放行" # 输出可靠

如果 0.2 ≤ HRI < 0.5：决策 = "重写" # 输出需要修订

如果 HRI ≥ 0.5： 决策 = "拦截" # 输出不安全

```

放行状态表示输出满足所有验证标准。输出继续执行，反馈循环记录成功的验证以进行正强化。

重写状态表示部分验证失败。系统启动重写周期：

1. 原始输出被标记为拒绝

2. 验证结果（包括具体失败原因）被反馈给LLM

3. LLM生成修订版本，并附带明确指示以解决失败

4. 修订输出再次进入验证器（最多重写3次）

5. 如果所有重写尝试都失败，输出转入拦截状态

拦截状态表示关键验证失败。输出将：

1. 被记录为安全/安保事件

2. 永不暴露给执行环境

3. 用作规则引擎更新的训练数据

4. 连同完整验证轨迹一起报告给系统管理员

2.2.5 规则引擎

规则引擎通过生产规则系统维护和演化系统行为：

规则表示

每条规则遵循以下结构：

```json

{

"id": "rule_uuid",

"condition": "hri > 0.3 AND context_domain == 'finance'",

"action": "BLOCK",

"weight": 0.85,

"created_at": "timestamp",

"update_count": 42,

"success_rate": 0.76

}

```

规则学习

规则通过以下方式演化：

1. 模式提取：从反馈日志中挖掘重复出现的验证模式

2. 关联规则挖掘：Apriori算法识别上下文特征与验证结果之间的相关性

3. 规则加权：每条规则基于历史表现维护成功率

4. 规则剪枝：定期移除低权重规则以防止膨胀

2.2.6 反馈循环

反馈循环通过收集和处理系统结果来关闭治理周期：

数据收集

对于每个验证周期，系统记录：

· 输入上下文（领域、任务类型、用户角色、对话历史）

· LLM输出（完整文本、令牌计数、困惑度分数）

· 验证分数（FCS、RCS、SAS、HRI）

· 决策结果（放行/重写/拦截）

· 执行结果（如果是放行，实际发生了什么）

· 用户反馈（明确的点赞/踩或隐式行为信号）

学习信号

反馈循环生成三种类型的学习信号：

1. 正强化（放行决策且执行成功）：增加对验证阈值的置信度

2. 负强化（放行决策但执行失败）：表明阈值过低，触发调整

3. 修正信号（拦截决策但用户覆盖）：表明阈值过高，触发调整

规则传播

学习的模式以可配置的间隔（每100次验证或每小时，以先到者为准）传播到规则引擎。传播包括：

· 插入新规则

· 更新现有规则权重

· 重新校准阈值

· 更新TSPR状态向量

---

第三章 数学框架

3.1 事实检查分数（FCS）

3.1.1 声明提取

给定输出O，定义事实声明集合C = {c₁, c₂, ..., cₙ}，其中每个声明cⱼ是形如（主语、谓语、宾语）的命题，通过依存句法分析和语义角色标注提取。

3.1.2 验证函数

对于每个声明cⱼ，定义验证函数V(cⱼ) → [0,1]为：

```

V(cⱼ) = α × S(cⱼ) + β × C(cⱼ) + γ × A(cⱼ)

```

其中：

· S(cⱼ)是来源权威性分数（基于域名排名、引用次数、同行评审状态）

· C(cⱼ)是佐证计数（独立同意的来源数量）

· A(cⱼ)是声明时效因子（越古老的声明需要越强的证据）

· α、β、γ是权重因子，满足α + β + γ = 1

3.1.3 总体FCS

```

FCS = (1/|C|) × Σ V(cⱼ) for j = 1 to |C|

```

性质：当所有声明完全验证时FCS = 0，当所有声明完全未验证时FCS = 1。

3.2 推理一致性分数（RCS）

3.2.1 逻辑图构建

将O解析为有向逻辑图G = (P, E)，其中：

· P是命题集合（原子语句）

· E是逻辑关系集合（蕴含、矛盾、支持）

3.2.2 矛盾检测

对于任意两个命题pᵢ, pⱼ ∈ P，定义矛盾函数：

```

Contradict(pᵢ, pⱼ) = 1 if (pᵢ ∧ ¬pⱼ) or (¬pᵢ ∧ pⱼ)

Contradict(pᵢ, pⱼ) = 0 otherwise

```

总矛盾数：

```

K = Σ Σ Contradict(pᵢ, pⱼ) for i < j

```

最大可能矛盾数：

```

K_max = |P| × (|P| - 1) / 2

```

3.2.3 有效性分数

```

RCS = 1 - (K / K_max) if |P| > 1

RCS = 0 if |P| ≤ 1

```

RCS范围从0（完全一致）到1（最大矛盾）。

3.3 状态对齐分数（SAS）

3.3.1 状态向量表示

定义时间t的系统状态为向量S(t) = (s₁, s₂, ..., sₘ)，其中每个sₖ是状态变量（例如user_location、conversation_topic、system_mode）。

3.3.2 状态偏离

对于在时间t生成的输出O，基于O的声明定义预测状态S'(t+1)。偏离度为：

```

Δ(t) = ||S'(t+1) - S_expected(t+1)||₂

```

其中S_expected是系统在没有O影响下期望的状态。

3.3.3 归一化分数

```

SAS = tanh(Δ(t) / Δ_max)

```

其中Δ_max是最大观测偏离度（动态更新）。SAS范围从0（完全对齐）到1（最大偏离）。

3.4 人因可靠性指数（HRI）

3.4.1 加权组合

```

HRI = 1 - (w_fcs × FCS + w_rcs × RCS + w_sas × SAS)

```

约束条件：w_fcs + w_rcs + w_sas = 1且每个w ≥ 0

3.4.2 动态权重调整

权重可根据领域和任务类型自适应调整：

```

w_fcs(domain) = base_fcs × (1 + α_domain × domain_sensitivity)

```

其中domain_sensitivity表示事实准确性对该领域的 criticality（例如医疗为0.9，创意写作为0.2）。

---

第四章 完整实现

4.1 后端实现

4.1.1 主应用入口点（app.py）

```python

"""

DLOS AI操作系统 v1.0

后端API入口点

"""

from fastapi import FastAPI, HTTPException, BackgroundTasks

from fastapi.middleware.cors import CORSMiddleware

from fastapi.responses import JSONResponse

from pydantic import BaseModel, Field

from typing import Optional, Dict, Any, List

from datetime import datetime

import uuid

import logging

import json

from validator.core import Validator

from rule_engine.engine import RuleEngine

from feedback_loop.collector import FeedbackCollector

from llm_engine.manager import LLMEngineManager

# 配置日志

logging.basicConfig(

level=logging.INFO,

format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'

)

logger = logging.getLogger("dlos-api")

# 初始化FastAPI应用

app = FastAPI(

title="DLOS AI操作系统API",

description="具有双环验证的AI输出治理操作系统",

version="1.0.0",

docs_url="/docs",

redoc_url="/redoc"

)

# 配置CORS

app.add_middleware(

CORSMiddleware,

allow_origins=["*"], # 生产环境需正确配置

allow_credentials=True,

allow_methods=["*"],

allow_headers=["*"],

)

# 初始化核心组件

validator = Validator()

rule_engine = RuleEngine()

feedback_collector = FeedbackCollector()

llm_manager = LLMEngineManager()

# 请求/响应模型

class ValidationRequest(BaseModel):

output: str = Field(..., description="待验证的LLM输出")

context: Dict[str, Any] = Field(..., description="包含领域、任务、历史的上下文")

session_id: Optional[str] = Field(None, description="用于状态跟踪的会话标识符")

rewrite_attempt: Optional[int] = Field(0, description="已进行的重写尝试次数")

class RewriteRequest(BaseModel):

original_output: str

validation_results: Dict[str, Any]

context: Dict[str, Any]

failure_reasons: List[str]

class ValidationResponse(BaseModel):

request_id: str

timestamp: datetime

fcs: float

rcs: float

sas: float

hri: float

decision: str

rewrite_suggestion: Optional[str]

validation_details: Dict[str, Any]

class FeedbackRequest(BaseModel):

request_id: str

user_rating: Optional[int] = Field(None, ge=1, le=5)

execution_success: Optional[bool]

user_correction: Optional[str]

metadata: Optional[Dict[str, Any]]

# 健康检查端点

@app.get("/health")

async def health_check():

return {

"status": "operational",

"version": "1.0.0",

"timestamp": datetime.now().isoformat(),

"components": {

"validator": validator.health_check(),

"rule_engine": rule_engine.health_check(),

"feedback_collector": feedback_collector.health_check()

}

}

# 主验证端点

@app.post("/v1/validate", response_model=ValidationResponse)

async def validate_output(request: ValidationRequest, background_tasks: BackgroundTasks):

"""

通过双环验证系统验证LLM输出。

该端点通过所有三个验证通道处理LLM输出：

- FCS（事实检查分数）

- RCS（推理一致性分数）

- SAS（状态对齐分数）

返回HRI分数和治理决策（放行/重写/拦截）。

"""

request_id = str(uuid.uuid4())

logger.info(f"处理验证请求 {request_id}")

try:

# 应用规则引擎预处理

applicable_rules = rule_engine.get_rules_for_context(request.context)

# 运行验证

validation_result = validator.process(

output=request.output,

context=request.context,

session_id=request.session_id,

applicable_rules=applicable_rules

)

# 处理重写决策

rewrite_suggestion = None

if validation_result["decision"] == "REWRITE" and request.rewrite_attempt < 3:

rewrite_suggestion = await generate_rewrite(

original_output=request.output,

validation_results=validation_result,

context=request.context,

attempt=request.rewrite_attempt + 1

)

# 准备响应

response = ValidationResponse(

request_id=request_id,

timestamp=datetime.now(),

fcs=validation_result["fcs"],

rcs=validation_result["rcs"],

sas=validation_result["sas"],

hri=validation_result["hri"],

decision=validation_result["decision"],

rewrite_suggestion=rewrite_suggestion,

validation_details=validation_result["details"]

)

# 为反馈循环记录验证（异步）

background_tasks.add_task(

feedback_collector.log_validation,

request_id=request_id,

request=request,

response=response

)

logger.info(f"验证 {request_id} 完成，决策: {response.decision}")

return response

except Exception as e:

logger.error(f"验证失败 {request_id}: {str(e)}", exc_info=True)

raise HTTPException(status_code=500, detail=f"验证失败: {str(e)}")

# 批量验证端点

@app.post("/v1/validate/batch")

async def validate_batch(requests: List[ValidationRequest]):

"""

批量验证多个输出以提高效率。

"""

results = []

for req in requests:

result = await validate_output(req, BackgroundTasks())

results.append(result)

return {"results": results}

# 反馈端点

@app.post("/v1/feedback")

async def submit_feedback(feedback: FeedbackRequest, background_tasks: BackgroundTasks):

"""

提交用户反馈以改进验证。

这会触发外学习循环。

"""

background_tasks.add_task(

feedback_collector.process_feedback,

feedback=feedback

)

# 如果累积了足够反馈，触发规则更新

if feedback_collector.should_update_rules():

background_tasks.add_task(rule_engine.update_from_feedback)

return {"status": "accepted", "feedback_id": feedback.request_id}

# 指标端点

@app.get("/v1/metrics")

async def get_metrics(start_time: Optional[str] = None, end_time: Optional[str] = None):

"""

获取系统指标，包括通过率、平均HRI和组件性能。

"""

metrics = {

"total_validations": feedback_collector.get_total_count(),

"pass_rate": feedback_collector.get_pass_rate(),

"average_hri": feedback_collector.get_avg_hri(),

"decision_distribution": feedback_collector.get_decision_distribution(),

"validator_performance": validator.get_performance_metrics(),

"rule_engine_stats": rule_engine.get_statistics()

}

return metrics

# 规则管理端点

@app.get("/v1/rules")

async def get_rules(active_only: bool = True):

"""从规则引擎获取当前规则。"""

return {"rules": rule_engine.get_rules(active_only=active_only)}

@app.post("/v1/rules")

async def add_rule(rule: Dict[str, Any]):

"""向引擎添加新规则。"""

rule_id = rule_engine.add_rule(rule)

return {"rule_id": rule_id, "status": "added"}

@app.delete("/v1/rules/{rule_id}")

async def delete_rule(rule_id: str):

"""从引擎删除规则。"""

success = rule_engine.delete_rule(rule_id)

return {"success": success}

# TSPR状态端点

@app.get("/v1/state")

async def get_state(session_id: str):

"""获取会话的当前TSPR状态。"""

state = validator.get_state(session_id)

return {"session_id": session_id, "state": state}

@app.post("/v1/state/reset")

async def reset_state(session_id: str):

"""重置会话的TSPR状态。"""

validator.reset_state(session_id)

return {"session_id": session_id, "status": "reset"}

async def generate_rewrite(

original_output: str,

validation_results: Dict[str, Any],

context: Dict[str, Any],

attempt: int

) -> str:

"""

基于验证失败生成输出的修订版本。

"""

# 提取具体失败原因

failure_reasons = []

if validation_results["fcs"] > 0.3:

failure_reasons.append("事实声明需要验证")

if validation_results["rcs"] > 0.3:

failure_reasons.append("检测到逻辑不一致")

if validation_results["sas"] > 0.3:

failure_reasons.append("识别出状态对齐问题")

# 构建重写提示

rewrite_prompt = f"""

以下响应需要重写以解决验证问题。

原始输出：{original_output}

需要解决的问题：

{chr(10).join(f'- {reason}' for reason in failure_reasons)}

要求：

1. 保持原始意图和有用性

2. 纠正所有事实声明

3. 确保整个输出的逻辑一致性

4. 与对话状态对齐

5. 对不确定的声明更加保守

请提供修订版本：

"""

# 调用LLM进行重写

rewrite_result = await llm_manager.generate(

prompt=rewrite_prompt,

context=context,

temperature=0.3, # 较低温度以获得更确定性的输出

max_tokens=2000

)

return rewrite_result["output"]

if __name__ == "__main__":

import uvicorn

uvicorn.run(

"app:app",

host="0.0.0.0",

port=8000,

reload=True,

log_level="info"

)

```

4.1.2 验证器核心实现（validator/core.py）

```python

"""

DLOS验证器核心

实现三通道验证系统

"""

from typing import Dict, Any, List, Optional, Tuple

import numpy as np

from datetime import datetime

import hashlib

import json

import re

from collections import defaultdict

from dataclasses import dataclass, field

from validator.web_checker import WebFactChecker

from validator.logic_checker import LogicConsistencyChecker

from validator.tspr_checker import TSPRStateChecker

from validator.decision_engine import DecisionEngine

@dataclass

class ValidationResult:

"""验证结果数据类"""

fcs: float

rcs: float

sas: float

hri: float

decision: str

details: Dict[str, Any]

timestamp: datetime

validation_id: str

class Validator:

"""

实现FCS、RCS和SAS验证通道的核心验证器。

"""

def __init__(self, config: Optional[Dict[str, Any]] = None):

self.config = config or {}

# 权重配置

self.weights = {

"fcs": self.config.get("weight_fcs", 0.4),

"rcs": self.config.get("weight_rcs", 0.3),

"sas": self.config.get("weight_sas", 0.3)

}

# 初始化验证组件

self.web_checker = WebFactChecker(

api_keys=self.config.get("api_keys", {}),

cache_size=self.config.get("cache_size", 1000),

timeout_seconds=self.config.get("timeout_seconds", 5)

)

self.logic_checker = LogicConsistencyChecker(

max_propositions=self.config.get("max_propositions", 100),

use_llm_for_complex=self.config.get("use_llm_for_complex", True)

)

self.tspr_checker = TSPRStateChecker(

state_dimension=self.config.get("state_dimension", 64),

decay_factor=self.config.get("decay_factor", 0.95)

)

self.decision_engine = DecisionEngine(

pass_threshold=self.config.get("pass_threshold", 0.2),

rewrite_threshold=self.config.get("rewrite_threshold", 0.5),

max_rewrites=self.config.get("max_rewrites", 3)

)

# 会话状态存储

self.session_states: Dict[str, Dict[str, Any]] = {}

# 性能指标

self.metrics = {

"total_validations": 0,

"total_processing_time_ms": 0,

"fcs_history": [],

"rcs_history": [],

"sas_history": [],

"decision_counts": defaultdict(int)

}

def process(

self,

output: str,

context: Dict[str, Any],

session_id: Optional[str] = None,

applicable_rules: Optional[List[Dict[str, Any]]] = None

) -> Dict[str, Any]:

"""

处理输出并返回验证结果。

参数：

output: LLM生成的输出

context: 包含领域、任务类型的上下文

session_id: 用于状态跟踪的会话ID

applicable_rules: 适用的规则列表

返回：

包含所有分数和决策的字典

"""

import time

start_time = time.time()

# 生成验证ID

validation_id = hashlib.md5(

f"{output}{datetime.now().isoformat()}".encode()

).hexdigest()[:16]

# 获取或创建会话状态

if session_id and session_id in self.session_states:

session_state = self.session_states[session_id]

else:

session_state = self._create_session_state(session_id)

if session_id:

self.session_states[session_id] = session_state

# 1. 执行网络事实检查（FCS）

fcs_result = self.web_checker.verify(

output=output,

context=context,

session_state=session_state

)

# 2. 执行逻辑一致性检查（RCS）

rcs_result = self.logic_checker.check(

output=output,

context=context

)

# 3. 执行TSPR状态检查（SAS）

sas_result = self.tspr_checker.check(

output=output,

context=context,

session_state=session_state

)

# 提取分数

fcs = fcs_result["score"]

rcs = rcs_result["score"]

sas = sas_result["score"]

# 应用动态权重调整

adjusted_weights = self._adjust_weights_by_domain(

context.get("domain", "general")

)

# 计算HRI

hri = self._compute_hri(fcs, rcs, sas, adjusted_weights)

# 应用规则覆盖

if applicable_rules:

hri, decision_override = self._apply_rules(

hri, fcs, rcs, sas, context, applicable_rules

)

else:

decision_override = None

# 做出决策

if decision_override:

decision = decision_override

else:

decision = self.decision_engine.decide(

hri=hri,

rewrite_attempt=context.get("rewrite_attempt", 0),

context=context

)

# 更新会话状态

self._update_session_state(

session_state=session_state,

output=output,

fcs=fcs,

rcs=rcs,

sas=sas,

decision=decision

)

# 更新指标

processing_time = (time.time() - start_time) * 1000

self._update_metrics(fcs, rcs, sas, decision, processing_time)

# 构建详细结果

result = ValidationResult(

fcs=fcs,

rcs=rcs,

sas=sas,

hri=hri,

decision=decision,

details={

"fcs_details": fcs_result["details"],

"rcs_details": rcs_result["details"],

"sas_details": sas_result["details"],

"weights_used": adjusted_weights,

"processing_time_ms": processing_time,

"rules_applied": [r.get("id") for r in (applicable_rules or [])]

},

timestamp=datetime.now(),

validation_id=validation_id

)

return {

"fcs": result.fcs,

"rcs": result.rcs,

"sas": result.sas,

"hri": result.hri,

"decision": result.decision,