企业官网建设流程全解析

Upsonic Reliability Layer 深度解析：基于 Verifier 与 Editor 子代理的幻觉防御管线

【免费下载链接】gpt-computer-assistantBuild autonomous AI agents in Python.项目地址: https://gitcode.com/GitHub_Trending/gp/gpt-computer-assistant

导读

reliability_layer/是 Upsonic（构建自主 AI Agent 的 Python 框架）中负责生成后幻觉防御的核心模块：它在Agent产出回复之后、把回复交还给调用方之前，通过一组小型verifier（验证）子代理对原始任务上下文重新校验输出，再由editor（编辑）子代理清洗被标记的可疑内容，从而识别并去除幻觉——包括编造的 URL、凭空捏造的数字、无法溯源的论断与合成代码。读完本文，你将掌握该层的架构、prevent_hallucination分级语义、ReliabilityProcessor.process_task的完整执行流程、子代理用量核算方式，以及如何为它新增一种自定义 verifier。

1. 这个模块是什么

src/upsonic/reliability_layer/实现了 Upsonic 的生成后可靠性管线（post-generation reliability pipeline）：它运行在Agent产生响应之后、响应被交还调用方之前，是一道额外的防御层。其目标是在 Agent 输出中检测并移除幻觉——编造的 URL、虚构的数字、不可追溯的论断以及被合成的代码——手段是：针对原始任务上下文重新运行一组小型verifier 子代理，再将任何被标记的内容交给editor 子代理清洗（或将可疑值置为None）。

概念上，这是一轮迭代式的agent-on-agent 质量提升：

整个层级是**可选启用（opt-in）**的：没有传reliability_layer参数的 agent 直接原样通过。启用后（通常通过一个暴露prevent_hallucination >= 10的配置对象），每次成功响应都会被扇出校验，且只有明显可疑的响应才承担 editor 重写的额外成本。

该目录本身刻意保持极小——仅一个文件——因为所有复杂度都集中在一个纪律严明的模块里。其余一切都位于该目录之外、agent 管线之中。

2. 目录结构

src/upsonic/reliability_layer/ └── reliability_layer.py # 唯一源文件——整个可靠性子系统

这就是全部目录树：没有子包、没有__init__.py再导出、没有辅助模块。使用方直接深入引用：

from upsonic.reliability_layer.reliability_layer import ReliabilityProcessor

作为对照，以下是外部代码引用它的方式：

src/upsonic/ ├── reliability_layer/ │ └── reliability_layer.py # ReliabilityProcessor + helpers + prompts ├── agent/ │ ├── agent.py # Agent.reliability_layer 属性（约 L249/L465） │ ├── pipeline/ │ │ └── steps.py # ReliabilityStep（L2485）— 管线集成 │ └── context_managers/ │ └── reliability_manager.py # ReliabilityManager — 轻量 async 封装

3. 顶层文件：reliability_layer.py

这个单一文件按源码顺序划分为六个逻辑区域。

3.1 区域地图

3.2strip_context_tags(text: str) -> str

Upsonic 的任务管线会在上下文块周围注入框架标签。当该上下文转发给 verifier 时，标签会膨胀提示词并干扰模式识别。该辅助函数执行一趟幂等的正则替换，移除以下标签的开闭实例（见 reliability_layer.py）：

• <Context>/</Context>
• <Knowledge Base>/</Knowledge Base>
• <Agents>/</Agents>
• <Tasks>/</Tasks>
• <Default Prompt>/</Default Prompt>

移除标签后，它会折叠连续的空白行（\n\s*\n→\n\n）并 trim。任何非字符串输入都会被原样返回，因此可以安全地在混合类型上下文条目的列表推导式中调用。

3.3 校验提示词（verifier 的“评判标准”）

四条提示词共享同一个反幻觉论点：

