构建AI代理受托程序框架：应对不确定性，确保可信行为

在实际工程中，将人工智能（AI）能力集成到现有系统，特别是那些涉及自动化、决策或与物理世界交互的机器人系统时，开发者面临的核心挑战远不止于调用一个API。模型输出的不确定性、对上下文理解的偏差（即“AI幻觉”）、以及如何让AI的行为与业务规则和安全边界对齐，构成了“可信AI”落地的关键难题。这不仅仅是算法问题，更是一个系统工程问题，需要在架构层面引入一套可靠的治理与约束机制。

本文将从工程实践角度，探讨如何为集成AI的机器人或智能代理（AI Agent）构建一个“受托程序”（Fiduciary Program）框架。这个框架的目标是确保AI的行为始终处于预设的可信边界内，具备可预测、可审查、可干预的特性。我们将以构建一个具备基础任务执行与安全校验能力的AI代理为例，贯穿环境搭建、核心模块实现、运行验证到生产级考量的全过程。通过本文，你将掌握为AI应用注入确定性的核心设计模式与实现方案。

1. 理解核心概念：AI不确定性、受托程序与约束框架

在深入代码之前，必须厘清几个关键概念，它们构成了后续所有设计的基础。

1.1 AI的不确定性与“幻觉”

AI模型，尤其是大语言模型（LLM），本质上是基于概率生成内容的系统。其输出具有以下不确定性：

• 非确定性：相同输入可能产生不同输出。
• 事实性偏差：可能生成看似合理但不符合事实或训练数据之外的信息，即“幻觉”。
• 指令遵循偏差：可能无法严格遵循复杂的、多步骤的指令或约束。

在机器人或自动化流程中，这种不确定性是危险的。例如，一个负责库存管理的AI代理，如果“幻觉”出一个不存在的商品并执行出库指令，将导致数据混乱。

1.2 何为“受托程序”？

“受托”一词源于法律，指受托人负有为了受益人最大利益而行为的义务。在AI工程中，“受托程序”指的是包裹在核心AI模型之外的一层软件框架。它的职责是：

1. 约束：解析并强制执行业务规则、安全策略和伦理边界。
2. 验证：对AI的决策或生成内容进行事实性、逻辑性和安全性校验。
3. 裁决：在AI输出不符合要求时，进行修正、重试或触发人工干预。
4. 记录：完整审计AI决策链路，确保过程可追溯。

它不是AI本身，而是AI的“安全带”和“导航仪”。

1.3 约束框架的设计模式

一个典型的约束框架遵循“链式处理”或“管道与过滤器”模式。AI的原始输出被视为需要被加工的“原材料”，依次通过多个“过滤器”（即约束检查器），只有通过所有检查的最终结果才会被交付给执行器。

[用户/系统指令] -> [AI模型生成] -> [约束过滤器1: 格式校验] -> [约束过滤器2: 事实核查] -> [约束过滤器3: 安全策略] -> [最终裁决] -> [执行]

如果任何过滤器失败，流程将中断，并进入错误处理或重试分支。

2. 环境准备与项目结构

我们将使用Python作为实现语言，因为它拥有丰富的AI和工程类库生态。本项目不依赖于某个特定的LLM服务，你可以接入OpenAI API、本地部署的Ollama模型或任何其他兼容接口。

2.1 基础环境与依赖

首先确保你的Python版本在3.8以上。建议使用虚拟环境。

# 创建并激活虚拟环境 python -m venv venv source venv/bin/activate # Linux/macOS # venv\Scripts\activate # Windows # 安装核心依赖 pip install openai>=1.0.0 # 以OpenAI SDK为例，也可替换为其他 pip install pydantic>=2.0.0 # 用于数据验证和约束定义 pip install tenacity>=8.0.0 # 用于重试逻辑 pip install loguru>=0.7.0 # 用于结构化日志

2.2 项目目录结构

一个清晰的结构有助于管理复杂的约束逻辑。建议按以下方式组织：

my_ai_fiduciary/ ├── config/ │ ├── __init__.py │ └── settings.py # 配置文件，存放API密钥、模型参数、约束阈值等 ├── core/ │ ├── __init__.py │ ├── ai_client.py # 封装AI模型调用 │ ├── constraints/ # 约束器模块 │ │ ├── __init__.py │ │ ├── base.py # 约束器基类 │ │ ├── content_filter.py # 内容安全过滤 │ │ ├── fact_checker.py # 事实核查（示例） │ │ └── schema_validator.py # 输出格式验证 │ ├── fiduciary_engine.py # 受托引擎，编排约束执行流程 │ └── models.py # Pydantic数据模型，定义输入输出结构 ├── tasks/ │ ├── __init__.py │ └── inventory_task.py # 具体业务任务示例：库存管理 ├── logs/ # 日志目录（需手动创建） ├── tests/ # 单元测试 └── main.py # 主程序入口

