Governed Agent架构：企业级AI Agent的可控智能决策实践

如果你正在构建企业级的 AI Agent 系统，是否遇到过这样的困境：大语言模型（LLM）生成的回答看似合理，但在关键的业务逻辑、数据权限或合规要求上却存在风险？传统的 Agent 框架往往将决策权完全交给 LLM，而 Governed Agent 提出了一种全新的思路——让 LLM 负责理解问题，但让规则和代码来做最终决定。

这不仅仅是另一个 Agent 框架的更新，而是对 AI 应用落地过程中"可控性"与"确定性"这一核心矛盾的架构级回应。在企业环境中，我们既需要 LLM 的语义理解能力，又必须保证业务流程的合规性、数据的安全性。Governed Agent 通过明确的职责分离：LLM 读请求，规则引擎和代码做决策，实现了"智能"与"控制"的平衡。

本文将深入解析 Governed Agent 的设计理念、核心架构，并通过完整的实战示例展示如何构建一个既灵活又可靠的企业级 AI Agent。无论你是正在探索 AI 落地的技术负责人，还是希望将 AI 能力集成到现有系统的开发者，这篇文章都将为你提供可落地的解决方案。

1. 这篇文章真正要解决的问题

在企业级 AI 应用开发中，我们面临着一个根本性的矛盾：LLM 的强大生成能力带来了灵活性，但业务系统需要的是确定性和可控性。传统 Agent 框架通常将整个决策流程交给 LLM，这导致了几个关键问题：

可靠性风险：LLM 可能生成看似合理但实际错误的业务逻辑，特别是在处理数值计算、权限判断等需要精确性的场景中。

安全合规挑战：在金融、医疗、法律等高度监管的行业，AI 的每一个决策都需要符合严格的合规要求，而纯 LLM 驱动的系统很难通过审计。

系统集成复杂度：企业现有系统通常有成熟的业务规则引擎、权限管理系统和工作流引擎，如何让 AI Agent 与这些系统无缝集成而非重新造轮子？

Governed Agent 的核心价值在于：它不试图用 LLM 替代现有的企业系统，而是让 LLM 成为这些系统的智能接口。LLM 负责将自然语言请求"翻译"成系统可理解的操作意图，而具体的业务逻辑仍然由经过验证的规则和代码来执行。

这种架构特别适合以下场景：

• 需要与现有 ERP、CRM 等企业系统集成的 AI 助手
• 涉及敏感数据查询或操作的内部知识库系统
• 需要严格合规审核的金融、医疗咨询场景
• 对响应准确性和可追溯性要求较高的生产环境

2. Governed Agent 的核心概念与架构设计

2.1 什么是 Governed Agent？

Governed Agent 是一种新型的 AI Agent 架构，其核心设计原则是"职责分离"：

• LLM 的角色：理解自然语言请求，将其解析为结构化的操作意图（Intent）
• 规则引擎的角色：根据预定义的业务规则，验证操作意图的合法性
• 代码执行器的角色：执行经过验证的具体业务逻辑，确保结果确定性

与传统 Agent 框架相比，Governed Agent 最大的不同在于：决策权不完全属于 LLM。LLM 更像是一个"前端解析器"，而真正的业务决策由后端的规则和代码完成。

2.2 核心架构组件

Governed Agent 通常包含以下关键组件：

┌─────────────────┐ ┌──────────────────┐ ┌─────────────────┐ │ 自然语言请求 │ -> │ 意图解析器(LLM) │ -> │ 规则验证引擎 │ └─────────────────┘ └──────────────────┘ └─────────────────┘ ↑ │ │ ↓ │ ┌─────────────────┐ │ │ 代码执行器 │ │ └─────────────────┘ │ │ └───────────────────────────────────────┘ │ ↓ ┌─────────────────┐ │ 结构化响应 │ └─────────────────┘

意图解析器（LLM）：接收用户的自然语言输入，输出结构化的操作意图。例如，将"帮我查询张三上个月的销售业绩"解析为：

{ "action": "query_sales_data", "params": { "employee_name": "张三", "time_range": "last_month" } }