“检查来源是否来自该内容。不要做假设，只检查上下文并尝试找到精确的事物。如果找不到就标记它。如果你能在上下文中看到这些内容，那么一切都没问题（Trusted Source / 可信来源）。”

这四条提示词都被写成推动 verifier 走向保守拒绝（conservative refusal）：无法验证的条目必须被标记，而不是被合理化放过。下方 editor 提示词则负责清理后果。

3.4editor_task_prompt

单条提示词模板，含一个{validation_feedback}插槽，规则刻意严格（完整文本见 reliability_layer.py）：

Processing Rules: 1. For ANY suspicious content identified in validation: - Replace the suspicious value with None - Do not suggest alternatives - Do not provide explanations - Do not modify other parts of the content 2. For non-suspicious content: - Keep the original value unchanged - Do not enhance or modify - Do not add additional information Processing Steps: - Set suspicious fields to None - Keep other fields as is - Remove any suspicious content entirely - Maintain original structure IMPORTANT: - Set ALL suspicious values to None - Keep verified values unchanged - No explanations or suggestions - No partial validations - Maintain response format - DO NOT output XML, code blocks, or any markup - Return only the clean content in the same format as the original

editor从不“修复”幻觉（无法修复凭空捏造的东西），它只是删除。这是一个关键不变量：该层是减法过滤器（subtractive filter），而非纠正性重写器。

3.5 Pydantic 模型

class SourceReliability(Enum): HIGH = "high" MEDIUM = "medium" LOW = "low" UNKNOWN = "unknown" class ValidationPoint(BaseModel): is_suspicious: bool feedback: str suspicious_points: list[str] = Field(description = "Suspicious informations raw name") source_reliability: SourceReliability = SourceReliability.UNKNOWN verification_method: str = "" confidence_score: float = 0.0 class ValidationResult(BaseModel): url_validation: ValidationPoint number_validation: ValidationPoint information_validation: ValidationPoint code_validation: ValidationPoint any_suspicion: bool suspicious_points: list[str] overall_feedback: str overall_confidence: float = 0.0

每个 verifier 代理都以response_format=ValidationPoint配置，因此它被强制返回结构化 JSON 对象；编排器随后把每个ValidationPoint填入单个ValidationResult的对应槽位。

ValidationResult.calculate_suspicion()

一个纯聚合器——不发起任何 LLM 调用（实现见 reliability_layer.py）：

1. 设置any_suspicion = 四个 is_suspicious 标志的 OR。
2. 把每个 verifier 的suspicious_points拼接成扁平列表。
3. 对每个可疑的 verifier，构建一段“标题 + 项目符号列表”的反馈小节（如URL Issues: ...）。
4. 计算overall_confidence = 四个 confidence_score 的均值。
5. 把一切拼成多行 “Validation Summary” 字符串并返回。（既产生副作用又返回值：调用方把它存到模型上，同时把返回的字符串传给 editor。）

4. 无子目录：扁平设计的取舍

该包没有子文件夹——就是一个 Python 模块。这是刻意为之：可靠性层只有一项工作，用一个文件实现，提示词、数据模型、编排器和辅助正则都可以在一屏滚动内完成审阅。

若未来新增子目录，自然的划分会是：

但在增加更多可靠性策略（例如基于引用的校验、工具调用重放、形式化验证）之前，扁平布局是正确的选择。

5. 跨文件关系

可靠性层不是自包含库——它被多个 Upsonic 子系统调用，并向其贡献数据。

