一、程序概述与架构设计Agent 身份与权限系统​的命令行交互入口采用分层架构 多Agent协作模式核心价值在于屏蔽底层 API 调用的复杂性让用户通过自然语言直接驱动完整的智能体工作流。程序整体分为四层CLI 交互层用户输入解析、菜单展示调度层本地模式 / API 模式路由、意图识别Agent 执行层DocAgent / WebAgent / DataAgent 职责隔离基础设施层LLM 客户端、HTTP 通信、配置管理二、核心功能详解2.1 双运行模式模式触发参数特点API 模式​默认依赖外部 HTTP 服务src.api.main适合生产部署本地模式​--local直接实例化 Agent 对象绕过网络调用适合开发与调试两种模式共享同一套自然语言指令解析逻辑差异仅在于执行载体。2.2 三大 Agent 权限模型核心设计亮点程序内置RBAC 最小权限原则三个 Agent 形成严格的职责隔离 文档助手 Agentdoc-assistant角色协调者Orchestrator✅ 可生成报告、委托其他 Agent、读写飞书文档❌ 禁止直连数据库、禁止访问外网典型场景接收用户指令 → 判断是否调用 WebAgent → 汇总结果生成最终报告 企业数据 Agentdata-agent角色内部数据守门人✅ 读写飞书多维表格、通讯录、日历❌ 禁止外网访问、禁止委托他人隐含逻辑所有敏感企业内部数据必须经过此 Agent 外部检索 Agentweb-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 IDUUID支持通过query_audit_log(trace_id)回溯完整调用链满足企业级合规审计要求三、数据结构设计3.1 配置结构YAMLllm: 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 结果总结算法将原始 JSON 序列化构造 Prompt强调“用户视角”调用 LLM 生成自然语言异常兜底LLM 失败时回退到 JSON 打印五、安全设计深度分析5.1 权限边界清晰外网 / 内网 Agent 物理隔离无超级 Agent5.2 越权测试内置专门提供try-access-internal接口用于渗透测试权限回归验证5.3 输入清洗正则提取 Trace ID超时控制10s / 30s / 300s 分级六、扩展性设计新增 Agent只需实现统一接口注册到 DocAgent新增指令扩展IntentRecognizer规则替换 LLM通过get_llm_client抽象支持多厂商切换七、典型使用场景运营人员一句话生成小说数据报告安全工程师定期运行越权测试审计人员通过 Trace ID 追溯任意操作开发者本地模式快速调试 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, encodingutf-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, jsonpayload, timeout300) # 超时时间调整为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, timeout30) 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, timeout10) 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_asciiFalse, indent2) 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(descriptionAgent系统命令行工具) parser.add_argument(--local, actionstore_true, help本地运行模式无需启动API服务) parser.add_argument(--non-interactive, actionstore_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_useruser_info, keywordkeyword, need_web_searchcall_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, instructionuser_input) # 生成LLM总结 summary summarize_result(user_input, result) # 打印结果 print(\n 执行结果) if summary: print(\n summary \n) else: # 如果没有总结打印原始JSON print(json.dumps(result, ensure_asciiFalse, indent2)) # 打印额外信息 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()