规则验证引擎：根据业务规则验证操作意图的合法性。例如：

• 当前用户是否有权限查询张三的销售数据？
• 查询时间范围是否符合公司政策？
• 请求频率是否在合理范围内？

代码执行器：执行具体的业务逻辑。这里的代码是预先编写、经过测试的确定性代码，而不是 LLM 实时生成的。

2.3 与传统 Agent 的对比

3. 环境准备与基础依赖

3.1 系统要求

• Python 3.8+：Governed Agent 主要基于 Python 生态
• LLM API 访问：OpenAI GPT、Claude、或本地部署的 LLM
• 规则引擎：可根据需求选择 Drools、Easy Rules 或自定义规则引擎

3.2 核心依赖安装

# 创建虚拟环境 python -m venv governed-agent-env source governed-agent-env/bin/activate # Linux/Mac # governed-agent-env\Scripts\activate # Windows # 安装核心依赖 pip install openai anthropic pydantic fastapi pip install sqlalchemy pymysql # 数据库连接 pip install python-dotenv # 环境变量管理

3.3 项目结构规划

governed-agent/ ├── src/ │ ├── agents/ │ │ ├── __init__.py │ │ ├── base_agent.py # 基础 Agent 类 │ │ └── sales_agent.py # 销售数据查询 Agent │ ├── rules/ │ │ ├── __init__.py │ │ ├── base_rule.py # 基础规则类 │ │ └── sales_rules.py # 销售业务规则 │ ├── executors/ │ │ ├── __init__.py │ │ ├── base_executor.py # 基础执行器 │ │ └── sales_executor.py # 销售数据执行器 │ └── models/ │ ├── __init__.py │ └── intent_models.py # 意图模型定义 ├── config/ │ ├── __init__.py │ └── settings.py # 配置管理 ├── tests/ # 测试用例 └── requirements.txt # 依赖列表

4. 核心实现：构建一个销售数据查询 Agent

让我们通过一个具体的例子来展示 Governed Agent 的实现过程：构建一个销售数据查询 Agent，允许用户用自然语言查询销售数据，但确保所有查询都符合公司权限政策。

4.1 定义意图模型

首先，我们需要定义 LLM 应该输出的结构化意图：

# src/models/intent_models.py from pydantic import BaseModel, Field from typing import Optional, Literal class SalesQueryIntent(BaseModel): """销售查询意图模型""" action: Literal["query_sales_data"] = Field(description="操作类型") employee_name: Optional[str] = Field(description="员工姓名") department: Optional[str] = Field(description="部门名称") time_range: Literal["last_week", "last_month", "last_quarter"] = Field(description="时间范围") metric: Literal["sales_amount", "order_count", "customer_count"] = Field(description="指标类型") class IntentResponse(BaseModel): """意图解析响应""" success: bool intent: Optional[SalesQueryIntent] = None error_message: Optional[str] = None

4.2 实现意图解析器（LLM 层）

# src/agents/base_agent.py import openai from typing import Dict, Any from src.models.intent_models import SalesQueryIntent, IntentResponse class BaseAgent: def __init__(self, api_key: str, model: str = "gpt-3.5-turbo"): self.client = openai.OpenAI(api_key=api_key) self.model = model def parse_intent(self, user_input: str, intent_model: Any) -> IntentResponse: """使用 LLM 解析用户意图""" prompt = f""" 请将以下用户输入解析为结构化意图。用户输入：{user_input} 请严格按照以下 JSON Schema 格式输出： {intent_model.schema_json()} 示例： 用户输入："查询张三上个月的销售额" 输出：{{"action": "query_sales_data", "employee_name": "张三", "time_range": "last_month", "metric": "sales_amount"}} 注意：如果无法解析或信息不全，返回错误信息。 """ try: response = self.client.chat.completions.create( model=self.model, messages=[{"role": "user", "content": prompt}], temperature=0.1 # 低温度确保输出稳定性 ) # 解析 LLM 响应 result = json.loads(response.choices[0].message.content) intent = intent_model(**result) return IntentResponse(success=True, intent=intent) except Exception as e: return IntentResponse(success=False, error_message=f"意图解析失败: {str(e)}")

