Agent身份与权限系统​命令行交互入口：分层架构 + 多 Agent 协作模式

一、程序概述与架构设计

Agent 身份与权限系统​的命令行交互入口，采用分层架构 + 多Agent协作模式，核心价值在于：屏蔽底层 API 调用的复杂性，让用户通过自然语言直接驱动完整的智能体工作流。

程序整体分为四层：

1. CLI 交互层（用户输入解析、菜单展示）

2. 调度层（本地模式 / API 模式路由、意图识别）

3. Agent 执行层（DocAgent / WebAgent / DataAgent 职责隔离）

4. 基础设施层（LLM 客户端、HTTP 通信、配置管理）

二、核心功能详解

2.1 双运行模式

两种模式共享同一套自然语言指令解析逻辑，差异仅在于执行载体。

2.2 三大 Agent 权限模型（核心设计亮点）

程序内置RBAC + 最小权限原则，三个 Agent 形成严格的职责隔离：

🤖 文档助手 Agent（doc-assistant）

• 角色：协调者（Orchestrator）

• ✅ 可生成报告、委托其他 Agent、读写飞书文档

• ❌ 禁止直连数据库、禁止访问外网

• 典型场景：接收用户指令 → 判断是否调用 WebAgent → 汇总结果生成最终报告

🤖 企业数据 Agent（data-agent）

• 角色：内部数据守门人

• ✅ 读写飞书多维表格、通讯录、日历

• ❌ 禁止外网访问、禁止委托他人

• 隐含逻辑：所有敏感企业内部数据必须经过此 Agent

🤖 外部检索 Agent（web-agent）

• 角色：唯一外网出口

• ✅ 网络搜索、网页抓取

• ❌ 禁止触碰任何内部资源

• 安全机制：一旦尝试访问内网，立即触发越权拦截

这一设计从架构层面杜绝了“一个 Agent 通吃所有权限”的安全风险。

2.3 自然语言执行引擎

用户无需记忆命令，直接输入如：

• “帮我生成番茄小说数据分析报告”

• “测试越权访问内部数据”

程序通过IntentRecognizer完成意图分类 + 参数提取，映射到具体任务：

{ "task_type": "generate_report", "call_web_agent": { "decision": "yes", "keyword": "都市风云" } }

2.4 智能降级机制（生产级健壮性）

当 WebAgent 离线时，系统自动降级：

• 检测/health接口

• 若 WebAgent 不可用 → 关闭need_web_search

• 仅使用内部数据生成报告

• 明确提示用户“已降级”，避免静默失败

2.5 审计与可追溯性

• 每次操作生成Trace ID（UUID）

• 支持通过query_audit_log(trace_id)回溯完整调用链

• 满足企业级合规审计要求

三、数据结构设计

3.1 配置结构（YAML）

llm: provider: "openai" model: "gpt-4" api_key: "***" server: base_url: "http://localhost:8000"

配置通过secrets.yaml注入，避免硬编码。

3.2 核心数据对象

用户信息对象

user_info = { "user_id": "user-001", "user_name": "张三", "user_role": "admin" }

API 请求载荷

payload = { "user_id": str, "user_name": str, "user_role": str, "instruction": str }

执行结果结构

{ "success": bool, "code": int, "data": dict, "error": str, "trace_id": str }

3.3 LLM 上下文结构

messages = [ {"role": "system", "content": "系统提示词"}, {"role": "user", "content": "用户指令 + 执行结果"} ]

Prompt 工程重点：

• 禁止返回 JSON

• 隐藏技术细节

• 突出业务价值

四、算法与逻辑流程

4.1 主控制流（简化版）

用户输入 ↓ 是否为系统指令？（help / exit / 权限说明） ↓ 是否为审计查询？ ↓ 是否为越权测试？ ↓ 否则 → 意图识别 ↓ 路由到对应 Agent ↓ LLM 总结结果 ↓ 输出给用户

4.2 意图识别算法

IntentRecognizer.analyze()内部通常基于：

• 规则匹配（关键词：报告 / 测试 / 查询）

• LLM 分类（少量样本提示）

• 结构化输出解析

输出严格约束为 JSON，便于程序消费。

4.3 健康预检算法

if call_web: health = doc_agent.check_agent_health() if not health["web_agent"]: 自动降级

这是典型的断路器模式（Circuit Breaker）​ 轻量实现。

4.4 LLM 结果总结算法

1. 将原始 JSON 序列化

2. 构造 Prompt，强调“用户视角”

3. 调用 LLM 生成自然语言

4. 异常兜底：LLM 失败时回退到 JSON 打印

五、安全设计深度分析

5.1 权限边界清晰

• 外网 / 内网 Agent 物理隔离

