在AI应用开发中我们经常面临一个核心矛盾如何让灵活的大语言模型LLM与需要严格控制的业务规则和安全要求协同工作。传统Agent架构通常将决策权完全交给LLM但在金融、医疗、法律等敏感领域这种黑盒决策方式存在明显风险。本文将介绍一种创新的Governed Agent架构通过规则引擎和代码逻辑来约束LLM行为实现可控的AI决策系统。1. Governed Agent架构核心概念1.1 什么是Governed AgentGoverned Agent是一种新型的AI代理架构其核心思想是将LLM的创意能力与确定性规则相结合。与传统Agent不同Governed Agent中LLM主要负责理解用户请求和生成候选方案而最终的决策权交给预先定义的业务规则和代码逻辑。这种架构特别适合需要遵守合规要求、确保结果一致性的业务场景。比如在金融风控中LLM可以分析用户请求的语义但最终的审批决策必须基于明确的风控规则。1.2 与传统Agent的对比传统Agent架构通常采用LLM-as-a-Decision-Maker模式LLM直接决定每一步的行动。这种方式虽然灵活但存在几个关键问题不可预测性相同的输入可能产生不同的输出合规风险难以通过审计和合规检查一致性挑战在不同环境中表现可能不一致Governed Agent采用LLM-as-an-Analyst模式将LLM定位为分析工具而非决策者通过规则引擎确保决策的确定性和可追溯性。1.3 核心组件架构Governed Agent包含三个核心组件LLM理解层负责解析用户意图和上下文规则引擎层基于业务规则进行逻辑判断代码执行层处理具体的业务逻辑和数据操作这种分层架构确保了每个组件各司其职既利用了LLM的理解能力又保持了系统的可控性。2. 环境准备与开发工具2.1 基础环境要求在开始构建Governed Agent之前需要准备以下开发环境Python 3.8 运行环境访问LLM API的权限OpenAI GPT系列或本地部署的开源模型规则引擎依赖如Drools、Easy Rules或自定义规则系统版本控制工具Git2.2 核心依赖库创建requirements.txt文件包含以下关键依赖# requirements.txt openai1.0.0 pydantic2.0.0 fastapi0.100.0 uvicorn0.23.0 python-dotenv1.0.0 jsonschema4.0.02.3 项目结构规划建议采用以下目录结构组织Governed Agent项目governed-agent/ ├── src/ │ ├── agents/ # Agent核心逻辑 │ ├── rules/ # 业务规则定义 │ ├── schemas/ # 数据模型定义 │ ├── services/ # 业务服务层 │ └── utils/ # 工具函数 ├── tests/ # 测试用例 ├── config/ # 配置文件 └── docs/ # 项目文档3. Governed Agent核心实现3.1 LLM请求解析模块首先实现LLM请求解析层负责理解用户意图并提取关键信息# src/agents/llm_parser.py from typing import Dict, Any import openai from pydantic import BaseModel class UserRequest(BaseModel): 用户请求数据模型 text: str context: Dict[str, Any] {} class ParsedIntent(BaseModel): 解析后的意图模型 intent: str confidence: float entities: Dict[str, Any] requires_approval: bool False class LLMParser: def __init__(self, api_key: str, model: str gpt-3.5-turbo): self.client openai.OpenAI(api_keyapi_key) self.model model async def parse_request(self, request: UserRequest) - ParsedIntent: 使用LLM解析用户请求意图 system_prompt 你是一个请求分析助手。请分析用户请求并提取以下信息 1. 主要意图如查询、审批、计算等 2. 关键实体信息如金额、日期、名称等 3. 是否需要特殊审批 请以JSON格式返回分析结果。 response self.client.chat.completions.create( modelself.model, messages[ {role: system, content: system_prompt}, {role: user, content: request.text} ], response_format{type: json_object} ) # 解析LLM返回的JSON结果 result json.loads(response.choices[0].message.content) return ParsedIntent(**result)3.2 规则引擎实现规则引擎是Governed Agent的核心负责执行预定义的业务规则# src/rules/rule_engine.py from typing import List, Dict, Any from abc import ABC, abstractmethod class BusinessRule(ABC): 业务规则基类 abstractmethod def evaluate(self, context: Dict[str, Any]) - Dict[str, Any]: 评估规则并返回结果 pass abstractmethod def get_rule_id(self) - str: 获取规则标识 pass class AmountLimitRule(BusinessRule): 金额限制规则 def __init__(self, max_amount: float 10000.0): self.max_amount max_amount self.rule_id AMOUNT_LIMIT_001 def evaluate(self, context: Dict[str, Any]) - Dict[str, Any]: amount context.get(amount, 0) result { rule_id: self.rule_id, passed: amount self.max_amount, message: f金额限制检查: {amount} {self.max_amount}, max_amount: self.max_amount } return result def get_rule_id(self) - str: return self.rule_id class RuleEngine: 规则引擎 def __init__(self): self.rules: List[BusinessRule] [] def add_rule(self, rule: BusinessRule): 添加业务规则 self.rules.append(rule) def execute_rules(self, context: Dict[str, Any]) - Dict[str, Any]: 执行所有规则并返回综合结果 results [] all_passed True for rule in self.rules: rule_result rule.evaluate(context) results.append(rule_result) if not rule_result[passed]: all_passed False return { all_passed: all_passed, rule_results: results, final_decision: APPROVED if all_passed else REJECTED }3.3 Agent协调器实现Agent协调器负责整合LLM解析结果和规则引擎决策# src/agents/agent_orchestrator.py from typing import Dict, Any from .llm_parser import LLMParser, UserRequest, ParsedIntent from ..rules.rule_engine import RuleEngine class GovernedAgent: 受控Agent主类 def __init__(self, llm_parser: LLMParser, rule_engine: RuleEngine): self.llm_parser llm_parser self.rule_engine rule_engine async def process_request(self, user_input: str, context: Dict[str, Any] None) - Dict[str, Any]: 处理用户请求的完整流程 if context is None: context {} # 步骤1: LLM解析用户意图 request UserRequest(textuser_input, contextcontext) parsed_intent await self.llm_parser.parse_request(request) # 步骤2: 准备规则评估上下文 rule_context { intent: parsed_intent.intent, entities: parsed_intent.entities, requires_approval: parsed_intent.requires_approval, **context # 合并外部上下文 } # 步骤3: 执行业务规则 rule_result self.rule_engine.execute_rules(rule_context) # 步骤4: 生成最终响应 return { user_input: user_input, parsed_intent: parsed_intent.dict(), rule_evaluation: rule_result, final_action: self._determine_final_action(parsed_intent, rule_result) } def _determine_final_action(self, parsed_intent: ParsedIntent, rule_result: Dict[str, Any]) - str: 根据解析结果和规则评估确定最终行动 if not rule_result[all_passed]: return REJECTED if parsed_intent.requires_approval: return NEEDS_MANUAL_APPROVAL return APPROVED4. 完整实战案例金融审批Agent4.1 场景需求分析假设我们需要构建一个金融交易审批Agent要求自动解析用户交易请求根据金额、用户等级、交易类型进行规则检查确保所有决策符合合规要求提供完整的审计日志4.2 业务规则定义首先定义金融交易相关的业务规则# src/rules/financial_rules.py from .rule_engine import BusinessRule from typing import Dict, Any class TransactionAmountRule(BusinessRule): 交易金额规则 def __init__(self): self.limits { SILVER: 5000.0, GOLD: 20000.0, PLATINUM: 50000.0 } def evaluate(self, context: Dict[str, Any]) - Dict[str, Any]: user_tier context.get(user_tier, SILVER) amount context.get(amount, 0) limit self.limits.get(user_tier, 5000.0) passed amount limit return { rule_id: TRANSACTION_AMOUNT, passed: passed, message: f用户等级{user_tier}的单笔交易限额为{limit}当前金额{amount}, user_tier: user_tier, amount_limit: limit } def get_rule_id(self) - str: return TRANSACTION_AMOUNT class TransactionTypeRule(BusinessRule): 交易类型规则 def evaluate(self, context: Dict[str, Any]) - Dict[str, Any]: transaction_type context.get(transaction_type, ) restricted_types [CRYPTO, HIGH_RISK_INVESTMENT] passed transaction_type not in restricted_types return { rule_id: TRANSACTION_TYPE, passed: passed, message: f交易类型{transaction_type}检查, restricted: not passed } def get_rule_id(self) - str: return TRANSACTION_TYPE class DailyLimitRule(BusinessRule): 日累计限额规则 def evaluate(self, context: Dict[str, Any]) - Dict[str, Any]: daily_total context.get(daily_total, 0) user_tier context.get(user_tier, SILVER) tier_limits { SILVER: 20000.0, GOLD: 50000.0, PLATINUM: 100000.0 } limit tier_limits.get(user_tier, 20000.0) passed daily_total limit return { rule_id: DAILY_LIMIT, passed: passed, message: f今日累计交易{daily_total}限额{limit}, daily_limit: limit } def get_rule_id(self) - str: return DAILY_LIMIT4.3 主程序实现创建完整的金融审批Agent应用# src/applications/financial_agent.py import os from dotenv import load_dotenv from agents.llm_parser import LLMParser from rules.rule_engine import RuleEngine from rules.financial_rules import TransactionAmountRule, TransactionTypeRule, DailyLimitRule from agents.agent_orchestrator import GovernedAgent class FinancialApprovalAgent: 金融审批Agent应用 def __init__(self): load_dotenv() # 初始化LLM解析器 llm_parser LLMParser(api_keyos.getenv(OPENAI_API_KEY)) # 初始化规则引擎并添加规则 self.rule_engine RuleEngine() self.rule_engine.add_rule(TransactionAmountRule()) self.rule_engine.add_rule(TransactionTypeRule()) self.rule_engine.add_rule(DailyLimitRule()) # 创建受控Agent self.agent GovernedAgent(llm_parser, self.rule_engine) async def process_transaction_request(self, user_request: str, user_context: dict) - dict: 处理交易请求 return await self.agent.process_request(user_request, user_context) # 使用示例 async def main(): agent FinancialApprovalAgent() # 模拟用户请求 user_request 我想转账5000元到张三的账户 user_context { user_tier: GOLD, daily_total: 15000.0, transaction_type: TRANSFER } result await agent.process_transaction_request(user_request, user_context) print(审批结果:, result) if __name__ __main__: import asyncio asyncio.run(main())4.4 运行结果分析运行上述代码后典型的输出结果如下{ user_input: 我想转账5000元到张三的账户, parsed_intent: { intent: 转账, confidence: 0.95, entities: { amount: 5000.0, recipient: 张三, transaction_type: TRANSFER }, requires_approval: false }, rule_evaluation: { all_passed: true, rule_results: [ { rule_id: TRANSACTION_AMOUNT, passed: true, message: 用户等级GOLD的单笔交易限额为20000.0当前金额5000.0, user_tier: GOLD, amount_limit: 20000.0 }, { rule_id: TRANSACTION_TYPE, passed: true, message: 交易类型TRANSFER检查, restricted: false }, { rule_id: DAILY_LIMIT, passed: true, message: 今日累计交易15000.0限额50000.0, daily_limit: 50000.0 } ], final_decision: APPROVED }, final_action: APPROVED }4.5 扩展功能审计日志为满足合规要求添加审计日志功能# src/services/audit_logger.py import json from datetime import datetime from typing import Dict, Any class AuditLogger: 审计日志服务 def __init__(self, log_file: str audit.log): self.log_file log_file def log_decision(self, request_id: str, user_input: str, context: Dict[str, Any], result: Dict[str, Any]): 记录决策日志 log_entry { timestamp: datetime.now().isoformat(), request_id: request_id, user_input: user_input, context: context, result: result, decision_reasoning: self._extract_reasoning(result) } with open(self.log_file, a, encodingutf-8) as f: f.write(json.dumps(log_entry, ensure_asciiFalse) \n) def _extract_reasoning(self, result: Dict[str, Any]) - str: 从结果中提取决策理由 rule_results result.get(rule_evaluation, {}).get(rule_results, []) reasons [] for rule in rule_results: if rule.get(passed, False): reasons.append(f通过规则: {rule.get(rule_id)}) else: reasons.append(f未通过规则: {rule.get(rule_id)} - {rule.get(message, )}) return ; .join(reasons)5. 常见问题与解决方案5.1 LLM解析准确性优化问题现象LLM对用户意图解析不准确导致规则评估错误解决方案优化系统提示词提供更明确的解析指导使用few-shot learning提供解析示例添加后处理验证逻辑# 优化后的提示词示例 improved_system_prompt 你是一个专业的金融请求分析助手。请严格按照以下要求分析用户请求 解析步骤 1. 识别主要意图转账、查询、投资、贷款等 2. 提取关键实体金额、账户、日期、交易类型 3. 判断风险等级高风险交易需要人工审批 示例 用户输入转账10000元到李四账户 输出{intent: 转账, amount: 10000, recipient: 李四, risk_level: 低} 请以JSON格式返回结果。 5.2 规则冲突处理问题现象多个规则产生冲突结果难以确定最终决策解决方案定义规则优先级实现规则冲突检测机制添加规则权重系统class PrioritizedRuleEngine(RuleEngine): 支持优先级的规则引擎 def execute_rules(self, context: Dict[str, Any]) - Dict[str, Any]: 按优先级执行规则 sorted_rules sorted(self.rules, keylambda x: getattr(x, priority, 0), reverseTrue) results [] for rule in sorted_rules: result rule.evaluate(context) results.append(result) # 高优先级规则否决后停止执行 if getattr(rule, is_veto_rule, False) and not result[passed]: break return self._aggregate_results(results)5.3 性能优化策略问题现象规则数量增多导致处理延迟解决方案实现规则缓存机制使用规则编译优化并行执行独立规则6. 生产环境最佳实践6.1 安全与合规考虑在生产环境中部署Governed Agent时需要特别注意以下安全事项API密钥管理使用环境变量或密钥管理服务避免硬编码输入验证对所有用户输入进行严格的验证和清理权限控制确保Agent只能访问必要的资源和数据审计追踪记录所有决策过程和修改历史6.2 监控与告警建立完整的监控体系# src/services/monitoring.py import time from prometheus_client import Counter, Histogram, Gauge # 定义监控指标 requests_total Counter(agent_requests_total, Total requests processed) request_duration Histogram(agent_request_duration_seconds, Request duration) rules_evaluated Gauge(agent_rules_evaluated, Number of rules evaluated) class MonitoringMiddleware: 监控中间件 async def process_with_monitoring(self, agent, request, context): start_time time.time() requests_total.inc() try: result await agent.process_request(request, context) rules_evaluated.set(len(result.get(rule_evaluation, {}).get(rule_results, []))) return result finally: duration time.time() - start_time request_duration.observe(duration)6.3 版本管理与回滚实现规则和模型的版本管理使用Git管理规则定义文件为每个规则版本打标签实现规则热更新机制保留历史版本以便回滚6.4 测试策略建立全面的测试体系# tests/test_financial_agent.py import pytest from src.applications.financial_agent import FinancialApprovalAgent class TestFinancialAgent: pytest.fixture def agent(self): return FinancialApprovalAgent() pytest.mark.asyncio async def test_approval_flow(self, agent): 测试正常审批流程 result await agent.process_transaction_request( 转账5000元, {user_tier: GOLD, daily_total: 10000} ) assert result[final_action] APPROVED pytest.mark.asyncio async def test_rejection_flow(self, agent): 测试拒绝流程 result await agent.process_transaction_request( 转账50000元, {user_tier: SILVER, daily_total: 10000} ) assert result[final_action] REJECTED7. 扩展应用场景7.1 客服自动化系统将Governed Agent应用于客服场景确保回复内容符合公司政策class CustomerServiceAgent: 客服Agent def __init__(self): self.rule_engine.add_rule(ResponseToneRule()) # 回复语气规则 self.rule_engine.add_rule(InformationDisclosureRule()) # 信息泄露规则 self.rule_engine.add_rule(EscalationRule()) # 升级规则7.2 内容审核系统用于社交媒体内容审核结合LLM理解和规则约束class ContentModerationAgent: 内容审核Agent def __init__(self): self.rule_engine.add_rule(HateSpeechRule()) # 仇恨言论规则 self.rule_engine.add_rule(SensitiveTopicRule()) # 敏感话题规则 self.rule_engine.add_rule(SpamDetectionRule()) # 垃圾信息规则7.3 智能合约审核在区块链领域应用Governed Agent进行智能合约安全审核class SmartContractAgent: 智能合约审核Agent def __init__(self): self.rule_engine.add_rule(SecurityVulnerabilityRule()) # 安全漏洞规则 self.rule_engine.add_rule(GasOptimizationRule()) # Gas优化规则 self.rule_engine.add_rule(ComplianceRule()) # 合规规则Governed Agent架构为AI应用提供了确定性与灵活性的最佳平衡特别适合需要严格控制的业务场景。通过将LLM的分析能力与规则引擎的确定性决策相结合我们既能享受AI的智能优势又能确保系统的可靠性和合规性。在实际项目中建议从简单的规则开始逐步完善规则体系同时建立完善的测试和监控机制。