4.3 实现规则验证引擎

# src/rules/sales_rules.py from datetime import datetime, timedelta from src.models.intent_models import SalesQueryIntent class SalesRules: def __init__(self, current_user: str): self.current_user = current_user def validate_query_permission(self, intent: SalesQueryIntent) -> bool: """验证查询权限""" # 规则1：只能查询自己或下属的数据 if intent.employee_name and intent.employee_name != self.current_user: # 这里应该查询组织架构，简化示例 if not self._is_subordinate(intent.employee_name): return False # 规则2：经理可以查询部门数据，普通员工只能查询个人数据 if intent.department and not self._is_manager(self.current_user): return False return True def validate_time_range(self, intent: SalesQueryIntent) -> bool: """验证时间范围合法性""" # 规则：不能查询超过1年前的数据 max_allowed_days = 365 time_range_mapping = { "last_week": 7, "last_month": 30, "last_quarter": 90 } if intent.time_range not in time_range_mapping: return False if time_range_mapping[intent.time_range] > max_allowed_days: return False return True def _is_subordinate(self, employee_name: str) -> bool: """检查是否为下属（简化实现）""" # 实际项目中应查询组织架构数据库 subordinates = ["李四", "王五"] # 示例数据 return employee_name in subordinates def _is_manager(self, user_name: str) -> bool: """检查是否为经理（简化实现）""" managers = ["张三", "赵六"] # 示例数据 return user_name in managers

4.4 实现代码执行器

# src/executors/sales_executor.py import pandas as pd from sqlalchemy import create_engine, text from src.models.intent_models import SalesQueryIntent class SalesExecutor: def __init__(self, db_url: str): self.engine = create_engine(db_url) def execute_query(self, intent: SalesQueryIntent) -> dict: """执行销售数据查询""" # 构建 SQL 查询 sql = self._build_query(intent) try: with self.engine.connect() as conn: result = conn.execute(text(sql)) data = result.fetchall() return { "success": True, "data": self._format_result(data, intent), "query": sql # 用于审计 } except Exception as e: return { "success": False, "error": str(e), "query": sql } def _build_query(self, intent: SalesQueryIntent) -> str: """根据意图构建 SQL 查询""" time_conditions = { "last_week": "sales_date >= DATE_SUB(CURDATE(), INTERVAL 7 DAY)", "last_month": "sales_date >= DATE_SUB(CURDATE(), INTERVAL 1 MONTH)", "last_quarter": "sales_date >= DATE_SUB(CURDATE(), INTERVAL 3 MONTH)" } base_query = f""" SELECT {intent.metric}, sales_date, employee_name FROM sales_data WHERE {time_conditions[intent.time_range]} """ if intent.employee_name: base_query += f" AND employee_name = '{intent.employee_name}'" elif intent.department: base_query += f" AND department = '{intent.department}'" return base_query def _format_result(self, data: list, intent: SalesQueryIntent) -> dict: """格式化查询结果""" if not data: return {"message": "未找到相关数据"} df = pd.DataFrame(data, columns=[intent.metric, 'sales_date', 'employee_name']) return df.to_dict('records')

4.5 整合完整的 Governed Agent

# src/agents/sales_agent.py from src.agents.base_agent import BaseAgent from src.rules.sales_rules import SalesRules from src.executors.sales_executor import SalesExecutor from src.models.intent_models import SalesQueryIntent, IntentResponse class SalesAgent: def __init__(self, api_key: str, db_url: str, current_user: str): self.llm_agent = BaseAgent(api_key) self.rules_engine = SalesRules(current_user) self.executor = SalesExecutor(db_url) self.current_user = current_user def process_query(self, user_input: str) -> dict: """处理用户查询的完整流程""" # 步骤1：LLM 解析意图 intent_response = self.llm_agent.parse_intent(user_input, SalesQueryIntent) if not intent_response.success: return {"success": False, "error": intent_response.error_message} intent = intent_response.intent # 步骤2：规则验证 if not self.rules_engine.validate_query_permission(intent): return {"success": False, "error": "权限验证失败"} if not self.rules_engine.validate_time_range(intent): return {"success": False, "error": "时间范围不符合政策"} # 步骤3：执行查询 result = self.executor.execute_query(intent) # 添加审计日志 self._log_audit(intent, result) return result def _log_audit(self, intent: SalesQueryIntent, result: dict): """记录审计日志""" audit_log = { "timestamp": datetime.now().isoformat(), "user": self.current_user, "intent": intent.dict(), "result_status": "success" if result.get("success") else "failure", "query_executed": result.get("query") } # 实际项目中应写入审计数据库 print(f"AUDIT_LOG: {audit_log}")