• 无超级 Agent

5.2 越权测试内置

专门提供try-access-internal接口，用于：

• 渗透测试

• 权限回归验证

5.3 输入清洗

• 正则提取 Trace ID

• 超时控制（10s / 30s / 300s 分级）

六、扩展性设计

• 新增 Agent：只需实现统一接口，注册到 DocAgent

• 新增指令：扩展IntentRecognizer规则

• 替换 LLM：通过get_llm_client抽象，支持多厂商切换

七、典型使用场景

1. 运营人员：一句话生成小说数据报告

2. 安全工程师：定期运行越权测试

3. 审计人员：通过 Trace ID 追溯任意操作

4. 开发者：本地模式快速调试 Agent 行为

八、总结

并非一个简单的脚本，而是一个具备生产级安全、可观测、可降级、可扩展的智能体调度系统。其最大价值在于：

用自然语言作为新的人机交互协议，在严格权限沙箱中，安全释放 LLM 的自动化能力。

源代码

#!/usr/bin/env python3 """命令行交互入口，无需手动输入curl命令""" import sys import requests import json import yaml import os from typing import Dict from src.llm import get_llm_client from src.llm.intent_recognizer import IntentRecognizer # 加载配置 config_path = os.path.join(os.path.dirname(__file__), "config", "secrets.yaml") with open(config_path, 'r', encoding='utf-8') as f: config = yaml.safe_load(f) # 初始化LLM llm = get_llm_client(config.get("llm", {})) API_BASE = config.get("server", {}).get("base_url", "http://localhost:8000") def print_banner(): print("=" * 70) print("🔥 Agent身份与权限系统 命令行交互工具 ") print("=" * 70) def print_main_menu(): print("\n📋 功能菜单：") print("1. 📊 生成番茄小说数据分析报告（完整Agent协作流程）") print("2. 🔍 生成包含指定小说网络热度的分析报告") print("3. 🔒 测试越权访问拦截功能") print("4. 📝 查询审计日志（按Trace ID）") print("5. ℹ️ 查看当前Agent权限配置说明") print("6. ❌ 退出程序") print("-" * 70) def print_agent_permissions(): """打印Agent权限配置说明""" print("\n📋 三个Agent的职责与权限说明：") print("\n🤖 文档助手Agent (doc-assistant) | 角色：协调者") print("✅ 权限：生成报告、委托其他Agent、读写飞书文档") print("❌ 禁止：直接读取内部多维表格、访问外部网络") print("-" * 50) print("\n🤖 企业数据Agent (data-agent) | 角色：数据访问者") print("✅ 权限：读取/写入飞书多维表格、通讯录、日历等内部数据") print("❌ 禁止：访问外部网络、委托其他Agent、直接写入飞书文档") print("-" * 50) print("\n🤖 外部检索Agent (web-agent) | 角色：外部检索者") print("✅ 权限：网络搜索、网页抓取公开信息") print("❌ 禁止：访问任何内部资源、委托其他Agent") print("-" * 70) def call_nl_api(user_id: str, user_name: str, user_role: str, instruction: str) -> Dict: """调用自然语言执行接口""" url = f"{API_BASE}/api/nl-execute" payload = { "user_id": user_id, "user_name": user_name, "user_role": user_role, "instruction": instruction } try: response = requests.post(url, json=payload, timeout=300) # 超时时间调整为5分钟，适配本地LLM生成报告 return response.json() except Exception as e: return {"success": False, "error": str(e), "code": 500} def call_unauthorized_test() -> Dict: """调用越权测试接口""" url = f"{API_BASE}/api/try-access-internal" try: response = requests.post(url, timeout=30) return response.json() except Exception as e: return {"success": False, "error": str(e), "code": 500} def query_audit_log(trace_id: str) -> Dict: """查询审计日志""" url = f"{API_BASE}/api/audit/trace/{trace_id}" try: response = requests.get(url, timeout=10) return response.json() except Exception as e: return {"success": False, "error": str(e), "code": 500} def summarize_result(user_instruction: str, result: Dict) -> str: """使用LLM总结执行结果，返回自然语言回答""" try: result_str = json.dumps(result, ensure_ascii=False, indent=2) prompt = f""" 请作为智能助手，根据用户的指令和执行结果，生成友好、清晰的自然语言回答，不要返回JSON格式，直接返回自然语言内容。 用户指令：{user_instruction} 执行结果：{result_str} 注意事项： 1. 如果执行成功，重点突出核心结果，比如报告链接、关键数据等 2. 如果执行失败，清晰说明失败原因和解决方案 3. 回答简洁明了，符合用户的问题需求 4. 不要暴露内部技术细节和JSON结构 """ messages = [ {"role": "system", "content": "你是一个友好的智能助手，擅长将技术执行结果转换成用户易懂的自然语言回答。"}, {"role": "user", "content": prompt} ] return llm.chat(messages) except Exception as e: # 如果LLM调用失败，返回默认格式 return "" def main(): import argparse parser = argparse.ArgumentParser(description="Agent系统命令行工具") parser.add_argument("--local", action="store_true", help="本地运行模式，无需启动API服务") parser.add_argument("--non-interactive", action="store_true", help="非交互模式，直接执行输入的指令") args = parser.parse_args() # 初始化本地模式需要的实例 doc_agent = None web_agent = None intent_recognizer = None if args.local: from src.agents.doc_agent import DocAgent from src.agents.web_agent import WebAgent doc_agent = DocAgent() web_agent = WebAgent() intent_recognizer = IntentRecognizer() print_banner() if args.local: print("💡 当前运行模式：本地模式（无需启动API服务）") else: print("💡 当前运行模式：API模式（需先启动 python -m src.api.main 服务）") print("💡 提示：直接输入自然语言指令即可执行，输入 'help' 查看功能列表，输入 'exit' 退出") print("=" * 70) # 默认用户信息 user_info = { "user_id": "user-001", "user_name": "张三", "user_role": "admin" } while True: try: user_input = input("\n请输入指令：").strip() except KeyboardInterrupt: print("\n再见！") sys.exit(0) if not user_input: continue user_input_lower = user_input.lower() if user_input_lower in ["exit", "quit", "q"]: print("再见！") sys.exit(0) if user_input_lower == "help": print("\n📋 支持的指令示例：") print("• 帮我生成番茄小说数据分析报告") print("• 帮我生成包含都市风云网络热度的分析报告") print("• 测试越权访问内部数据") print("• 查询审计日志 ") print("• 查看Agent权限说明") print("• help 显示帮助信息") print("• exit 退出程序") continue if "查看权限" in user_input or "agent权限" in user_input: print_agent_permissions() continue if "查询审计日志" in user_input or "trace id" in user_input_lower: # 提取Trace ID import re trace_match = re.search(r'([a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12})', user_input, re.IGNORECASE) if trace_match: trace_id = trace_match.group(1) print(f"\n📝 正在查询Trace ID: {trace_id} 的审计日志...") result = query_audit_log(trace_id) else: trace_id = input("请输入Trace ID：").strip() if trace_id: result = query_audit_log(trace_id) else: print("❌ 未检测到Trace ID！") continue elif "越权" in user_input or "测试访问内部" in user_input or "尝试读取表格" in user_input: print("\n🔒 正在执行越权访问测试...") if args.local: result = web_agent.try_access_internal_data() else: result = call_unauthorized_test() else: # 自然语言指令直接执行 print(f"\n🚀 正在执行指令：{user_input}") if args.local: intent = intent_recognizer.analyze(user_input) if intent["task_type"] == "generate_report": call_web = intent["call_web_agent"]["decision"] == "yes" keyword = intent["call_web_agent"].get("keyword") # ---- 健康预检：WebAgent离线时自动降级 ---- if call_web: health = doc_agent.check_agent_health() if not health["web_agent"]: print("\n[DocAgent] 外部检索Agent(8003)未启动！系统自动降级：仅使用内部数据。") call_web = False result = doc_agent.generate_novel_analysis_report( delegated_user=user_info, keyword=keyword, need_web_search=call_web ) result["intent"] = intent result["instruction"] = user_input elif intent["task_type"] == "test_unauthorized_access": result = web_agent.try_access_internal_data() result["intent"] = intent result["instruction"] = user_input else: result = {"success": False, "error": "不支持的指令类型", "intent": intent, "code": 400} else: # API模式调用接口 result = call_nl_api(**user_info, instruction=user_input) # 生成LLM总结 summary = summarize_result(user_input, result) # 打印结果 print("\n📌 执行结果：") if summary: print("\n" + summary + "\n") else: # 如果没有总结，打印原始JSON print(json.dumps(result, ensure_ascii=False, indent=2)) # 打印额外信息 if result.get("success"): # 显示 WebAgent 降级警告 warning = (result.get("data") or {}).get("web_agent_warning") if warning: print(f"\n⚠️ {warning.replace('[DocAgent] ', '').strip()}") if "report_url" in result.get("data", {}): print(f"✅ 报告链接：{result['data']['report_url']}") else: print(f"\n❌ 执行失败：{result.get('error', '未知错误')}") print(f"错误码：{result.get('code')}") if result.get("trace_id"): print(f"\n🔍 Trace ID：{result['trace_id']}，可用于查询审计日志") if __name__ == "__main__": main()