┌───────────────────────────────────────────────────────────────────────┐ │ Agent.do_async / Agent.run │ │ └── pipeline.run_step(...) │ │ ├── ... (other steps) ... │ │ └── ReliabilityStep (src/upsonic/agent/pipeline/ │ │ │ steps.py — L2485) │ │ │ │ │ ▼ │ │ ReliabilityManager (src/upsonic/agent/context_ │ │ │ managers/reliability_manager.py│ │ │ │ │ ▼ │ │ ReliabilityProcessor.process_task (reliability_layer.py) │ │ │ │ │ ├──► 生成 AgentConfiguration("URL Validation Agent") │ │ ├──► 生成 AgentConfiguration("Number Validation Agent") │ │ ├──► 生成 AgentConfiguration("Information Validation A.") │ │ ├──► 生成 AgentConfiguration("Code Validation Agent") │ │ │ (asyncio.gather 并行执行) │ │ │ │ │ └──► 若有任意怀疑： │ │ 生成 AgentConfiguration("Information Editor Agent") │ │ 重写 task._response │ └───────────────────────────────────────────────────────────────────────┘

5.1 编排器从外部拉取的输入

5.2 编排器写回的输出

5.3 从本目录导入的文件

src/upsonic/agent/context_managers/reliability_manager.py from upsonic.reliability_layer.reliability_layer import ReliabilityProcessor

这是代码库中唯一的直接导入。其余所有交互都由 manager 中介。管线步骤（ReliabilityStep）导入的是ReliabilityManager（经由 src/upsonic/agent/context_managers/init.py 的惰性工厂），从不直接导入ReliabilityProcessor。

6. 公共 API

该目录暴露了刻意收窄的接口面。只有ReliabilityProcessor.process_task意图被模块外部代码调用；其余内容虽然可自由导入，但属于内部实现。

6.1ReliabilityProcessor

class ReliabilityProcessor: def __init__(self, confidence_threshold: float = 0.7) -> None: ... @staticmethod async def process_task( task: "Task", reliability_layer: Optional[Any] = None, model: Optional[Union[Model, str]] = None, ) -> "Task": """ 变更 task，使任何可验证的幻觉被移除。 行为矩阵 ----------------- reliability_layer is None -> 原样返回 task prevent_hallucination == 0 -> 原样返回 task prevent_hallucination == 10 -> 运行完整 verifier+editor 管线 其他正值 -> 保留（当前为 no-op） 返回同一个 task 实例。若有怀疑，task._response 被替换为 editor 清洗后的响应。 """

注意：__init__中的confidence_threshold（默认 0.7）字段会被存储，但当前并不被process_task消费，属于为未来“软怀疑”门控预留（见 reliability_layer.py）。此外prevent_hallucination若是 property，处理器会通过property.fget(reliability_layer)取值（见 reliability_layer.py）。

6.2 Pydantic 模型

class SourceReliability(Enum): HIGH | MEDIUM | LOW | UNKNOWN class ValidationPoint(BaseModel): is_suspicious: bool feedback: str suspicious_points: list[str] source_reliability: SourceReliability verification_method: str confidence_score: float class ValidationResult(BaseModel): url_validation: ValidationPoint number_validation: ValidationPoint information_validation: ValidationPoint code_validation: ValidationPoint any_suspicion: bool suspicious_points: list[str] overall_feedback: str overall_confidence: float def calculate_suspicion(self) -> str: ...

6.3 自由函数（正则层）

def strip_context_tags(text: str) -> str: ... def find_urls_in_text(text: str) -> List[str]: ... def find_numbers_in_text(text: str) -> List[str]: ... def find_code_in_text(text: str) -> bool: ... def contains_urls(texts: List[str]) -> bool: ... def contains_numbers(texts: List[str]) -> bool: ... def contains_code(texts: List[str]) -> bool: ...

具体正则实现（reliability_layer.py）：

• find_urls_in_text匹配http[s]://...（也覆盖ftp://）。
• find_numbers_in_text匹配整数、浮点、百分比、货币（$1.99）、科学计数法（1e5）。
• find_code_in_text扫描代码围栏（...）、行内反引号、关键字（def/class/import等）、常见代码标点{}[]();、f(x)形 token 与obj.method模式，返回bool。
• contains_*批量版本会跳过非字符串条目。

注意：没有find_information_in_text——信息永远是候选，所以information_validation始终运行。

6.4 模块级提示词常量（字符串类型）

url_validation_prompt number_validation_prompt information_validation_prompt code_validation_prompt editor_task_prompt # 含 {validation_feedback} 插槽

它们在“可导入”意义上是公开的，但属于实现契约的一部分、可能变化。使用方应将其视为不透明内容。

7. 与 Upsonic 其余部分的集成

7.1 如何嵌入 agent 管线

ReliabilityStep作为标准 agent 管线中的一个固定步骤注册（见 src/upsonic/agent/agent.py）：

# src/upsonic/agent/agent.py — 标准管线摘录 ModelExecutionStep(), # 14 ResponseProcessingStep(), # 15 ReflectionStep(), # 16 TaskManagementStep(), # 17 ReliabilityStep(), # 18 <-- "Verify and clean output" AgentPolicyStep(), # 19 CacheStorageStep(), # 20 FinalizationStep(), # 21 MemorySaveStep(), # 22

在流式管线（stream_*）中该步骤同样存在（紧随ReflectionStep之后）。流式变体中，步骤还会通过ayield_reliability_event发出一个reliability事件（见 src/upsonic/utils/agent/events.py），让下游观察者知道是否发生了修改；对应事件类型为 src/upsonic/run/events/events.py 中的ReliabilityEvent。

7.2 如何启用

Agent构造时接受可选的reliability_layer参数（任何带有prevent_hallucination整数属性的对象，见 agent.py 与属性存储处 agent.py）：

class HighReliability: prevent_hallucination = 10 # 完整管线 ON agent = Agent( model="openai/gpt-4o", reliability_layer=HighReliability(), )

同样的机制也流过AutonomousAgent.__init__，它会把自己的reliability_layer转发给内部的Agent（见 src/upsonic/agent/autonomous_agent/autonomous_agent.py 与转发处 autonomous_agent.py）。

7.3 管线步骤在处理器周围做了什么

ReliabilityStep.execute的真实实现（steps.py）要点如下：

if not agent.reliability_layer: # no-op + （若流式）发出 reliability_applied=False 事件 return StepResult(..., status=COMPLETED, message="No reliability layer") if task._cached_result: return StepResult(..., message="Skipped due to cache hit") if task._policy_blocked: return StepResult(..., message="Skipped due to policy block") original_output = context.output reliability_manager = ReliabilityManager(task, agent.reliability_layer, model) await reliability_manager.aprepare() try: processed_task = await reliability_manager.process_task(task) task = processed_task context.output = processed_task.response finally: await reliability_manager.afinalize() # 子代理用量已通过 contextvars 进入注册表；清理暂存字段即可 task._reliability_sub_agent_usage = None modifications_made = str(original_output) != str(context.output)

有三点值得注意：

1. 缓存与策略阻断会绕过该层。若任务已返回缓存答案或被AgentPolicyStep阻断，就没有新内容可验证——运行它会浪费 token。ReliabilityStep.execute在构造 manager 之前先检查task._cached_result与task._policy_blocked。
2. 用量遥测被保留。verifier 与 editor 的调用在处理器内部累加进task._reliability_sub_agent_usage（RunUsage）；由于子代理通过 contextvars 继承父作用域标签，其 LLM 用量已经进入注册表，步骤只需把暂存字段清空即可（见 steps.py）。
3. 原始输出在处理前被捕获，因此步骤可以计算modifications_made = str(original_output) != str(context.output)，供流式事件与调试日志使用。

7.4 遥测

当agent.debug且agent.debug_level >= 2时，步骤发出一个标记为"ReliabilityStep"的debug_log_level2条目，包含：

{ "modifications_made": bool, "original_output_preview": str[:300], "processed_output_preview": str[:300], "reliability_layer_type": type(agent.reliability_layer).__name__, }

这是诊断误报（false positives）的主要可观测性钩子。

8. 端到端流程

下面完整走一遍prevent_hallucination = 10路径：从ReliabilityStep.execute调用ReliabilityProcessor.process_task开始。

8.1 阶段 A —— 提前退出检查

if reliability_layer is None: return task # 未挂载层 prevent_hallucination = getattr(reliability_layer, 'prevent_hallucination', 0) if isinstance(prevent_hallucination, property): prevent_hallucination = prevent_hallucination.fget(reliability_layer) if prevent_hallucination == 0: return task # 显式禁用 if prevent_hallucination != 10: return task # 目前只有等级 10 被接线

8.2 阶段 B —— 快照并解包响应

copy_task = deepcopy(task) copy_task._response = task.response # 尝试提取包装器下的原始 output old_task_output = task.response if hasattr(task.response, 'output'): old_task_output = task.response.output elif hasattr(task.response, 'model_dump'): old_task_output = task.response.model_dump()

deepcopy 保证原始task在处理过程中被保留，同时处理器为 verifier 构建上下文。

8.3 阶段 C —— 构建共享上下文包

同一上下文块被发给每个 verifier：

context_strings = [f"Given Task: {copy_task.description}"] if copy_task.context: items = copy_task.context if isinstance(copy_task.context, list) else [copy_task.context] if copy_task.response_format: items.append(copy_task.response_format) for item in items: if isinstance(item, str): cleaned = strip_context_tags(item) context_strings.append( f"Context That Came From User (Trusted Source): " f"```User given context {cleaned}```" ) elif item.__bases__[0].__name__ == BaseModel.__name__: context_strings.append( f"User requested output: ```Requested Output {item.model_fields}```" ) context_strings.append( f"Current AI Response (Untrusted Source, last AI response that we are checking now): " f"{old_task_output}" ) context_strings = [strip_context_tags(s) if isinstance(s, str) else s for s in context_strings]

这是信任模型的核心：用户提供的上下文被标记为Trusted Source（可信来源），AI 自己的输出被标记为Untrusted Source（不可信来源），verifier 被告知检查前者是否扎根于后者。

8.4 阶段 D —— 生成 verifier 子代理

对四种校验类型分别执行：

8.5 阶段 E —— 并行运行 verifier

validation_coroutines = [ validator_agents[vt].do_async(validation_tasks[i], return_output=True) for i, vt in enumerate(validation_types) ] validation_outputs = await asyncio.gather(*validation_coroutines) # 用量核算 from upsonic.usage import RunUsage if task._reliability_sub_agent_usage is None: task._reliability_sub_agent_usage = RunUsage() for v_output in validation_outputs: if hasattr(v_output, 'usage') and v_output.usage: task._reliability_sub_agent_usage.incr(v_output.usage) for i, vt in enumerate(validation_types): setattr(validation_result, vt, validation_tasks[i].response)

使用asyncio.gather而非顺序 await——四个 verifier 相互独立，因此墙钟成本大约是一次LLM 调用，而非四次。

8.6 阶段 F —— 汇总

validation_result.calculate_suspicion()

这会填充any_suspicion、扁平的suspicious_points列表、人类可读的overall_feedback，以及平均的overall_confidence。

8.7 阶段 G —— Editor 重写

editor_agent = AgentConfiguration(model=model, name="Information Editor Agent") formatted_prompt = editor_task_prompt.format( validation_feedback=validation_result.overall_feedback ) formatted_prompt += ( f"\n\nORIGINAL AI RESPONSE TO CLEAN:\n{old_task_output}\n\n" f"Return the cleaned version of this response in the same format " f"(not as code or XML):" ) cleaned_context = [] if copy_task.context: for item in copy_task.context: if isinstance(item, str): cleaned_context.append(strip_context_tags(item)) else: cleaned_context.append(item) editor_task = Task( formatted_prompt, images=task.images, context=[copy_task.response_format, validation_result] + cleaned_context, response_format=task.response_format, # 与父任务相同的 schema tools=task.tools, task_usage_id_=task.task_usage_id, not_main_task=True, ) editor_output = await editor_agent.do_async(editor_task, return_output=True) # Editor 用量核算 if hasattr(editor_output, 'usage') and editor_output.usage: if not hasattr(task, '_reliability_sub_agent_usage'): task._reliability_sub_agent_usage = RunUsage() task._reliability_sub_agent_usage.incr(editor_output.usage) task._response = editor_task.response return task

三个承重细节：

1. response_format被保留——清洗后的输出仍须匹配调用方的 Pydantic schema，因此 editor 收到相同 schema，且原始 Pydantic 模型类作为上下文元素一并给出。
2. ValidationResult是 editor 上下文的一部分——editor 看到的不仅是要清洗什么，还有为什么（逐字段的suspicious_points、校验反馈）。
3. not_main_task=True——verifier 与 editor 子任务都被打标，使框架其他部分（遥测、回调）能区分可靠性内部调用与用户主任务。

8.8 阶段 H —— 管线返回清洗后的任务

处理器返回task。管线步骤随后：

context.output = processed_task.response # 设置新输出 task._reliability_sub_agent_usage = None # 清理暂存标记 modifications_made = str(original_output) != str(context.output) # 发出 reliability 事件 / 调试日志（debug_level >= 2 时）

…然后 agent 运行继续到其后的步骤（通常是结果格式化与收尾）。

8.9 时序图

ReliabilityStep ReliabilityManager ReliabilityProcessor Verifier×4 Editor │ │ │ │ │ │── aprepare() ────────►│ │ │ │ │ │ │ │ │ │── process_task ──────►│── process_task ────────►│ │ │ │ │ │ │ │ │ │── strip_context_tags / regex ── │ │ │ │ │ │ │ │── Task(URL) ────►Agent.do_async ──┤ │ │ │── Task(Num) ────►Agent.do_async ──┤ │ │ │── Task(Info) ────►Agent.do_async ──┤ asyncio.gather │ │ │── Task(Code) ────►Agent.do_async ──┤ │ │ │ │ │ │ │◄── 4×ValidationPoint ──────────────┤ │ │ │ │ │ │ │── calculate_suspicion() │ │ │ │ │ │ │ │── if any_suspicion: ───────────────►│ │ │ │ │── do_async │ │ │◄── cleaned response ───────────────│ │ │ │ │ │ │◄── task (mutated) ──────│ │ │◄── processed_task ────│ │ │ │── afinalize() ───────►│ │ │ │ │ │ │ │── fold usage / emit reliability event ─────────────────────────────────────────────► │

8.10 成本特征

每一次这样的调用都会贡献给task._reliability_sub_agent_usage，管线步骤随后将其并入父运行用量，使用户看到单一准确的成本数字。

8.11 失败模式与不变量

附录 A —— 事实依据文件路径

附录 B —— 新增一个 verifier

若要新增一个，比如citation（引用）verifier：

1. 在现有四条提示词旁新增一个提示词常量（如citation_validation_prompt）。
2. 在ValidationResult中新增citation_validation: ValidationPoint，并更新calculate_suspicion把它的标志、要点与置信度纳入聚合。
3. 在process_task内部的迭代列表中追加("citation_validation", citation_validation_prompt)。
4. （可选）新增find_citations_in_text正则预过滤，以及对应的if validation_type == "citation_validation": ...短路分支。
5. 无需管线改动——ReliabilityStep是数据驱动的，会原样处理新增类型。

这种最小触碰设计是有意为之：每个新 verifier 的成本就是一条提示词 + 一个模型字段 + 一个元组条目。

【免费下载链接】gpt-computer-assistantBuild autonomous AI agents in Python.项目地址: https://gitcode.com/GitHub_Trending/gp/gpt-computer-assistant