5. 完整示例：部署与测试

5.1 配置环境变量

创建.env文件：

# .env OPENAI_API_KEY=your_openai_api_key_here DATABASE_URL=mysql+pymysql://user:password@localhost/sales_db CURRENT_USER=张三

5.2 主程序入口

# main.py import os from dotenv import load_dotenv from src.agents.sales_agent import SalesAgent def main(): # 加载环境变量 load_dotenv() # 初始化 Agent agent = SalesAgent( api_key=os.getenv("OPENAI_API_KEY"), db_url=os.getenv("DATABASE_URL"), current_user=os.getenv("CURRENT_USER") ) # 测试用例 test_queries = [ "查询我上个月的销售额", "查看销售部上个季度的订单数量", "帮我查一下李四上周的客户数量" ] for query in test_queries: print(f"\n=== 测试查询: {query} ===") result = agent.process_query(query) print(f"结果: {result}") if __name__ == "__main__": main()

5.3 运行与验证

# 运行示例 python main.py # 预期输出示例 === 测试查询: 查询我上个月的销售额 === 结果: {'success': True, 'data': [{'sales_amount': 150000, 'sales_date': '2024-01-15', 'employee_name': '张三'}], 'query': 'SELECT sales_amount, sales_date, employee_name FROM sales_data WHERE sales_date >= DATE_SUB(CURDATE(), INTERVAL 1 MONTH) AND employee_name = '张三''} === 测试查询: 查看销售部上个季度的订单数量 === 结果: {'success': False, 'error': '权限验证失败'} # 如果当前用户不是经理 === 测试查询: 帮我查一下李四上周的客户数量 === 结果: {'success': True, 'data': [{'customer_count': 23, 'sales_date': '2024-02-01', 'employee_name': '李四'}], 'query': 'SELECT customer_count, sales_date, employee_name FROM sales_data WHERE sales_date >= DATE_SUB(CURDATE(), INTERVAL 7 DAY) AND employee_name = '李四''}

6. 高级特性与扩展实践

6.1 支持多步骤复杂查询

对于需要多个操作步骤的复杂查询，可以扩展意图模型支持工作流：

# src/models/intent_models.py class ComplexIntent(BaseModel): """复杂意图模型，支持多步骤操作""" steps: List[SalesQueryIntent] = Field(description="操作步骤序列") dependencies: Dict[str, str] = Field(description="步骤间依赖关系") class WorkflowEngine: """工作流引擎，管理多步骤执行""" def execute_workflow(self, complex_intent: ComplexIntent) -> dict: results = {} for step in complex_intent.steps: # 检查依赖条件 if self._check_dependencies(step, results): step_result = self.execute_single_step(step) results[step.step_id] = step_result return results

6.2 集成企业规则引擎

对于已有 Drools 等规则引擎的企业，可以集成而不是重写：

# src/rules/drools_integration.py import requests class DroolsRuleEngine: def __init__(self, drools_server_url: str): self.server_url = drools_server_url def validate_intent(self, intent: dict, context: dict) -> bool: """调用 Drools 规则引擎进行验证""" payload = { "intent": intent, "context": context # 用户上下文、权限信息等 } response = requests.post( f"{self.server_url}/validate", json=payload, timeout=10 ) return response.json().get("approved", False)

6.3 性能优化与缓存策略