3. 构建受托引擎与约束系统

这是框架的核心。我们将从定义数据模型开始，逐步实现约束器和引擎。

3.1 定义输入输出与任务上下文

使用Pydantic来严格定义数据格式，这本身就是第一道约束。

# core/models.py from pydantic import BaseModel, Field from typing import Any, Dict, Optional, List from enum import Enum class TaskStatus(str, Enum): PENDING = "pending" AI_PROCESSING = "ai_processing" CONSTRAINT_CHECKING = "constraint_checking" APPROVED = "approved" REJECTED = "rejected" ERROR = "error" class AgentTask(BaseModel): """代理任务基类""" task_id: str = Field(..., description="任务唯一标识") user_instruction: str = Field(..., description="用户原始指令") context: Dict[str, Any] = Field(default_factory=dict, description="任务上下文信息") status: TaskStatus = Field(default=TaskStatus.PENDING, description="任务状态") raw_ai_output: Optional[str] = Field(default=None, description="AI原始输出") validated_output: Optional[Dict[str, Any]] = Field(default=None, description="经验证后的结构化输出") error_message: Optional[str] = Field(default=None, description="错误信息") audit_trail: List[str] = Field(default_factory=list, description="审计追踪日志") class InventoryAction(str, Enum): CHECK = "check_stock" ADD = "add_item" REMOVE = "remove_item" class InventoryTask(AgentTask): """库存管理任务，继承自基类""" expected_action: Optional[InventoryAction] = Field(default=None, description="期望的动作类型") item_name: Optional[str] = Field(default=None, description="物品名称") quantity: Optional[int] = Field(default=None, description="数量，用于添加或移除")

3.2 实现约束器基类与具体约束

所有约束器都应遵循统一的接口。

# core/constraints/base.py from abc import ABC, abstractmethod from loguru import logger from core.models import AgentTask class BaseConstraint(ABC): """约束器抽象基类""" name: str = "base_constraint" @abstractmethod async def check(self, task: AgentTask) -> bool: """ 执行约束检查。 返回True表示通过，False表示拒绝。 应修改task.audit_trail以记录检查结果。 """ pass def _log_check(self, task: AgentTask, passed: bool, message: str): """统一的检查日志记录""" log_msg = f"[约束器:{self.name}] {message} - 通过: {passed}" task.audit_trail.append(log_msg) if passed: logger.info(f"任务 {task.task_id}: {log_msg}") else: logger.warning(f"任务 {task.task_id}: {log_msg}") # core/constraints/schema_validator.py import json import re from core.constraints.base import BaseConstraint from core.models import AgentTask, InventoryTask, InventoryAction class SchemaValidator(BaseConstraint): """输出格式与结构验证器""" name = "schema_validator" async def check(self, task: AgentTask) -> bool: if not task.raw_ai_output: self._log_check(task, False, "AI原始输出为空") return False # 示例：尝试将AI输出解析为JSON，并验证基本结构 try: parsed = json.loads(task.raw_ai_output) except json.JSONDecodeError: # 如果AI没有返回标准JSON，尝试提取 match = re.search(r'\{.*\}', task.raw_ai_output, re.DOTALL) if match: try: parsed = json.loads(match.group()) except json.JSONDecodeError: self._log_check(task, False, "无法从输出中解析出有效JSON") return False else: self._log_check(task, False, "输出不是有效的JSON格式") return False # 基础字段验证 required_fields = ["action", "reasoning"] for field in required_fields: if field not in parsed: self._log_check(task, False, f"解析结果缺少必要字段: {field}") return False # 针对库存任务的额外验证 if isinstance(task, InventoryTask): try: action = InventoryAction(parsed["action"]) task.expected_action = action # 验证数量字段，如果动作为ADD或REMOVE，则必须存在且为正整数 if action in [InventoryAction.ADD, InventoryAction.REMOVE]: if "quantity" not in parsed or not isinstance(parsed["quantity"], int) or parsed["quantity"] <= 0: self._log_check(task, False, f"动作{action.value}需要有效的正数quantity字段") return False task.quantity = parsed["quantity"] if "item_name" in parsed: task.item_name = parsed["item_name"] except ValueError: self._log_check(task, False, f"解析的动作值'{parsed.get('action')}'不是有效的库存操作") return False task.validated_output = parsed self._log_check(task, True, f"输出格式验证通过，解析结果: {parsed}") return True # core/constraints/content_filter.py class ContentSafetyFilter(BaseConstraint): """内容安全过滤（示例：关键词过滤）""" name = "content_safety_filter" def __init__(self, banned_keywords: list = None): self.banned_keywords = banned_keywords or ["删除所有", "格式化", "root", "sudo"] async def check(self, task: AgentTask) -> bool: if not task.raw_ai_output: return True # 空输出不进行安全过滤 text_to_check = task.raw_ai_output.lower() for keyword in self.banned_keywords: if keyword in text_to_check: self._log_check(task, False, f"输出包含禁止关键词: '{keyword}'") return False self._log_check(task, True, "内容安全检查通过") return True

