一、程序概述与核心定位
一个面向分布式智能体协作系统的全链路验证脚本，主要用于检测多智能体架构下的核心组件可用性、意图识别准确性、服务健康状态、权限控制有效性及审计日志完整性。该脚本通过模块化测试覆盖从底层智能体初始化、自然语言意图解析到上层服务治理、安全审计的完整业务流程，是保障多智能体系统（DocAgent、DataAgent、WebAgent 协同架构）稳定运行的关键验证工具。
脚本采用「轻量级探测 + 结果持久化」设计模式，通过 6 组独立测试用例（Test 1至 Test 6）对系统进行无侵入式检测，最终将结果写入 verify_result.txt文件并输出执行状态。这种设计既避免了复杂测试框架的依赖负担，又能快速定位系统故障点，适用于开发环境调试、预发布环境校验及生产环境巡检场景。
二、整体架构与执行流程
2.1 执行时序
脚本执行遵循「初始化→组件探测→服务校验→结果输出」的线性流程，各阶段严格解耦：
环境准备：通过 sys.path.insert注入项目根目录路径，确保跨目录导入模块（如 src.agents.doc_agent）的可行性，解决 Python 包路径问题。
测试执行：按顺序执行 6 个独立测试模块，每个模块专注单一验证目标，测试结果实时追加至 res列表（内存存储，避免频繁 IO）。
结果持久化：所有测试完成后，将 res列表按行写入 verify_result.txt，并通过 print('OK')输出最终执行状态。
2.2 核心依赖
|
系统库 |
sys、os |
路径管理，确保自定义模块可被正确导入 |
|
HTTP 客户端 |
requests |
探测微服务健康状态、模拟未授权请求 |
|
业务逻辑模块 |
src.agents.doc_agent |
文档智能体核心类，封装多智能体协作入口 |
|
自然语言处理 |
src.llm.intent_recognizer |
意图识别器，解析用户输入的自然语言指令 |
|
审计与权限 |
src.audit_service.query、src.policy_engine.static_policy |
审计日志查询、静态权限策略校验 |
三、核心测试模块深度解析
3.1 Test 1：DocAgent 组件依赖验证
功能目标
验证文档智能体（DocAgent）是否正确集成了下游协作组件，确保多智能体协作的基础架构完整。DocAgent作为系统入口，需依赖数据智能体（data_agent）、网络智能体（web_agent）和飞书客户端（feishu_client）完成复杂任务。
代码实现
from src.agents.doc_agent import DocAgent
da = DocAgent()
res.append(f"DocAgent.has_data_agent={hasattr(da,'data_agent')}")
res.append(f"DocAgent.has_web_agent={hasattr(da,'web_agent')}")
res.append(f"DocAgent.has_feishu_client={hasattr(da,'feishu_client')}")
技术细节
-
hasattr动态检测：通过 Python 内置函数 hasattr(obj, attr)检查对象是否包含指定属性，避免直接访问未定义属性导致的 AttributeError。这种方式适合验证「依赖注入是否成功」，而非组件功能本身。
-
依赖关系设计：DocAgent采用「聚合依赖」模式，即自身不直接实现数据处理或网络检索功能，而是通过持有 data_agent（负责表格分析、数据统计）、web_agent（负责外部知识检索）、feishu_client（负责飞书文档读写）的引用实现能力扩展。这种设计的优势是降低耦合度，便于单独升级某个智能体。
验证意义
若某属性返回 False，则表明：
-
data_agent缺失：文档智能体无法处理结构化数据（如 Excel、CSV）；
-
web_agent缺失：无法结合外部实时信息生成报告；
-
feishu_client缺失：无法与飞书生态打通（如导出报告到飞书文档）。
3.2 Test 2：意图识别器（IntentRecognizer）功能验证
功能目标
验证自然语言意图解析的准确性，确保系统能正确判断用户输入是否需要调用数据智能体或网络智能体。这是多智能体协作的「决策中枢」，直接影响任务路由效率。
代码实现
from src.llm.intent_recognizer import IntentRecognizer
ir = IntentRecognizer()
for ins in ["生成番茄小说数据分析报告","读取表格数据同时结合外部检索到的知识生成分析报告"]:
parsed = ir.analyze(ins)
res.append(f"INTENT: {ins[:30]} -> call_data={parsed['call_data_agent']['decision']}, call_web={parsed['call_web_agent']['decision']}")
数据结构：parsed字典
意图识别结果采用嵌套字典结构，核心字段如下：
{
"call_data_agent": {
"decision": True/False, # 是否需要调用数据智能体
"confidence": 0.95, # 决策置信度（示例中未显式输出）
"reason": "包含'数据分析'关键词" # 决策依据（示例中未显式输出）
},
"call_web_agent": {
"decision": True/False, # 是否需要调用网络智能体
"confidence": 0.88,
"reason": "包含'外部检索'关键词"
}
}
算法逻辑推测
尽管代码未展示 IntentRecognizer内部实现，但基于测试用例可推断其采用混合意图识别策略：
关键词匹配层：对输入文本进行分词，匹配预设关键词（如「数据」「表格」→ 触发 call_data_agent；「外部」「检索」「知识」→ 触发 call_web_agent）。
语义理解层：结合 LLM（大语言模型）进行上下文语义分析，处理复杂表述（如示例中的长句「读取表格数据同时结合外部检索到的知识生成分析报告」需同时触发两个智能体）。
决策融合层：综合关键词权重与语义相似度，输出最终 decision（布尔值）。
测试案例设计
|
「生成番茄小说数据分析报告」 |
True |
False |
仅涉及本地数据分析，无需外部检索 |
|
「读取表格+外部知识生成报告」 |
True |
True |
需同时调用数据智能体（处理表格）和网络智能体（检索外部知识） |
3.3 Test 3：微服务健康状态探测
功能目标
验证下游微服务（data-agent和 web-agent）的存活性与可用性，确保分布式系统中各节点正常运行。
代码实现
import requests
for p,n in [(8002,'DA'),(8003,'WA')]:
try:
r = requests.get(f'http://localhost:{p}/health',timeout=3)
res.append(f"HEALTH {n}: {r.status_code} {r.json().get('status','?')}")
except Exception as e:
res.append(f"HEALTH {n}: DOWN ({str(e)[:60]})")
服务端口映射
|
8002 |
DA |
DataAgent |
数据智能体服务，处理表格分析、统计计算 |
|
8003 |
WA |
WebAgent |
网络智能体服务，负责外部网页检索、知识爬取 |
健康检测机制
-
HTTP 健康检查端点：微服务通过 /health路径暴露健康状态，返回 JSON 格式数据（如 {"status": "healthy", "timestamp":}）。
-
超时控制：设置 timeout=3秒，避免单个服务故障阻塞整个验证流程。
-
异常处理：捕获 requests.exceptions.ConnectionError（服务未启动）、Timeout（响应超时）、JSONDecodeError（返回非 JSON 数据）等异常，标记服务为 DOWN并记录错误原因（截断前 60 字符避免日志冗余）。
系统设计启示
该测试反映了系统的微服务化架构：DocAgent作为前端智能体，通过 HTTP 调用后端 DataAgent和 WebAgent服务，实现「前端交互-后端计算」分离，提升系统扩展性（可独立扩容某个微服务）。
3.4 Test 4：未授权访问防护验证
功能目标
检测系统是否存在未授权访问漏洞，验证 API 接口的认证机制是否生效。
代码实现
try:
r = requests.post('',timeout=10)
d = r.json()
res.append(f"UNAUTH: success={d.get('success')}, code={d.get('code')}")
except Exception as e:
res.append(f"UNAUTH: err={str(e)[:80]}")
测试逻辑分析
-
空请求构造：发送 POST请求至空 URL（实际场景中可能指向系统核心 API，如 /api/generate-report），不携带认证头（如 Authorization: Bearer <token>）。
-
预期结果：正常情况下，未授权请求应返回 401 Unauthorized或 403 Forbidden，且 success=False、code=401。
-
异常捕获：若服务端未做认证拦截，可能返回 200 OK并执行敏感操作（如生成报告），此时 success=True即为安全漏洞；若服务端崩溃或返回非 JSON 响应，则捕获异常并记录错误。
安全意义
该测试是最小权限原则的体现，确保只有经过身份验证的请求才能访问系统资源，防止未授权用户滥用智能体能力（如恶意生成大量报告消耗算力）。
3.5 Test 5：审计日志完整性验证
功能目标
验证审计系统是否正常记录操作日志，确保可追溯性（合规要求）和问题排查能力。
代码实现
from src.audit_service.query import AuditQuery
aq = AuditQuery()
logs = aq._load_all_logs()
res.append(f"AUDIT: total={len(logs)}, allow={sum(1 for l in logs if l.get('decision')=='ALLOW')}, deny={sum(1 for l in logs if l.get('decision')=='DENY')}")
审计日志数据结构
logs为列表类型，每个元素为单条审计日志记录，典型结构如下：
统计逻辑
-
total：总日志条数，反映系统操作频次；
-
allow：decision == 'ALLOW'的日志数，代表合法操作；
-
deny：decision == 'DENY'的日志数，代表被权限策略拦截的操作。
审计系统价值
-
安全分析：通过 deny日志发现异常访问尝试（如暴力破解、越权操作）；
-
问题追溯：当生成错误报告时，可通过日志反查用户操作路径。
3.6 Test 6：静态权限策略验证
功能目标
验证权限引擎是否正确配置静态策略，确保不同智能体仅拥有最小必要权限（Least Privilege Principle）。
代码实现
from src.policy_engine.static_policy import StaticPolicy; sp=StaticPolicy()
res.append(f"PERM doc: delegate={sp.check_static_capability('doc-assistant','delegate:*')}, bitable={sp.check_static_capability('doc-assistant','feishu:bitable')}")
res.append(f"PERM data: web={sp.check_static_capability('data-agent','web:search')}, delegate={sp.check_static_capability('data-agent','delegate:*')}")
res.append(f"PERM web: bitable={sp.check_static_capability('web-agent','feishu:bitable')}, delegate={sp.check_static_capability('web-agent','delegate:*')}")
权限模型设计
采用基于角色的访问控制（RBAC）+ 资源粒度权限：
-
主体（Subject）：智能体身份（如 doc-assistant、data-agent、web-agent）；
-
权限（Permission）：格式为 <资源类型>:<操作>，如 feishu:bitable（飞书多维表格操作权限）、web:search（网络搜索权限）；
-
通配符（）*：delegate:*表示「委托所有操作」的高级权限，仅允许特定智能体持有。
权限校验逻辑
check_static_capability(subject, permission)方法实现如下逻辑：
加载静态策略配置文件（如 policy.yaml），定义角色-权限映射；
检查 subject是否直接拥有 permission或通配符权限（如 delegate:*覆盖所有 delegate:前缀权限）；
返回 True（有权限）或 False（无权限）。
预期权限矩阵
|
doc-assistant |
feishu:bitable（读写飞书表格） |
delegate:*（不可委托高级操作） |
文档智能体仅需操作飞书资源，无需系统级委托权限 |
|
data-agent |
web:search（搜索外部数据） |
delegate:*（不可委托） |
数据智能体需结合外部数据，但不可越权操作系统其他组件 |
|
web-agent |
无 feishu:bitable权限 |
delegate:*（不可委托） |
网络智能体仅负责检索，不可直接操作飞书文档（避免数据泄露） |
四、数据结构与存储设计
4.1 核心数据结构
|
list[str] |
存储测试结果（res变量） |
["DocAgent.has_data_agent=True", "HEALTH DA: 200 healthy", …] |
|
dict |
意图识别结果、HTTP 响应解析 |
{"call_data_agent": {"decision": True}, "call_web_agent": {…}} |
|
list[dict] |
审计日志集合 |
[{"timestamp": "…", "decision": "ALLOW"}, …] |
4.2 结果存储
-
临时存储：res列表在内存中累积测试结果，避免频繁磁盘 IO 影响性能；
-
持久化存储：最终写入 verify_result.txt（UTF-8 编码），每行一条测试结果，格式统一为 KEY: VALUE，便于后续脚本解析（如通过 grep "HEALTH DA: DOWN"快速定位故障）。
五、算法与设计模式
5.1 设计模式应用
|
门面模式 |
DocAgent封装多智能体调用 |
简化上层调用，隐藏下游复杂性 |
|
策略模式 |
意图识别（关键词+语义双策略） |
可灵活切换识别算法（如替换 LLM 模型） |
|
模板方法模式 |
各测试模块统一遵循「探测→记录→异常处理」流程 |
确保测试逻辑一致性，降低维护成本 |
5.2 容错算法
-
优雅降级：单个测试失败（如服务健康检测超时）不影响其他测试执行，res列表继续追加结果；
-
错误截断：异常信息截断为前 60/80 字符（如 str(e)[:60]），避免冗长错误信息污染结果文件；
-
无状态设计：所有测试不依赖外部状态（如数据库事务），可重复执行且结果一致。
六、系统局限性与优化建议
6.1 现有局限性
测试覆盖不全：缺少对 FeishuClient功能的验证（仅检测存在性，未测试实际读写能力）；
硬编码依赖：端口号（8002、8003）、URL 路径（/health）硬编码，环境变更需修改脚本；
无自动化告警：仅输出结果文件，未对接监控系统（如 Prometheus、企业微信告警）。
6.2 优化建议
参数化配置：将端口、URL、权限主体等通过配置文件（如 config.yaml）注入，支持多环境切换；
增强断言：对关键指标（如 AUDIT deny > 0）添加阈值校验，超过阈值时返回非零退出码（便于 CI/CD 流水线捕获失败）；
可视化报告：将 verify_result.txt转换为 HTML 报告，用颜色标注通过/失败状态（如绿色√、红色×）。
七、总结
作为分布式智能体系统的「健康检查中心」，通过轻量级、模块化的测试设计，实现了对组件依赖、意图识别、服务可用性、安全防护、审计合规、权限控制六大核心维度的验证。其背后反映的是现代 AI 系统的工程化实践：不仅需要关注模型效果，更需要通过完善的验证机制保障系统稳定性、安全性和合规性。
该脚本的价值远超单一测试工具——它定义了多智能体系统的「验收标准」，为后续迭代提供了可靠的回归验证能力，是连接 AI 算法研发与生产落地的关键桥梁。
源代码
import sys,os;sys.path.insert(0,os.path.dirname(os.path.abspath(__file__)))
res = []
# Test 1
from src.agents.doc_agent import DocAgent
da = DocAgent()
res.append(f"DocAgent.has_data_agent={hasattr(da,'data_agent')}")
res.append(f"DocAgent.has_web_agent={hasattr(da,'web_agent')}")
res.append(f"DocAgent.has_feishu_client={hasattr(da,'feishu_client')}")
# Test 2 – Intent
from src.llm.intent_recognizer import IntentRecognizer
ir = IntentRecognizer()
for ins in ["生成番茄小说数据分析报告","读取表格数据同时结合外部检索到的知识生成分析报告"]:
parsed = ir.analyze(ins)
res.append(f"INTENT: {ins[:30]} -> call_data={parsed['call_data_agent']['decision']}, call_web={parsed['call_web_agent']['decision']}")
# Test 3 – Health
import requests
for p,n in [(8002,'DA'),(8003,'WA')]:
try:
r = requests.get(f'http://localhost:{p}/health',timeout=3)
res.append(f"HEALTH {n}: {r.status_code} {r.json().get('status','?')}")
except Exception as e:
res.append(f"HEALTH {n}: DOWN ({str(e)[:60]})")
# Test 4 – Unauthorized
try:
r = requests.post('',timeout=10)
d = r.json()
res.append(f"UNAUTH: success={d.get('success')}, code={d.get('code')}")
except Exception as e:
res.append(f"UNAUTH: err={str(e)[:80]}")
# Test 5 – Audit
from src.audit_service.query import AuditQuery
aq = AuditQuery()
logs = aq._load_all_logs()
res.append(f"AUDIT: total={len(logs)}, allow={sum(1 for l in logs if l.get('decision')=='ALLOW')}, deny={sum(1 for l in logs if l.get('decision')=='DENY')}")
# Test 6 – Permissions
from src.policy_engine.static_policy import StaticPolicy; sp=StaticPolicy()
res.append(f"PERM doc: delegate={sp.check_static_capability('doc-assistant','delegate:*')}, bitable={sp.check_static_capability('doc-assistant','feishu:bitable')}")
res.append(f"PERM data: web={sp.check_static_capability('data-agent','web:search')}, delegate={sp.check_static_capability('data-agent','delegate:*')}")
res.append(f"PERM web: bitable={sp.check_static_capability('web-agent','feishu:bitable')}, delegate={sp.check_static_capability('web-agent','delegate:*')}")
open('verify_result.txt','w',encoding='utf-8').write('\\n'.join(res))
print('OK')