# src/agents/cached_agent.py import redis import hashlib import json class CachedSalesAgent(SalesAgent): def __init__(self, *args, redis_url: str, **kwargs): super().__init__(*args, **kwargs) self.redis_client = redis.from_url(redis_url) def process_query(self, user_input: str) -> dict: # 生成缓存键 cache_key = self._generate_cache_key(user_input) # 检查缓存 cached_result = self.redis_client.get(cache_key) if cached_result: return json.loads(cached_result) # 执行查询 result = super().process_query(user_input) # 缓存结果（仅缓存成功的查询） if result.get("success"): self.redis_client.setex( cache_key, 300, # 5分钟缓存 json.dumps(result) ) return result def _generate_cache_key(self, user_input: str) -> str: """生成基于用户输入和上下文的缓存键""" base_string = f"{user_input}_{self.current_user}" return hashlib.md5(base_string.encode()).hexdigest()

7. 常见问题与排查指南

7.1 意图解析失败

问题现象：LLM 无法正确解析用户意图，返回错误或不符合预期的结构。

排查步骤：

1. 检查提示词（Prompt）是否清晰定义了输出格式
2. 验证 Pydantic 模型定义是否与 LLM 输出匹配
3. 确认温度（Temperature）设置是否过低导致创造性不足，或过高导致输出不稳定
4. 检查 API 调用是否超时或限流

解决方案：

# 优化提示词设计 def create_enhanced_prompt(self, user_input: str, intent_model: Any) -> str: schema_example = intent_model.schema() return f""" 请严格按以下要求解析用户意图： 用户输入：{user_input} 输出必须是有效的 JSON，符合此 schema： {json.dumps(schema_example, indent=2)} 请确保： 1. 所有字段类型正确 2. 枚举值从指定选项中选择 3. 可选字段可以为 null 4. 不要添加额外字段 如果信息不全，在 error_message 中说明缺少什么信息。 """

7.2 规则验证误判

问题现象：合理的请求被规则引擎拒绝，或不应允许的请求被通过。

排查步骤：

1. 检查规则逻辑是否正确实现了业务政策
2. 验证用户上下文信息（如角色、权限）是否准确传递
3. 确认规则执行的顺序和条件判断
4. 查看审计日志分析具体拒绝原因

解决方案：

# 添加规则调试信息 class DebuggableSalesRules(SalesRules): def validate_query_permission(self, intent: SalesQueryIntent) -> tuple[bool, str]: """返回验证结果和详细原因""" checks = [] # 检查1：员工数据权限 if intent.employee_name and intent.employee_name != self.current_user: is_subordinate = self._is_subordinate(intent.employee_name) checks.append(f"下属检查: {is_subordinate}") if not is_subordinate: return False, " | ".join(checks) # 检查2：部门数据权限 if intent.department and not self._is_manager(self.current_user): checks.append("经理权限检查: 失败") return False, " | ".join(checks) checks.append("所有检查通过") return True, " | ".join(checks)

7.3 数据库查询性能问题

问题现象：查询响应慢，特别是在数据量大的情况下。

优化策略：

1. 为常用查询字段添加数据库索引
2. 实现查询结果缓存
3. 限制查询时间范围，避免全表扫描
4. 使用分页查询大数据集

-- 添加优化索引 CREATE INDEX idx_sales_date ON sales_data(sales_date); CREATE INDEX idx_employee_date ON sales_data(employee_name, sales_date); CREATE INDEX idx_department_date ON sales_data(department, sales_date);

7.4 安全与权限问题

关键检查点：

• SQL 注入防护：使用参数化查询而非字符串拼接
• 权限最小化原则：每个用户只能访问必要的数据
• 敏感数据脱敏：在响应中隐藏个人身份信息等敏感字段
• API 密钥安全管理：使用环境变量或密钥管理服务

# 安全的参数化查询 def _build_safe_query(self, intent: SalesQueryIntent) -> text: """使用参数化查询防止 SQL 注入""" base_query = """ SELECT {metric}, sales_date, employee_name FROM sales_data WHERE sales_date >= DATE_SUB(CURDATE(), INTERVAL :days DAY) """ params = {"days": self._get_days_from_range(intent.time_range)} if intent.employee_name: base_query += " AND employee_name = :employee_name" params["employee_name"] = intent.employee_name elif intent.department: base_query += " AND department = :department" params["department"] = intent.department return text(base_query), params