3.3 实现受托引擎

引擎负责按顺序调用AI和一系列约束器，并管理任务状态。

# core/fiduciary_engine.py from typing import List, Type from loguru import logger from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type from core.models import AgentTask, TaskStatus from core.constraints.base import BaseConstraint from core.ai_client import AIClient # 假设已实现 class FiduciaryEngine: """受托引擎：编排AI调用与约束检查流程""" def __init__(self, ai_client: AIClient, constraints: List[BaseConstraint]): self.ai_client = ai_client self.constraints = constraints async def execute_task(self, task: AgentTask) -> AgentTask: """执行任务的主流程""" logger.info(f"开始处理任务: {task.task_id}") task.status = TaskStatus.AI_PROCESSING # 步骤1: 调用AI获取原始输出 try: task.raw_ai_output = await self._call_ai_with_retry(task) except Exception as e: task.status = TaskStatus.ERROR task.error_message = f"AI调用失败: {str(e)}" logger.error(f"任务 {task.task_id} AI调用失败: {e}") return task # 步骤2: 依次执行约束检查 task.status = TaskStatus.CONSTRAINT_CHECKING for constraint in self.constraints: try: passed = await constraint.check(task) if not passed: task.status = TaskStatus.REJECTED task.error_message = f"约束检查失败于: {constraint.name}" logger.warning(f"任务 {task.task_id} 被约束器 {constraint.name} 拒绝") return task except Exception as e: task.status = TaskStatus.ERROR task.error_message = f"约束检查执行异常 {constraint.name}: {str(e)}" logger.error(f"任务 {task.task_id} 约束器 {constraint.name} 执行异常: {e}") return task # 步骤3: 所有约束通过 task.status = TaskStatus.APPROVED logger.success(f"任务 {task.task_id} 处理成功，最终输出: {task.validated_output}") return task @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10), retry=retry_if_exception_type((TimeoutError, ConnectionError)), reraise=True ) async def _call_ai_with_retry(self, task: AgentTask) -> str: """带重试机制的AI调用""" # 这里构建更精确的提示词，引导AI输出结构化JSON system_prompt = """你是一个库存管理助手。请根据用户指令，判断其意图并生成一个JSON对象。 JSON必须包含以下字段： 1. `action`: 字符串，必须是 `check_stock`, `add_item`, `remove_item` 中的一个。 2. `item_name`: 字符串，指令中提到的物品名称。如果未提及，则为null。 3. `quantity`: 整数，当action为add_item或remove_item时，表示操作数量。否则为null。 4. `reasoning`: 字符串，简要说明你为什么做出这个判断。 请只输出JSON，不要有其他任何解释。""" user_prompt = task.user_instruction return await self.ai_client.generate(system_prompt, user_prompt)

3.4 实现AI客户端封装

# core/ai_client.py from openai import AsyncOpenAI import os from typing import Optional from loguru import logger class AIClient: """AI客户端封装，便于切换不同模型后端""" def __init__(self, api_key: Optional[str] = None, base_url: Optional[str] = None, model: str = "gpt-3.5-turbo"): api_key = api_key or os.getenv("OPENAI_API_KEY") base_url = base_url or os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1") self.client = AsyncOpenAI(api_key=api_key, base_url=base_url) self.model = model async def generate(self, system_prompt: str, user_prompt: str) -> str: try: response = await self.client.chat.completions.create( model=self.model, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt} ], temperature=0.2, # 降低随机性，使输出更确定 max_tokens=500 ) content = response.choices[0].message.content.strip() logger.debug(f"AI原始返回: {content}") return content except Exception as e: logger.error(f"AI调用异常: {e}") raise

4. 运行验证与结果分析

现在，我们将上述模块组合起来，创建一个完整的库存管理任务流程进行验证。

4.1 编写主程序与任务示例

