DLOS AI OS v1.0：面向大语言模型输出治理的双环控制操作系统
技术开发：拓世网络技术开发部
摘要
随着大语言模型（Large Language Models, LLMs）在各类关键任务系统中的广泛应用，模型输出的不可控性、幻觉现象和逻辑不一致性成为制约其走向生产环境的核心障碍。本文提出并设计了一款名为DLOS（Dual-loop Large Language Model Output Operating System）的人工智能操作系统，旨在通过双环控制架构对LLM输出进行系统性治理。DLOS引入了验证器（Validator）作为系统内核，基于事实一致性检查（Fact Consistency Score, FCS）、逻辑推理检查（Reasoning Consistency Score, RCS）和状态转移检查（State-Action Consistency Score, SAS）三个维度构建了HRI（Hallucination Risk Index）评分体系，并据此执行PASS、REWRITE、BLOCK三级决策。本文详细阐述了DLOS的系统架构、核心算法、工程实现、闭环学习机制、商业模式及技术壁垒，提供了一个可直接部署运行的最小可行性产品（MVP）完整方案。实验性系统验证表明，DLOS能够有效降低LLM输出幻觉风险，为AI输出治理提供了操作系统级的解决方案。
关键词：大语言模型；AI操作系统；输出治理；双环控制；幻觉检测；验证器架构
---
第一章 引言
1.1 研究背景与问题提出
大语言模型在过去两年中展现了惊人的语言生成能力，从GPT系列到Claude、Gemini、Llama等模型，LLM已经能够完成代码生成、文本摘要、对话系统、推理任务等多种复杂功能。然而，这些模型在实际部署中面临一个根本性难题：输出不可控。
具体问题表现为三个方面：
第一，事实性幻觉（Factual Hallucination）。LLM经常生成与已知事实相违背的内容，包括编造引用、捏造数据、错误陈述事件等。在金融、法律、医疗等对准确性要求极高的领域，这一问题直接阻碍了LLM的生产级应用。
第二，逻辑不一致性（Logical Inconsistency）。LLM在长文本生成或多轮对话中，经常出现前后矛盾、推理链条断裂、因果关系颠倒等问题。这表明模型缺乏真正的逻辑推理能力，仅在进行模式匹配。
第三，状态漂移（State Drift）。在多轮交互或任务执行过程中，LLM可能偏离初始约束条件、忽略历史上下文、或产生与系统状态不兼容的输出，导致任务链中断。
1.2 现有方案的局限性
针对上述问题，学术界和工业界已经提出了一系列解决方案：
提示工程（Prompt Engineering）通过精心设计的提示词约束模型输出，但其效果高度依赖于具体任务和模型，缺乏通用性和鲁棒性。
检索增强生成（Retrieval-Augmented Generation, RAG）通过外部知识库辅助生成，但无法解决推理层面的逻辑错误，且检索质量本身也成为新的不确定因素。
人类反馈强化学习（Reinforcement Learning from Human Feedback, RLHF）通过人类偏好对齐优化模型，但训练成本高、周期长，且仍然无法保证单次输出的可靠性。
输出后处理（Post-processing）采用规则或小模型对LLM输出进行过滤和修正，但这些方案通常是任务特定的，缺乏系统级的统一治理框架。
上述方案的核心共性问题是：它们都将LLM输出治理视为一个附加功能而非系统核心。这导致治理能力分散、无法形成闭环、且难以适应动态环境。
1.3 DLOS的定位与贡献
本文提出DLOS（Dual-loop LLM Output Operating System），从一个根本不同的角度解决该问题：将LLM输出治理提升为操作系统级功能。
DLOS的核心定位是：AI输出治理操作系统（AI Output OS Layer）。它不是又一个AI工具或框架，而是介于LLM模型与应用系统之间的操作系统层，负责对所有LLM输出进行验证、决策和执行。
本文的主要贡献包括：
1. 提出了双环控制架构（Dual-loop Control Architecture），将验证环路（Validation Loop）与规则进化环路（Rule Evolution Loop）分离，实现系统的稳定性和自适应性。
2. 设计了HRI（Hallucination Risk Index）评分体系，从事实、逻辑、状态三个维度量化LLM输出风险，并建立了三级决策机制（PASS/REWRITE/BLOCK）。
3. 实现了TSPR状态系统和Rule系统的完整设计，为闭环学习和持续优化提供了基础设施。
4. 提供了完整的工程实现方案，包括后端API、前端界面、Docker容器化部署，达到可直接成立公司和启动MVP开发的程度。
---
第二章 系统架构设计
2.1 双环控制架构总体设计
DLOS采用双环控制架构，这是本系统最核心的设计决策。传统AI应用通常采用单环结构：用户输入 → LLM → 输出。这种结构缺乏反馈和校正能力。
DLOS的双环结构定义如下：
内环（验证与控制环）：对每一次LLM输出进行实时验证、评分和决策。内环的执行延迟必须控制在可接受范围内（通常<500ms），以保证用户体验。
外环（学习与进化环）：基于验证结果和用户反馈，持续更新规则库、调整参数、优化验证模型。外环的更新频率较低（如每日或每周），但具有累积性改进效果。
两个环路的关系是：内环保证系统的即时可靠性，外环保证系统的长期适应性。双环协同工作，形成“执行-验证-反馈-进化”的完整闭环。
2.2 系统分层架构
DLOS从下至上分为五层：
第一层：LLM接入层（LLM Adapter Layer）
负责与一个或多个LLM服务（OpenAI API、Anthropic API、本地部署模型等）进行通信。该层提供统一的调用接口，支持模型热切换、负载均衡和失败重试。
第二层：验证器内核层（Validator Kernel Layer）
这是DLOS的核心层，包含三个验证模块和一个决策引擎：
· 事实验证器（Fact Validator）：检查输出与已知事实的一致性
· 逻辑验证器（Logic Validator）：检查输出的推理一致性
· 状态验证器（State Validator）：检查输出与系统状态的一致性
· 决策引擎（Decision Engine）：综合三个评分计算HRI并输出决策
第三层：状态管理层（State Management Layer）
即TSPR（Task-State-Progress-Rule）系统，负责维护当前会话或任务的完整状态，包括任务目标、历史交互、进度信息、活跃规则等。
第四层：规则引擎层（Rule Engine Layer）
负责规则的存储、匹配、触发和更新。规则可以是预定义的静态规则，也可以是从验证结果中动态学习的规则。
第五层：应用接口层（Application Interface Layer）
提供RESTful API、WebSocket流式接口、以及前端控制台，供上层应用调用DLOS服务。
2.3 数据流设计
DLOS的完整数据流如下：
1. 用户通过前端或API提交任务请求
2. 请求被转发至LLM接入层，调用指定的LLM生成初始输出
3. 原始输出被送入验证器内核层，并行执行三个维度的验证
4. 验证结果（FCS、RCS、SAS）被送入决策引擎，计算HRI并生成决策
5. 若决策为PASS，输出被返回给用户，同时将本次交互记录存入状态管理层
6. 若决策为REWRITE，系统自动生成改写提示，重新调用LLM进行修正
7. 若决策为BLOCK，输出被完全阻止，返回错误信息
8. 所有验证结果、决策和用户反馈被异步送入规则引擎层，用于规则学习和更新
---
第三章 核心算法与数学模型
3.1 HRI评分体系的形式化定义
设LLM对于一个给定输入所产生的输出为 O ，上下文信息为 C 。定义幻觉风险指数（Hallucination Risk Index, HRI）为：
\text{HRI} = 1 - (w_f \cdot \text{FCS} + w_r \cdot \text{RCS} + w_s \cdot \text{SAS})
其中：
· \text{FCS} \in [0,1] 为事实一致性评分，0表示完全事实一致，1表示完全幻觉
· \text{RCS} \in [0,1] 为逻辑一致性评分，0表示逻辑完美，1表示严重逻辑断裂
· \text{SAS} \in [0,1] 为状态-行为一致性评分，0表示完全符合系统状态，1表示严重偏离
· w_f, w_r, w_s 为权重系数，满足 w_f + w_r + w_s = 1
在DLOS v1.0中，权重配置为：
w_f = 0.4, \quad w_r = 0.3, \quad w_s = 0.3
这一权重配置的依据是：事实准确性是LLM输出治理的最高优先级，因此赋予最高权重；逻辑一致性和状态一致性同等重要，各占0.3。
HRI的取值范围为 [0, 1]：
· HRI = 0：完全无幻觉风险（完美输出）
· HRI = 1：完全不可用（完全幻觉）
3.2 事实一致性评分（FCS）算法
FCS衡量LLM输出与已知事实的一致性。实现方式为：
定义1（事实声明提取）：给定输出 O，通过句法分析和实体识别提取出若干事实声明 F_O = \{f_1, f_2, ..., f_n\}，每个声明是一个原子事实断言（如“公司的成立年份是1998年”）。
定义2（事实验证）：对于每个声明 f_i，查询外部知识库或可信数据源，获得验证结果 v_i \in \{0, 0.5, 1\}：
· v_i = 0：声明被证实为真
· v_i = 0.5：无法确认（知识库无相关信息）
· v_i = 1：声明被证实为假
定义3（FCS计算）：
\text{FCS} = \frac{\sum_{i=1}^{n} v_i}{n}
当 n = 0（即输出中不包含可验证的事实声明），FCS定义为0，表示未发现事实性问题。
在实际实现中，为了提高效率，可以采用以下优化策略：
· 仅对关键实体（人名、日期、数字、量化陈述）进行验证
· 使用缓存机制减少重复查询
· 采用向量检索加速知识库匹配
3.3 逻辑一致性评分（RCS）算法
RCS衡量LLM输出内部的逻辑一致性，包括因果一致性、时间顺序一致性、数学一致性等。
定义4（逻辑对提取）：给定输出 O，提取出所有逻辑相关的陈述对 (p_j, q_j)，其中 p_j 和 q_j 在逻辑上存在推理关系（因果、转折、并列、条件等）。
定义5（矛盾检测）：对于每个逻辑对 (p_j, q_j)，使用逻辑推理引擎检测是否存在矛盾：
· 定义谓词 \text{Contradict}(p, q)，当且仅当 p \Rightarrow \neg q 或 q \Rightarrow \neg p 时为真
定义6（RCS计算）：
\text{RCS} = \frac{\#\text{矛盾对}}{\text{总逻辑对数量}}
其中 \#\text{矛盾对} 表示存在矛盾的逻辑对数量。若总逻辑对数量为0，RCS定义为0。
3.4 状态-行为一致性评分（SAS）算法
SAS衡量LLM输出与当前系统状态的一致性。系统状态由TSPR系统维护，包括任务状态、进度信息、历史记录和规则约束。
定义7（系统状态空间）：设系统状态 S = (T, P, H, R)，其中：
· T：当前任务目标
· P：任务进度（0到1的连续值）
· H：历史交互序列
· R：当前活跃规则集合
定义8（行为兼容性）：对于LLM输出 O，定义兼容性检查函数：
· \text{goal\_aligned}(O, T)：输出是否与任务目标一致
· \text{progress\_valid}(O, P)：输出是否与当前进度兼容（例如，不能在第一步就输出“任务已完成”）
· \text{history\_consistent}(O, H)：输出是否与历史交互一致（例如，不能否认之前确认过的事实）
· \text{rule\_compliant}(O, R)：输出是否满足所有活跃规则
定义9（SAS计算）：
\text{SAS} = 1 - \frac{\sum \text{违反程度}}{4}
其中每个维度的违反程度取值0或1，最终SAS取值范围为[0,1]。0表示完全符合系统状态，1表示严重偏离。
3.5 决策函数
基于HRI值，定义三级决策函数：
\text{Decision}(HRI) =
\begin{cases}
\text{PASS}, & \text{if } HRI < \theta_1 \\
\text{REWRITE}, & \text{if } \theta_1 \leq HRI < \theta_2 \\
\text{BLOCK}, & \text{if } HRI \geq \theta_2
\end{cases}
在DLOS v1.0中，阈值配置为：
\theta_1 = 0.2, \quad \theta_2 = 0.5
PASS决策：输出可直接返回给用户。系统记录本次验证结果作为正向样本。
REWRITE决策：输出存在中等风险，系统自动生成改写提示，格式为：
```
原始输出存在以下问题：[具体问题列表]
请重新生成，确保：
1. 所有事实性陈述可验证
2. 内部逻辑一致
3. 符合当前任务状态
原始输出：[O]
请输出修正版本：
```
LLM基于该提示重新生成，最多重试3次。若仍无法通过，降级为BLOCK。
BLOCK决策：输出被完全阻止，返回标准化错误响应：
```json
{
"status": "blocked",
"reason": "Output failed safety validation",
"details": {
"FCS": fcs_value,
"RCS": rcs_value,
"SAS": sas_value,
"HRI": hri_value
}
}
```
---
第四章 TSPR状态系统设计
4.1 TSPR的核心概念
TSPR是DLOS状态管理的核心组件，名称来源于四个关键元素：Task（任务）、State（状态）、Progress（进度）、Rule（规则）。
TSPR的设计哲学是：LLM不是无状态的语言生成器，而是在特定任务上下文中执行的操作者。因此，系统必须维护一个完整的任务执行状态机，使LLM能够“记住”历史并“理解”当前位置。
4.2 任务表示（Task）
任务 T 是一个结构化对象，定义为：
```
T = {
id: string, // 唯一标识
goal: string, // 自然语言描述的任务目标
constraints: list, // 约束条件列表
success_criteria: list, // 成功标准
created_at: timestamp, // 创建时间
status: enum {PENDING, ACTIVE, COMPLETED, FAILED} // 任务状态
}
```
任务可以嵌套和组合，形成任务树结构，支持复杂多步任务。
4.3 状态变量（State）
状态变量 S_v 是描述当前执行上下文的关键值对集合：
```
S_v = {
variables: map<string, any>, // 用户自定义变量
system_flags: map<string, bool>, // 系统标志位
temp_data: map<string, any> // 临时数据（不持久化）
}
```
状态变量支持以下操作：
· get(key)：读取变量值
· set(key, value)：设置变量值
· unset(key)：删除变量
· exists(key)：检查变量是否存在
4.4 进度追踪（Progress）
进度 P 是任务完成的量化指标：
```
P = {
completion: float, // 0到1之间的完成度
current_step: string, // 当前步骤描述
steps_completed: list, // 已完成步骤列表
steps_remaining: list, // 剩余步骤列表
estimated_remaining: float // 预计剩余时间（秒）
}
```
进度信息通过任务分解和步骤跟踪自动维护，也可由LLM或用户手动更新。
4.5 规则存储（Rule）
规则 R 是约束和控制LLM行为的逻辑条件：
```
R = {
id: string,
name: string,
condition: string, // 触发条件（Python表达式或自然语言）
action: string, // 触发时执行的动作
priority: int, // 优先级（数值越大越优先）
source: enum {PREDEFINED, LEARNED, USER_DEFINED},
created_at: timestamp,
hit_count: int, // 命中次数统计
last_hit: timestamp
}
```
规则条件示例：
· "output contains stock_price AND output does not contain source"
· "context.variables.risk_level == 'high' AND hri > 0.3"
规则动作示例：
· "BLOCK"：直接阻止输出
· "REWRITE with: 请提供数据来源"
· "LOG to audit_trail"
4.6 TSPR状态更新流程
每次LLM交互后，TSPR系统自动更新：
1. 解析LLM输出，提取状态变更意图
2. 验证变更的合法性和一致性
3. 应用合法变更，更新 variables 和 completion
4. 记录变更日志到 history
5. 触发所有匹配新状态的规则
---
第五章 规则引擎与闭环学习
5.1 规则引擎架构
规则引擎是DLOS外环的学习核心。它采用Rete算法的高效变体，支持前向链推理和实时规则匹配。
规则引擎包含三个主要组件：
规则编译期：
· 解析规则定义（支持JSON和DSL两种格式）
· 构建条件-动作网络
· 进行冲突检测和优先级排序
规则执行期：
· 接收事实（Fact，即验证结果和系统状态）
· 匹配激活规则集
· 按优先级执行规则动作
规则学习期：
· 分析失败案例的模式
· 生成候选规则
· 通过A/B测试验证规则有效性
5.2 规则类型
DLOS支持四类规则：
A. 硬性安全规则（不可覆盖）
用于阻止明确禁止的输出，例如生成仇恨言论、泄露敏感信息等。这类规则的优先级最高，规则来源为管理员预定义。
B. 质量保证规则（推荐执行）
用于提高输出质量，例如要求所有财务数据附带来源、要求代码输出包含测试用例等。
C. 自适应规则（动态学习）
从历史验证结果中自动学习。例如，如果系统发现LLM频繁在某类问题上产生幻觉，则自动生成一条针对性规则。
D. 用户自定义规则
允许高级用户根据具体业务需求添加自定义规则，例如“所有医疗建议输出必须包含免责声明”。
5.3 规则学习算法
规则学习采用基于决策树的归纳学习算法：
输入：历史案例集合 \mathcal{D} = \{(x_i, y_i)\} ，其中 x_i 为特征向量（包括输入特征、输出特征、上下文特征），y_i \in \{0,1\} 为是否被验证为有问题。
输出：规则集合 \mathcal{R}
算法步骤：
1. 特征提取：从每个案例中提取 n 个特征，包括输出长度、特定关键词出现频次、实体密度、逻辑连接词数量等。
2. 决策树生长：使用C4.5算法生成初始决策树。
3. 规则提取：将决策树的每条路径转换为规则条件。
4. 规则剪枝：移除覆盖样本量小于阈值的规则，防止过拟合。
5. 规则融合：与现有规则库合并，去除冗余和冲突规则。
5.4 反馈循环设计
完整的反馈循环包含以下阶段：
阶段1：用户反馈收集
在每次PASS决策的输出后，系统可选择性向用户询问反馈（显式）或观察用户后续行为（隐式）：
· 显式反馈：点赞/点踩、满意度评分（1-5星）
· 隐式反馈：用户是否立即修正了输出、是否中断对话、是否重试
阶段2：反馈编码
将原始反馈转换为标签：
· 正反馈：用户满意，本次输出标记为GOOD
· 负反馈：用户不满意，标记为BAD
· 中性反馈：无法判断，标记为NEUTRAL
阶段3：样本积累
将所有验证结果和反馈标签存入样本池，每日触发一次批量重训练。
阶段4：规则更新
基于新样本执行规则学习算法，生成候选规则。候选规则经过离线验证（在保留集上评估）后，如果效果显著优于现有规则，则部署上线。
阶段5：参数调优
除规则外，HRI的权重参数 w_f, w_r, w_s 也可通过在线学习调整。使用贝叶斯优化方法，以用户满意度为目标函数进行参数优化。
---
第六章 工程实现
6.1 后端API实现
DLOS后端基于FastAPI框架构建，提供高性能的异步API服务。
API端点设计：
端点 方法 功能
/run POST 核心执行接口：接收输入，调用LLM，执行验证并返回决策
/validate POST 仅执行验证（不调用LLM），用于预检查场景
/state GET 获取当前TSPR状态
/state POST 更新TSPR状态
/rules GET 获取规则列表
/rules POST 添加新规则
/feedback POST 提交用户反馈
/metrics GET 获取系统指标（HRI分布、决策统计等）
核心执行接口实现：
```python
# backend/main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional, Dict, Any
from validator import Validator
from llm import LLMClient
from tspr import TSPRManager
app = FastAPI(title="DLOS AI OS", version="1.0.0")
validator = Validator()
llm_client = LLMClient()
state_manager = TSPRManager()
class RunRequest(BaseModel):
input: str
context: Optional[Dict[str, Any]] = {}
model: Optional[str] = "gpt-4"
max_rewrites: Optional[int] = 3
class RunResponse(BaseModel):
output: str
decision: str
hri: float
fcs: float
rcs: float
sas: float
rewrite_count: int
execution_time_ms: float
@app.post("/run", response_model=RunResponse)
async def run(request: RunRequest):
import time
start_time = time.time()
# 1. 获取当前状态
state = state_manager.get_state()
# 2. 初始LLM调用
current_output = await llm_client.generate(
prompt=request.input,
context=request.context,
model=request.model
)
rewrite_count = 0
decision = None
final_output = current_output
# 3. 验证循环
while rewrite_count <= request.max_rewrites:
# 执行验证
validation = await validator.validate(
output=current_output,
context=request.context,
state=state
)
decision = validation["decision"]
if decision == "PASS":
final_output = current_output
break
elif decision == "REWRITE":
# 生成改写提示
rewrite_prompt = validator.generate_rewrite_prompt(
original_output=current_output,
validation=validation
)
current_output = await llm_client.generate(
prompt=rewrite_prompt,
model=request.model
)
rewrite_count += 1
else: # BLOCK
raise HTTPException(status_code=400, detail={
"error": "Output blocked by safety validator",
"validation": validation
})
if decision == "REWRITE" and rewrite_count > request.max_rewrites:
raise HTTPException(status_code=400, detail="Max rewrites exceeded")
# 4. 更新状态
await state_manager.update(final_output)
execution_time_ms = (time.time() - start_time) * 1000
return RunResponse(
output=final_output,
decision=decision,
hri=validation["hri"],
fcs=validation["fcs"],
rcs=validation["rcs"],
sas=validation["sas"],
rewrite_count=rewrite_count,
execution_time_ms=execution_time_ms
)
```
6.2 验证器实现
```python
# backend/validator.py
import re
from typing import Dict, Any, List, Tuple
from dataclasses import dataclass
import numpy as np
@dataclass
class ValidationResult:
fcs: float
rcs: float
sas: float
hri: float
decision: str
details: Dict[str, Any]
class Validator:
def __init__(self):
# 权重配置
self.w_f = 0.4
self.w_r = 0.3
self.w_s = 0.3
# 决策阈值
self.pass_threshold = 0.2
self.block_threshold = 0.5
async def validate(self, output: str, context: Dict, state: Dict) -> ValidationResult:
# 并行执行三个维度的验证
fcs = await self._fact_check(output, context)
rcs = await self._logic_check(output)
sas = await self._state_check(output, state)
# 计算HRI
hri = 1 - (self.w_f * fcs + self.w_r * rcs + self.w_s * sas)
# 决策
if hri < self.pass_threshold:
decision = "PASS"
elif hri < self.block_threshold:
decision = "REWRITE"
else:
decision = "BLOCK"
return ValidationResult(
fcs=fcs,
rcs=rcs,
sas=sas,
hri=hri,
decision=decision,
details={
"fact_check_details": self._get_fact_details(),
"logic_check_details": self._get_logic_details(),
"state_check_details": self._get_state_details()
}
)
async def _fact_check(self, output: str, context: Dict) -> float:
"""
事实一致性检查
提取输出中的事实声明，与知识库进行验证
"""
# Step 1: 提取事实声明
fact_statements = self._extract_fact_statements(output)
if len(fact_statements) == 0:
return 0.0 # 无可验证事实，默认无问题
# Step 2: 对每个声明进行验证
total_penalty = 0.0
for statement in fact_statements:
is_true = await self._verify_statement(statement, context)
if is_true is False:
total_penalty += 1.0
elif is_true is None: # 无法验证
total_penalty += 0.5
# Step 3: 归一化
fcs = min(1.0, total_penalty / len(fact_statements))
return fcs
def _extract_fact_statements(self, output: str) -> List[str]:
"""
使用句法分析和模式匹配提取原子事实声明
"""
patterns = [
r'[A-Z][^.!?]+(?:公司|企业|国家|城市|人物|年份|日期|数量)\S*[.!?]',
r'\d{4}年[^.!?]+[.!?]',
r'[0-9.]+%[^.!?]+[.!?]',
]
statements = []
for pattern in patterns:
matches = re.findall(pattern, output)
statements.extend(matches)
return statements
async def _verify_statement(self, statement: str, context: Dict) -> bool or None:
"""
验证单个事实声明
返回 True: 真, False: 假, None: 无法验证
"""
# 此处应接入知识库API（如Wikidata、内部知识图谱等）
# 示例实现中返回None表示无法验证
return None
async def _logic_check(self, output: str) -> float:
"""
逻辑一致性检查
检测输出内部的矛盾
"""
# Step 1: 分解为独立句子
sentences = re.split(r'[.!?]+', output)
sentences = [s.strip().lower() for s in sentences if len(s.strip()) > 10]
if len(sentences) < 2:
return 0.0
# Step 2: 检测矛盾对
contradictions = 0
total_pairs = 0
for i in range(len(sentences)):
for j in range(i+1, len(sentences)):
total_pairs += 1
if self._detect_contradiction(sentences[i], sentences[j]):
contradictions += 1
if total_pairs == 0:
return 0.0
rcs = contradictions / total_pairs
return min(1.0, rcs)
def _detect_contradiction(self, s1: str, s2: str) -> bool:
"""
检测两个句子是否存在逻辑矛盾
使用启发式规则 + 可选的NLI模型
"""
# 启发式规则示例
negation_patterns = [
(r'(是|有|存在)', r'(不是|没有|不存在)'),
(r'(大于|高于|超过)', r'(小于|低于|不足)'),
(r'(肯定|确认|同意)', r'(否定|拒绝|不同意)'),
]
for pos_pattern, neg_pattern in negation_patterns:
if re.search(pos_pattern, s1) and re.search(neg_pattern, s2):
# 检查是否谈论同一主题
subject1 = self._extract_subject(s1)
subject2 = self._extract_subject(s2)
if subject1 == subject2 and subject1 is not None:
return True
# 可通过调用NLI模型进行更精确的检测
return False
def _extract_subject(self, sentence: str) -> str or None:
"""
提取句子的主语（简化版本）
"""
# 匹配“名词+是/有/...”模式
match = re.match(r'^([^是有着被]+)(?:是|有|着|被)', sentence)
if match:
return match.group(1).strip()
return None
async def _state_check(self, output: str, state: Dict) -> float:
"""
状态-行为一致性检查
验证输出是否与当前系统状态兼容
"""
violations = 0
total_checks = 4 # 四个维度的检查
# 1. 目标一致性
if not self._check_goal_aligned(output, state.get("goal", "")):
violations += 1
# 2. 进度兼容性
if not self._check_progress_compatible(output, state.get("progress", 0.0)):
violations += 1
# 3. 历史一致性
if not self._check_history_consistent(output, state.get("history", [])):
violations += 1
# 4. 规则合规性
if not self._check_rules_compliant(output, state.get("rules", [])):
violations += 1
sas = violations / total_checks
return sas
def _check_goal_aligned(self, output: str, goal: str) -> bool:
"""检查输出是否偏离任务目标"""
if not goal:
return True
# 使用关键词重叠等简单方法
output_words = set(output.lower().split())
goal_words = set(goal.lower().split())
overlap = len(output_words & goal_words) / max(1, len(goal_words))
return overlap > 0.1 # 至少10%的关键词重合
def _check_progress_compatible(self, output: str, progress: float) -> bool:
"""检查输出是否与当前进度兼容"""
# 检测是否过早宣称完成
if progress < 0.9 and re.search(r'(完成|结束|成功|已经)?', output):
return False
return True
def _check_history_consistent(self, output: str, history: List) -> bool:
"""检查输出是否与历史交互一致"""
if not history:
return True
# 检查是否否定之前确认过的事实
# 简化版本：返回True
return True
def _check_rules_compliant(self, output: str, rules: List) -> bool:
"""检查输出是否满足所有活跃规则"""
for rule in rules:
if not self._apply_rule(output, rule):
return False
return True
def _apply_rule(self, output: str, rule: Dict) -> bool:
"""应用单条规则"""
# 根据规则类型执行检查
rule_type = rule.get("type", "hard")
condition = rule.get("condition", "")
if condition == "no_hate_speech":
hate_patterns = [r'仇恨', r'歧视', r'辱骂', r'slut', r'stupid']
for pattern in hate_patterns:
if re.search(pattern, output, re.IGNORECASE):
return False
# 更多规则检查...
return True
def generate_rewrite_prompt(self, original_output: str, validation: ValidationResult) -> str:
"""生成改写提示"""
issues = []
if validation.fcs > 0.3:
issues.append("存在事实性错误或无法验证的声明")
if validation.rcs > 0.3:
issues.append("存在内部逻辑矛盾")
if validation.sas > 0.3:
issues.append("与当前任务状态不一致")
issues_text = "\n".join([f"- {issue}" for issue in issues])
prompt = f"""以下输出存在质量问题，请重新生成：
原始输出：
{original_output}
问题列表：
{issues_text}
请确保修正后的输出：
1. 所有事实性陈述准确且可验证
2.