8. 生产环境最佳实践

8.1 监控与可观测性

在生产环境中，需要完善的监控体系：

# src/monitoring/agent_monitor.py import prometheus_client from prometheus_client import Counter, Histogram, Gauge class AgentMonitor: def __init__(self): self.requests_total = Counter('agent_requests_total', 'Total requests', ['agent_type', 'status']) self.request_duration = Histogram('agent_request_duration_seconds', 'Request duration') self.active_requests = Gauge('agent_active_requests', 'Active requests') def track_request(self, agent_type: str): """跟踪请求开始""" self.active_requests.inc() return self.request_duration.time() def track_success(self, agent_type: str): """标记请求成功""" self.requests_total.labels(agent_type=agent_type, status='success').inc() self.active_requests.dec() def track_failure(self, agent_type: str, error_type: str): """标记请求失败""" self.requests_total.labels(agent_type=agent_type, status=f'failure_{error_type}').inc() self.active_requests.dec()

8.2 错误处理与重试机制

# src/utils/retry_utils.py import time from typing import Callable, Any def retry_with_backoff( func: Callable, max_retries: int = 3, initial_delay: float = 1.0, backoff_factor: float = 2.0 ) -> Any: """指数退避重试机制""" last_exception = None for attempt in range(max_retries + 1): try: return func() except Exception as e: last_exception = e if attempt == max_retries: break delay = initial_delay * (backoff_factor ** attempt) time.sleep(delay) raise last_exception

8.3 配置管理

使用分层配置管理不同环境：

# config/settings.py from pydantic_settings import BaseSettings from typing import Optional class Settings(BaseSettings): # 基础配置 app_name: str = "Governed Agent" environment: str = "development" # LLM 配置 openai_api_key: Optional[str] = None llm_model: str = "gpt-3.5-turbo" llm_temperature: float = 0.1 # 数据库配置 database_url: Optional[str] = None redis_url: Optional[str] = None # 规则配置 max_query_days: int = 365 enable_audit_log: bool = True class Config: env_file = ".env" env_file_encoding = "utf-8" # 环境特定配置 def get_settings() -> Settings: return Settings()

8.4 测试策略

完善的测试覆盖是质量保证的关键：

# tests/test_sales_agent.py import pytest from src.agents.sales_agent import SalesAgent from src.models.intent_models import SalesQueryIntent class TestSalesAgent: @pytest.fixture def agent(self): return SalesAgent("test_key", "sqlite:///test.db", "test_user") def test_permission_validation(self, agent): """测试权限验证逻辑""" # 测试正常查询 intent = SalesQueryIntent( action="query_sales_data", employee_name="test_user", # 查询自己 time_range="last_week", metric="sales_amount" ) assert agent.rules_engine.validate_query_permission(intent) == True # 测试越权查询 intent.employee_name = "other_user" # 查询他人 assert agent.rules_engine.validate_query_permission(intent) == False def test_intent_parsing(self, agent): """测试意图解析""" # 模拟 LLM 响应 test_response = { "action": "query_sales_data", "employee_name": "张三", "time_range": "last_month", "metric": "sales_amount" } # 验证模型解析 intent = SalesQueryIntent(**test_response) assert intent.action == "query_sales_data" assert intent.time_range == "last_month"

Governed Agent 架构为企业级 AI 应用提供了一条切实可行的路径：既享受 LLM 的自然语言理解能力，又保持业务系统的确定性和可控性。这种设计模式特别适合对可靠性、安全性和合规性要求较高的生产环境。

在实际项目中，建议从简单的用例开始，逐步扩展规则复杂度和集成范围。重点关注权限管理、审计日志和错误处理等企业级特性，确保系统既智能又可靠。随着业务需求的发展，可以进一步探索工作流引擎集成、多模态能力扩展等高级特性。