# main.py import asyncio import uuid from loguru import logger from core.ai_client import AIClient from core.fiduciary_engine import FiduciaryEngine from core.constraints.schema_validator import SchemaValidator from core.constraints.content_filter import ContentSafetyFilter from tasks.inventory_task import InventoryTask async def main(): # 1. 初始化组件 ai_client = AIClient(model="gpt-3.5-turbo") # 确保已设置环境变量 OPENAI_API_KEY constraints = [ ContentSafetyFilter(banned_keywords=["删除所有数据", "drop table"]), SchemaValidator(), # 未来可以添加 FactChecker（需要接入知识库或搜索API） ] engine = FiduciaryEngine(ai_client, constraints) # 2. 创建测试任务 test_tasks = [ InventoryTask( task_id=str(uuid.uuid4()), user_instruction="帮我查一下仓库里还有多少台笔记本电脑？", context={"warehouse": "北京仓"} ), InventoryTask( task_id=str(uuid.uuid4()), user_instruction="我们需要增加50个鼠标的库存。", context={"warehouse": "上海仓"} ), InventoryTask( task_id=str(uuid.uuid4()), user_instruction="从库存里移除3个损坏的键盘。", context={"warehouse": "广州仓"} ), # 一个可能触发约束的指令 InventoryTask( task_id=str(uuid.uuid4()), user_instruction="我觉得应该删除所有库存记录然后重新开始。", context={"warehouse": "测试仓"} ), ] # 3. 执行所有任务 for task in test_tasks: logger.info(f"\n{'='*50}") logger.info(f"处理指令: '{task.user_instruction}'") result_task = await engine.execute_task(task) # 4. 打印结果 logger.info(f"任务状态: {result_task.status.value}") logger.info(f"验证后输出: {result_task.validated_output}") if result_task.error_message: logger.error(f"错误信息: {result_task.error_message}") logger.info("审计追踪:") for log in result_task.audit_trail: print(f" - {log}") if __name__ == "__main__": # 配置日志 logger.add("logs/ai_fiduciary_{time:YYYY-MM-DD}.log", rotation="1 day", level="INFO") asyncio.run(main())

4.2 预期输出与分析

运行python main.py，你应当看到类似以下的输出（具体内容因AI模型输出而异）：

2024-05-XX ... INFO 开始处理任务: xxxxx-xxxx-... ================================================== 处理指令: '帮我查一下仓库里还有多少台笔记本电脑？' 2024-05-XX ... INFO 任务 xxxxx AI原始返回: {"action": "check_stock", "item_name": "笔记本电脑", "quantity": null, "reasoning": "用户询问仓库中笔记本电脑的数量，这是一个查询操作。"} 2024-05-XX ... INFO 任务 xxxxx: [约束器:content_safety_filter] 内容安全检查通过 - 通过: True 2024-05-XX ... INFO 任务 xxxxx: [约束器:schema_validator] 输出格式验证通过，解析结果: {...} - 通过: True 2024-05-XX ... SUCCESS 任务 xxxxx 处理成功，最终输出: {'action': 'check_stock', ...} 任务状态: approved 验证后输出: {'action': 'check_stock', 'item_name': '笔记本电脑', 'quantity': null, 'reasoning': '...'} 审计追踪: - [约束器:content_safety_filter] 内容安全检查通过 - 通过: True - [约束器:schema_validator] 输出格式验证通过，解析结果: {...} - 通过: True ================================================== 处理指令: '我觉得应该删除所有库存记录然后重新开始。' 2024-05-XX ... INFO 任务 xxxxx AI原始返回: {"action": "remove_item", "item_name": "所有库存记录", "quantity": 999, "reasoning": "用户要求删除所有记录，我将其理解为移除操作。"} 2024-05-XX ... WARNING 任务 xxxxx: [约束器:content_safety_filter] 输出包含禁止关键词: '删除所有' - 通过: False 2024-05-XX ... WARNING 任务 xxxxx 被约束器 content_safety_filter 拒绝 任务状态: rejected 验证后输出: None 错误信息: 约束检查失败于: content_safety_filter 审计追踪: - [约束器:content_safety_filter] 输出包含禁止关键词: '删除所有' - 通过: False

结果分析：

1. 正常任务：成功通过所有约束，validated_output被正确解析和填充，状态为approved。后续系统可以安全地根据action和item_name执行数据库查询或更新。
2. 危险任务：AI模型可能依然会遵循危险指令生成输出（如“删除所有”），但ContentSafetyFilter约束器成功拦截，任务状态变为rejected，流程被终止，从而防止了危险操作的发生。audit_trail清晰记录了拦截原因。

5. 常见问题排查与约束设计进阶

在实际部署中，你会遇到各种边界情况。以下是典型问题及排查路径。

5.1 AI输出不符合JSON格式

