如果你正在构建企业级的 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 AgentGoverned 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 的对比特性传统 AgentGoverned Agent决策主体LLM规则引擎 代码确定性低LLM 可能产生变异高代码逻辑固定可审计性困难容易有明确的规则路径集成成本高需要重写业务逻辑低复用现有系统适用场景创意生成、探索性任务企业业务流程、数据操作3. 环境准备与基础依赖3.1 系统要求Python 3.8Governed 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] None4.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_keyapi_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( modelself.model, messages[{role: user, content: prompt}], temperature0.1 # 低温度确保输出稳定性 ) # 解析 LLM 响应 result json.loads(response.choices[0].message.content) intent intent_model(**result) return IntentResponse(successTrue, intentintent) except Exception as e: return IntentResponse(successFalse, error_messagef意图解析失败: {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 managers4.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: 处理用户查询的完整流程 # 步骤1LLM 解析意图 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(fAUDIT_LOG: {audit_log})5. 完整示例部署与测试5.1 配置环境变量创建.env文件# .env OPENAI_API_KEYyour_openai_api_key_here DATABASE_URLmysqlpymysql://user:passwordlocalhost/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_keyos.getenv(OPENAI_API_KEY), db_urlos.getenv(DATABASE_URL), current_useros.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 results6.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, jsonpayload, timeout10 ) 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 无法正确解析用户意图返回错误或不符合预期的结构。排查步骤检查提示词Prompt是否清晰定义了输出格式验证 Pydantic 模型定义是否与 LLM 输出匹配确认温度Temperature设置是否过低导致创造性不足或过高导致输出不稳定检查 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, indent2)} 请确保 1. 所有字段类型正确 2. 枚举值从指定选项中选择 3. 可选字段可以为 null 4. 不要添加额外字段 如果信息不全在 error_message 中说明缺少什么信息。 7.2 规则验证误判问题现象合理的请求被规则引擎拒绝或不应允许的请求被通过。排查步骤检查规则逻辑是否正确实现了业务政策验证用户上下文信息如角色、权限是否准确传递确认规则执行的顺序和条件判断查看审计日志分析具体拒绝原因解决方案# 添加规则调试信息 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 数据库查询性能问题问题现象查询响应慢特别是在数据量大的情况下。优化策略为常用查询字段添加数据库索引实现查询结果缓存限制查询时间范围避免全表扫描使用分页查询大数据集-- 添加优化索引 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), params8. 生产环境最佳实践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_typeagent_type, statussuccess).inc() self.active_requests.dec() def track_failure(self, agent_type: str, error_type: str): 标记请求失败 self.requests_total.labels(agent_typeagent_type, statusffailure_{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_exception8.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( actionquery_sales_data, employee_nametest_user, # 查询自己 time_rangelast_week, metricsales_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_monthGoverned Agent 架构为企业级 AI 应用提供了一条切实可行的路径既享受 LLM 的自然语言理解能力又保持业务系统的确定性和可控性。这种设计模式特别适合对可靠性、安全性和合规性要求较高的生产环境。在实际项目中建议从简单的用例开始逐步扩展规则复杂度和集成范围。重点关注权限管理、审计日志和错误处理等企业级特性确保系统既智能又可靠。随着业务需求的发展可以进一步探索工作流引擎集成、多模态能力扩展等高级特性。