现象：SchemaValidator约束器频繁失败，日志显示“无法从输出中解析出有效JSON”。原因与排查：

1. 提示词不明确：检查_call_ai_with_retry方法中的system_prompt，是否明确要求“只输出JSON”。可以加强提示，如“你的响应必须是且仅是一个合法的JSON对象，不要有任何其他文本。”
2. 模型能力或温度值：如果使用较小或未经调优的模型，可能无法稳定输出JSON。尝试更换模型（如gpt-4）或进一步降低temperature（如0.1）。
3. 输出解析策略不足：当前的SchemaValidator使用了正则表达式提取，可能不够健壮。可以考虑使用更复杂的解析库或引入一个“输出修复”约束器，在JSON解析失败后，尝试调用另一个AI专门修复格式。

5.2 约束检查导致性能瓶颈

现象：任务处理时间过长，尤其是引入需要调用外部API的事实核查器时。优化方案：

1. 异步并行检查：如果约束器之间没有依赖关系，可以在FiduciaryEngine中使用asyncio.gather并行执行。

   async def execute_task(self, task: AgentTask) -> AgentTask: # ... AI调用之后 ... constraint_tasks = [constraint.check(task) for constraint in self.constraints] results = await asyncio.gather(*constraint_tasks, return_exceptions=True) for constraint, result in zip(self.constraints, results): if isinstance(result, Exception): # ... 异常处理 ... return task if not result: # ... 拒绝处理 ... return task # ... 通过处理 ...

2. 短路与优先级：将最可能失败或开销最小的约束器放在前面。例如，ContentSafetyFilter可以在SchemaValidator之前运行，如果内容不安全，则无需进行格式验证。
3. 缓存与限流：对事实核查等昂贵操作的结果进行缓存（如基于item_name），并实施限流策略。

5.3 如何设计更复杂的业务规则约束

SchemaValidator只做了基础格式验证。真正的业务规则可能更复杂，例如：“单次出库数量不能超过当前库存量”。这需要引入一个能访问领域数据（如数据库）的约束器。

# core/constraints/business_rules.py from core.constraints.base import BaseConstraint from core.models import InventoryTask, TaskStatus # 假设有一个库存服务 from services.inventory_service import InventoryService class InventoryBusinessRuleChecker(BaseConstraint): name = "inventory_business_rule" def __init__(self, inventory_service: InventoryService): self.inventory_service = inventory_service async def check(self, task: InventoryTask) -> bool: if not task.validated_output: return False if task.expected_action == "remove_item": current_stock = await self.inventory_service.get_stock(task.item_name) if current_stock is None: self._log_check(task, False, f"物品 '{task.item_name}' 不存在于库存中") return False if task.quantity > current_stock: self._log_check(task, False, f"出库数量({task.quantity})超过当前库存({current_stock})") return False self._log_check(task, True, "业务规则检查通过") return True

6. 生产环境最佳实践与扩展方向

将“受托程序”框架投入生产，需要超越基础功能，关注可靠性、可观测性和可维护性。

6.1 生产级考量清单

6.2 扩展方向：构建更智能的约束

基础的关键词和格式过滤只是开始。可以考虑集成以下更高级的约束能力：

1. 基于向量数据库的事实核查：将内部知识库文档向量化存储。当AI输出涉及具体事实（如产品规格、流程步骤）时，从向量库检索相关片段进行比对，计算相关性分数，低于阈值则拒绝。
2. 输出毒性/偏见检测：集成专门的内容安全API（如Google Perspective API）或本地模型，检测生成内容是否包含仇恨、歧视或极端言论。
3. 逻辑一致性检查：对于多轮对话或复杂任务，检查AI本次输出是否与历史上下文或已承诺的行动计划相矛盾。
4. 成本与资源约束：为任务设置“信用点”系统，如果AI建议的操作（如调用昂贵API、进行大规模计算）超出预算，则拒绝该方案并要求其提出更经济的替代方案。

6.3 框架的通用化改造

当前示例围绕“库存任务”设计。要将其通用化，可以：

• 定义更抽象的Task基类和Constraint接口。
• 引入“插件”机制，通过配置文件动态加载不同业务领域的约束器组合。
• 开发一个管理界面，用于实时查看任务流水线、调整约束器参数、手动干预被拒绝的任务。

通过本文的实践，你构建的不仅仅是一个AI调用封装，而是一个具备初步“受托”能力的智能代理管控框架。它的价值在于将AI的“黑盒”不确定性，纳入了软件工程可管理、可控制的范畴。下一步，你可以根据具体业务场景，深化约束器的逻辑，并将其与你的业务系统深度集成，让AI真正成为可靠的生产力组件。