用 AgentMesh 治理 CrewAI为多智能体 Crew 构建零信任身份、范围链委派与信任评分【免费下载链接】agent-governance-toolkitAI Agent Governance Toolkit — Policy enforcement, zero-trust identity, execution sandboxing, and reliability engineering for autonomous AI agents. Covers 10/10 OWASP Agentic Top 10.项目地址: https://gitcode.com/GitHub_Trending/ag/agent-governance-toolkitCrewAI 提供了协作式多智能体工作流Crew / Agent / Task但多个 LLM Agent 协同执行任务时谁来验证这个 Agent 是谁、它被授权做什么、它与同伴的信任关系是否足够是生产部署的核心难题。Agent Governance Toolkit 中的 agent-mesh 包针对这一问题将 AgentMesh 的加密身份DID、可缩窄的范围链委派ScopeChain、信任握手TrustHandshake与协作信任评分RewardEngine接入 CrewAI 的 Crew 执行模型。读完本文你将掌握如何为 CrewAI 的 crew 成员签发可审计身份、以最小权限委派任务、在任务交接前执行信任校验并理解仓库中真正落地的TrustAwareAgent/TrustAwareCrew包装器的实现细节与测试验证方式。为什么要把 AgentMesh 接入 CrewAICrewAI 本身专注于多 Agent 协作定义若干 Agent 与 Task组建 Crew 后通过kickoff()顺序或层级执行。但协作框架并不内置安全语义。按 CrewAI 集成文档 的描述AgentMesh 在此基础上补齐了四类治理能力加密身份每个 crew 成员持有独立的AgentIdentityDID范围链Scope Chains委派链路中子 Agent 的能力只能缩窄、不能扩张跨 Agent 信任握手任务交接前先验证对端 DID 的信任分是否达标协作信任评分根据任务完成质量、协作表现与策略合规情况持续更新每个 Agent 的 0–1000 信任分。这正好对应仓库源码中 agent-mesh 的核心模块AgentIdentity定义于 identity/agent_id.pyScopeChain定义于 identity/delegation.py策略与审计分别位于 governance/policy.pyPolicyEngine与 governance/audit.pyAuditLog它们都通过顶层包 agentmesh/__init__.py 统一导出因此集成代码可以直接from agentmesh import ...。快速开始给 Crew 成员签发身份并委派能力安装pip install agentmesh-platform crewai crewai-tools基本集成文档给出的完整示例展示了supervisor 签发身份 → 逐级委派缩窄能力 → 挂到 CrewAI Agent的标准流程from crewai import Agent, Task, Crew from agentmesh import AgentIdentity, ScopeChain, PolicyEngine # Create supervisor identity supervisor_identity AgentIdentity.create( namecrew-supervisor, sponsorteam-leadcompany.com, capabilities[research, writing, review] ) # Create scope chain scope_chain ScopeChain(rootsupervisor_identity) # Delegate to crew members researcher_identity scope_chain.delegate( nameresearcher-agent, capabilities[research] # Narrowed from supervisor ) writer_identity scope_chain.delegate( namewriter-agent, capabilities[writing] # Narrowed from supervisor ) reviewer_identity scope_chain.delegate( namereviewer-agent, capabilities[review] # Narrowed from supervisor ) # Create CrewAI agents with AgentMesh identities researcher Agent( roleResearcher, goalResearch the topic thoroughly, backstoryExpert researcher with 10 years experience, agentmesh_identityresearcher_identity # Attach identity ) writer Agent( roleWriter, goalWrite engaging content, backstoryProfessional content writer, agentmesh_identitywriter_identity ) reviewer Agent( roleReviewer, goalReview and improve content, backstorySenior editor with high standards, agentmesh_identityreviewer_identity ) # Define tasks research_task Task( descriptionResearch the topic: AgentMesh governance for AI agents, agentresearcher ) writing_task Task( descriptionWrite a blog post based on the research, agentwriter ) review_task Task( descriptionReview and improve the blog post, agentreviewer ) # Create governed crew crew Crew( agents[researcher, writer, reviewer], tasks[research_task, writing_task, review_task], verboseTrue ) # Run with governance result crew.kickoff() print(fResult: {result}) print(fSupervisor DID: {supervisor_identity.did}) print(fCrew members: {len(scope_chain.links)})需要说明两点前提示例代码表达的是集成模式为每个 CrewAI Agent 关联一个 AgentMesh 身份对象。仓库中实际随包发布的 CrewAI 集成采用的是包装器/混入模式见下文仓库中的落地实现一节crewai是可选依赖即使不安装 CrewAI 也能运行纯信任操作能力缩窄是硬性约束。从源码结构看delegation.py 中DelegationLink的注释明确写着每个链接代表父级向子级授予能力子级的能力必须是父级的子集The childs capabilities MUST be a subset of the parents。因此示例中researcher只能拿到[research]而不能越权申请 supervisor 之外的能力委派深度也受DEFAULT_DELEGATION_MAX_DEPTH常量约束超出会抛出DelegationDepthError。进阶特性一crew 成员之间的信任握手在 writer 接受 researcher 交接的任务之前先用TrustHandshake校验对端 DID 的信任分。文档给出的模式如下from agentmesh import TrustHandshake # Before writer accepts work from researcher async def governed_task_handoff(from_agent, to_agent, task): handshake TrustHandshake() # Verify peer result await handshake.verify( peer_didfrom_agent.agentmesh_identity.did, required_score700 ) if not result.verified: raise SecurityError(fUntrusted peer: {result.reason}) # Accept task return to_agent.execute(task)TrustHandshake在源码中位于 trust/handshake.py由顶层包导出见 agentmesh/__init__.py。在 CrewAI 场景下它的语义是Crew 内部任一 Agent 在接收来自同伴的产出研究结论、初稿等之前必须确认对端信任分不低于required_score否则拒绝交接并抛出异常——这相当于把零信任原则落到了任务级数据流上。进阶特性二协作信任评分RewardEngine信任分不是静态配置而是随协作行为演化的。文档展示了如何用RewardEngine基于三个维度更新 crew 各成员的得分from agentmesh import RewardEngine reward_engine RewardEngine() # Update scores based on collaboration quality def update_crew_scores(crew): for agent in crew.agents: identity agent.agentmesh_identity # Score based on: # - Task completion quality # - Collaboration with other agents # - Policy compliance score reward_engine.update_score( agent_ididentity.did, dimensions{ task_quality: 0.9, collaboration: 0.85, policy_compliance: 1.0 } ) print(f{agent.role}: {score.total}/1000)RewardEngine的实现在 reward/engine.py同目录还提供trust_decay.py信任随时间衰减与distribution.py/distributor.py奖励分发因此可以推断生产环境中一个长期不活跃或长期违规的 Agent其信任分会因衰减与惩罚自然降到握手阈值之下从而被后续的信任校验自动隔离——无需人工下线。进阶特性三对 crew 任务执行策略强制在任务执行前插入PolicyEngine校验违规即拒绝执行并留痕policy_engine PolicyEngine.from_file(policies/crew.yaml) # Wrap task execution with policy checks def governed_task_execution(task, agent): # Check policy before execution result policy_engine.check( actionexecute_task, agentagent.agentmesh_identity.did, tasktask.description ) if not result.allowed: raise PermissionError(fPolicy violation: {result.reason}) # Execute task output agent.execute_task(task) # Audit audit_log.log(task_completed, agentagent.role, tasktask.description) return outputPolicyEngine定义于 governance/policy.py。该引擎是 agent-mesh 中多语言共享的策略求值核心与 policy-engine 目录中的 Rego/Cedar 后端配套check()返回allowed/reason配合 governance/audit.py 中的AuditLog即可实现先校验、再执行、后审计的闭环。仓库中的落地实现TrustAwareAgent 与 TrustAwareCrew上面三节对应文档中的集成模式仓库实际随agent-mesh包发布、可直接 import 的 CrewAI 集成在 integrations/crewai/ 下导出InMemoryTrustStore、InteractionRecord、TrustAwareAgent、TrustAwareCrew与TrustStore协议见 integrations/crewai/__init__.py。其设计要点值得逐条对照TrustAwareAgent包装器模式CrewAI 为可选依赖agent.py 中的TrustAwareAgent采用包装器/混入模式构造时接收agent_did如did:mesh:abc123、min_trust_score默认 500取值 0–1000、可选trust_store后端其余**kwargs透传给 CrewAI 的Agent构造函数。若环境中未安装 crewai则进入纯信任模式trust-only mode信任操作照常可用任务执行返回 stub 结果。核心方法有三个verify_peer(peer_did)L128查询对端信任分是否 ≥min_trust_score对应文档中信任握手的同步简化版execute_with_trust(task, context)L140先校验自身信任分不足则记录一次失败交互并抛出TrustViolationError执行成功/失败都会通过_record()写入交互历史并回调trust_store.record_interaction()更新得分delegate_with_trust(task, peer_did)L175仅当对端信任分达标才允许委派返回{status: delegated, ...}并留痕。每次执行/委派都会生成一条InteractionRecord含from_did、to_did、时间戳、成败、事件类型与元数据get_trust_report()L199可输出该 Agent 的当前分、阈值、成功/失败计数与完整交互历史直接可作为合规审计的原始材料。InMemoryTrustStore内置速率限制防刷分默认的内存信任存储 InMemoryTrustStore 有两个工程细节成功交互 5 分、失败交互 -10 分分数被钳制在 0–1000 区间set_trust_score中的max(0, min(1000, score))源码注释标注了 V33 修复MAX_UPDATES_PER_MINUTE 10即每个 DID 每分钟最多接受 10 次分数更新超出的更新被静默丢弃防止通过快速成功刷量人为抬高信任分。生产环境可替换为符合TrustStore协议L24要求实现get_trust_score与record_interaction两个方法的持久化后端。TrustAwareCrewkickoff 前的全员信任闸crew.py 中的TrustAwareCrew包装Crewverify_crew_trust()L58逐个检查成员信任分返回每个 DID 的score/threshold/trusted及汇总字段all_trustedkickoff()L80先做全员验证任一成员不达标即抛出TrustViolationError并列出全部不可信 DID全部达标后再透传 kwargs 构造真实 CrewAICrew并执行最终返回{trust_report: ..., result: ...}结构把信任验证报告与业务结果一并交还给调用方。这套先验证、后启动的行为有完整测试覆盖tests/test_crewai_integration.py 中的TestTrustAwareAgentVerifyPeer、TestTrustAwareAgentDelegation、TestTrustAwareAgentExecute、TestTrustAwareCrewVerify、TestTrustAwareCrewKickoff分别验证了阈值边界恰好等于阈值视为可信、低信任对端被拒委派并记录失败交互、以及未安装 crewai 时仍可正常导入并使用纯信任 stub的降级路径TestCrewAINotInstalled。独立能力层crewai-agentmesh 包的信任门选员与能力闸除上述集成模块外仓库还提供一个独立组件包 agentmesh-integrations/crewai-agentmesh聚焦任务分派前的选员与能力匹配核心实现在 crewai_agentmesh/trust.pyAgentProfileL20携带did、name、capabilities、trust_score0–1000默认 500、role、statusactive / suspended / revoked与metadata的身份档案提供has_capability/has_all_capabilities/has_any_capability查询CapabilityGateL73任务分派前的能力校验门。默认require_allTrue要求 Agent 具备任务所需的全部能力也可放宽为任一能力匹配Agent 状态非 active 时直接拒绝并返回原因如 Agent X is suspendedTrustTrackerL107跨 crew 运行周期维护信任分默认成功 10success_reward、失败 -50failure_penalty分数钳制在 0–1000并保留含新旧分数、任务描述与时间戳的完整变更历史TrustedCrewL166信任门控的选员器min_trust_score默认 100。select_for_task(required_capabilities, min_trust)返回同时满足active 信任分达标 能力匹配的候选并按信任分降序排列assign_task()返回结构化的TaskAssignmenttrust_sufficient、capability_match、allowed、reasonrecord_task_result()回写成功/失败到 TrustTrackerget_stats()汇总总 Agent 数、active 数、可信数以及被允许/被拒绝的分派计数。该包的 Quick Start 与测试用例可参考 README 和 tests/test_crewai_trust.pyfrom crewai_agentmesh import TrustedCrew, AgentProfile # Define trusted agents agents [ AgentProfile(diddid:mesh:researcher, nameResearcher, capabilities[research, analysis], trust_score800), AgentProfile(diddid:mesh:writer, nameWriter, capabilities[writing, editing], trust_score700), ] # Create trust-gated crew crew TrustedCrew(agentsagents, min_trust_score500) # Select agents for a task selected crew.select_for_task(required_capabilities[research]) assert len(selected) 1 assert selected[0].name Researcher版本提示从包入口 crewai_agentmesh/__init__.py 的显式DeprecationWarning可知该独立包已标记为弃用官方建议迁移到agent-governance-toolkit-integrations[crewai]extras。新项目建议直接使用 agent-mesh 内置的 integrations/crewai 模块AgentProfile/TrustedCrew这类能力门 选员思路仍可参考。实战示例内容创作 Crew文档给出的端到端示例覆盖supervisor 初始化 → 三个专职 Agent 委派 → 顺序流水线 → 合规报告完整继承如下from crewai import Agent, Task, Crew, Process from agentmesh import AgentIdentity, ScopeChain, PolicyEngine, AuditLog # Initialize AgentMesh supervisor AgentIdentity.create( namecontent-crew-supervisor, sponsormarketingcompany.com, capabilities[research, writing, seo, social_media] ) scope_chain ScopeChain(rootsupervisor) policy_engine PolicyEngine.from_file(policies/content-crew.yaml) audit_log AuditLog(agent_idsupervisor.did) # Create specialized agents seo_specialist Agent( roleSEO Specialist, goalOptimize content for search engines, agentmesh_identityscope_chain.delegate( nameseo-specialist, capabilities[research, seo] ) ) content_writer Agent( roleContent Writer, goalWrite engaging, SEO-optimized content, agentmesh_identityscope_chain.delegate( namecontent-writer, capabilities[writing] ) ) social_media_manager Agent( roleSocial Media Manager, goalCreate social media posts, agentmesh_identityscope_chain.delegate( namesocial-media-manager, capabilities[social_media, writing] ) ) # Define workflow tasks [ Task( descriptionResearch keywords for AI agent governance, agentseo_specialist ), Task( descriptionWrite a 1000-word blog post about AI agent governance, agentcontent_writer ), Task( descriptionCreate 5 social media posts to promote the blog, agentsocial_media_manager ), ] # Create governed crew crew Crew( agents[seo_specialist, content_writer, social_media_manager], taskstasks, processProcess.sequential, # Sequential execution verboseTrue ) # Execute with governance result crew.kickoff() # Generate compliance report print(\n Governance Report ) print(fSupervisor: {supervisor.did}) print(fCrew size: {len(crew.agents)}) print(fTasks completed: {len(tasks)}) print(fAudit entries: {len(audit_log.entries)})注意该示例体现的治理要点Process.sequential保证任务按研究→写作→社媒的顺序交接每一次交接都是信任握手与策略检查的天然插桩点AuditLog(agent_idsupervisor.did)以 supervisor DID 为锚点记录整条链路的审计条目supervisor.did本身即为可验证的加密身份。策略示例品牌安全与 API 限流文档提供了两个 crew 级策略 YAML分别覆盖输出内容风险审批与外部调用限流两类典型场景防止品牌风险——输出命中敏感主题时不直接放行而是转入人工审批流policies: - name: brand-safety rules: - condition: output contains controversial_topic action: require_approval approvers: [legalcompany.com]限制外部 API 调用速率——超出配额直接阻断policies: - name: api-rate-limit rules: - condition: action api_call limit: 1000/day action: block这两份策略文件分别落位到前面PolicyEngine.from_file(policies/crew.yaml)与from_file(policies/content-crew.yaml)加载的路径即可对 crew 的任务执行生效。最佳实践与生产就绪性文档最后给出的五条最佳实践与仓库实现一一呼应用范围链表达 crew 层级——委派即授权边界子 Agent 能力只能是父集的子集为专职 Agent 缩窄能力——示例中 SEO 专员拿不到social_media写手拿不到seo越权在签发时即被拒绝对协作启用信任握手——对应TrustHandshake.verify()或TrustAwareAgent.verify_peer()阈值建议 ≥700跨 crew 监控信任分——结合RewardEngine/TrustTracker的奖惩与衰减机制让长期低质协作的 Agent 自动降级审计每一次任务完成——AuditLog/InteractionRecord保留 DID、时间戳、成败与事件类型可直接导出为合规报告。关于生产就绪性原文档的结论是Production Ready: Yes, with monitoring and proper secret management.在生产环境需配套监控与密钥管理。此外仓库侧的 tests/test_crewai_integration.py 已经验证了信任阈值边界、委派拦截、交互留痕与无 crewai 环境的降级行为可作为集成改造后的回归基线。关键文件索引内容路径本文对应的集成文档agent-governance-python/agent-mesh/examples/integrations/crewai.mdTrustAwareAgent / InMemoryTrustStoreagent-governance-python/agent-mesh/src/agentmesh/integrations/crewai/agent.pyTrustAwareCrewagent-governance-python/agent-mesh/src/agentmesh/integrations/crewai/crew.py范围链与委派约束agent-governance-python/agent-mesh/src/agentmesh/identity/delegation.py信任握手agent-governance-python/agent-mesh/src/agentmesh/trust/handshake.py奖励引擎与信任衰减agent-governance-python/agent-mesh/src/agentmesh/reward/策略引擎 / 审计日志governance/policy.py、governance/audit.py集成测试agent-governance-python/agent-mesh/tests/test_crewai_integration.py独立能力层包已弃用agent-governance-python/agentmesh-integrations/crewai-agentmesh/【免费下载链接】agent-governance-toolkitAI Agent Governance Toolkit — Policy enforcement, zero-trust identity, execution sandboxing, and reliability engineering for autonomous AI agents. Covers 10/10 OWASP Agentic Top 10.项目地址: https://gitcode.com/GitHub_Trending/ag/agent-governance-toolkit创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 用 AgentMesh 治理 CrewAI：为多智能体 Crew 构建零信任身份、范围链委派与信任评分
用 AgentMesh 治理 CrewAI为多智能体 Crew 构建零信任身份、范围链委派与信任评分【免费下载链接】agent-governance-toolkitAI Agent Governance Toolkit — Policy enforcement, zero-trust identity, execution sandboxing, and reliability engineering for autonomous AI agents. Covers 10/10 OWASP Agentic Top 10.项目地址: https://gitcode.com/GitHub_Trending/ag/agent-governance-toolkitCrewAI 提供了协作式多智能体工作流Crew / Agent / Task但多个 LLM Agent 协同执行任务时谁来验证这个 Agent 是谁、它被授权做什么、它与同伴的信任关系是否足够是生产部署的核心难题。Agent Governance Toolkit 中的 agent-mesh 包针对这一问题将 AgentMesh 的加密身份DID、可缩窄的范围链委派ScopeChain、信任握手TrustHandshake与协作信任评分RewardEngine接入 CrewAI 的 Crew 执行模型。读完本文你将掌握如何为 CrewAI 的 crew 成员签发可审计身份、以最小权限委派任务、在任务交接前执行信任校验并理解仓库中真正落地的TrustAwareAgent/TrustAwareCrew包装器的实现细节与测试验证方式。为什么要把 AgentMesh 接入 CrewAICrewAI 本身专注于多 Agent 协作定义若干 Agent 与 Task组建 Crew 后通过kickoff()顺序或层级执行。但协作框架并不内置安全语义。按 CrewAI 集成文档 的描述AgentMesh 在此基础上补齐了四类治理能力加密身份每个 crew 成员持有独立的AgentIdentityDID范围链Scope Chains委派链路中子 Agent 的能力只能缩窄、不能扩张跨 Agent 信任握手任务交接前先验证对端 DID 的信任分是否达标协作信任评分根据任务完成质量、协作表现与策略合规情况持续更新每个 Agent 的 0–1000 信任分。这正好对应仓库源码中 agent-mesh 的核心模块AgentIdentity定义于 identity/agent_id.pyScopeChain定义于 identity/delegation.py策略与审计分别位于 governance/policy.pyPolicyEngine与 governance/audit.pyAuditLog它们都通过顶层包 agentmesh/__init__.py 统一导出因此集成代码可以直接from agentmesh import ...。快速开始给 Crew 成员签发身份并委派能力安装pip install agentmesh-platform crewai crewai-tools基本集成文档给出的完整示例展示了supervisor 签发身份 → 逐级委派缩窄能力 → 挂到 CrewAI Agent的标准流程from crewai import Agent, Task, Crew from agentmesh import AgentIdentity, ScopeChain, PolicyEngine # Create supervisor identity supervisor_identity AgentIdentity.create( namecrew-supervisor, sponsorteam-leadcompany.com, capabilities[research, writing, review] ) # Create scope chain scope_chain ScopeChain(rootsupervisor_identity) # Delegate to crew members researcher_identity scope_chain.delegate( nameresearcher-agent, capabilities[research] # Narrowed from supervisor ) writer_identity scope_chain.delegate( namewriter-agent, capabilities[writing] # Narrowed from supervisor ) reviewer_identity scope_chain.delegate( namereviewer-agent, capabilities[review] # Narrowed from supervisor ) # Create CrewAI agents with AgentMesh identities researcher Agent( roleResearcher, goalResearch the topic thoroughly, backstoryExpert researcher with 10 years experience, agentmesh_identityresearcher_identity # Attach identity ) writer Agent( roleWriter, goalWrite engaging content, backstoryProfessional content writer, agentmesh_identitywriter_identity ) reviewer Agent( roleReviewer, goalReview and improve content, backstorySenior editor with high standards, agentmesh_identityreviewer_identity ) # Define tasks research_task Task( descriptionResearch the topic: AgentMesh governance for AI agents, agentresearcher ) writing_task Task( descriptionWrite a blog post based on the research, agentwriter ) review_task Task( descriptionReview and improve the blog post, agentreviewer ) # Create governed crew crew Crew( agents[researcher, writer, reviewer], tasks[research_task, writing_task, review_task], verboseTrue ) # Run with governance result crew.kickoff() print(fResult: {result}) print(fSupervisor DID: {supervisor_identity.did}) print(fCrew members: {len(scope_chain.links)})需要说明两点前提示例代码表达的是集成模式为每个 CrewAI Agent 关联一个 AgentMesh 身份对象。仓库中实际随包发布的 CrewAI 集成采用的是包装器/混入模式见下文仓库中的落地实现一节crewai是可选依赖即使不安装 CrewAI 也能运行纯信任操作能力缩窄是硬性约束。从源码结构看delegation.py 中DelegationLink的注释明确写着每个链接代表父级向子级授予能力子级的能力必须是父级的子集The childs capabilities MUST be a subset of the parents。因此示例中researcher只能拿到[research]而不能越权申请 supervisor 之外的能力委派深度也受DEFAULT_DELEGATION_MAX_DEPTH常量约束超出会抛出DelegationDepthError。进阶特性一crew 成员之间的信任握手在 writer 接受 researcher 交接的任务之前先用TrustHandshake校验对端 DID 的信任分。文档给出的模式如下from agentmesh import TrustHandshake # Before writer accepts work from researcher async def governed_task_handoff(from_agent, to_agent, task): handshake TrustHandshake() # Verify peer result await handshake.verify( peer_didfrom_agent.agentmesh_identity.did, required_score700 ) if not result.verified: raise SecurityError(fUntrusted peer: {result.reason}) # Accept task return to_agent.execute(task)TrustHandshake在源码中位于 trust/handshake.py由顶层包导出见 agentmesh/__init__.py。在 CrewAI 场景下它的语义是Crew 内部任一 Agent 在接收来自同伴的产出研究结论、初稿等之前必须确认对端信任分不低于required_score否则拒绝交接并抛出异常——这相当于把零信任原则落到了任务级数据流上。进阶特性二协作信任评分RewardEngine信任分不是静态配置而是随协作行为演化的。文档展示了如何用RewardEngine基于三个维度更新 crew 各成员的得分from agentmesh import RewardEngine reward_engine RewardEngine() # Update scores based on collaboration quality def update_crew_scores(crew): for agent in crew.agents: identity agent.agentmesh_identity # Score based on: # - Task completion quality # - Collaboration with other agents # - Policy compliance score reward_engine.update_score( agent_ididentity.did, dimensions{ task_quality: 0.9, collaboration: 0.85, policy_compliance: 1.0 } ) print(f{agent.role}: {score.total}/1000)RewardEngine的实现在 reward/engine.py同目录还提供trust_decay.py信任随时间衰减与distribution.py/distributor.py奖励分发因此可以推断生产环境中一个长期不活跃或长期违规的 Agent其信任分会因衰减与惩罚自然降到握手阈值之下从而被后续的信任校验自动隔离——无需人工下线。进阶特性三对 crew 任务执行策略强制在任务执行前插入PolicyEngine校验违规即拒绝执行并留痕policy_engine PolicyEngine.from_file(policies/crew.yaml) # Wrap task execution with policy checks def governed_task_execution(task, agent): # Check policy before execution result policy_engine.check( actionexecute_task, agentagent.agentmesh_identity.did, tasktask.description ) if not result.allowed: raise PermissionError(fPolicy violation: {result.reason}) # Execute task output agent.execute_task(task) # Audit audit_log.log(task_completed, agentagent.role, tasktask.description) return outputPolicyEngine定义于 governance/policy.py。该引擎是 agent-mesh 中多语言共享的策略求值核心与 policy-engine 目录中的 Rego/Cedar 后端配套check()返回allowed/reason配合 governance/audit.py 中的AuditLog即可实现先校验、再执行、后审计的闭环。仓库中的落地实现TrustAwareAgent 与 TrustAwareCrew上面三节对应文档中的集成模式仓库实际随agent-mesh包发布、可直接 import 的 CrewAI 集成在 integrations/crewai/ 下导出InMemoryTrustStore、InteractionRecord、TrustAwareAgent、TrustAwareCrew与TrustStore协议见 integrations/crewai/__init__.py。其设计要点值得逐条对照TrustAwareAgent包装器模式CrewAI 为可选依赖agent.py 中的TrustAwareAgent采用包装器/混入模式构造时接收agent_did如did:mesh:abc123、min_trust_score默认 500取值 0–1000、可选trust_store后端其余**kwargs透传给 CrewAI 的Agent构造函数。若环境中未安装 crewai则进入纯信任模式trust-only mode信任操作照常可用任务执行返回 stub 结果。核心方法有三个verify_peer(peer_did)L128查询对端信任分是否 ≥min_trust_score对应文档中信任握手的同步简化版execute_with_trust(task, context)L140先校验自身信任分不足则记录一次失败交互并抛出TrustViolationError执行成功/失败都会通过_record()写入交互历史并回调trust_store.record_interaction()更新得分delegate_with_trust(task, peer_did)L175仅当对端信任分达标才允许委派返回{status: delegated, ...}并留痕。每次执行/委派都会生成一条InteractionRecord含from_did、to_did、时间戳、成败、事件类型与元数据get_trust_report()L199可输出该 Agent 的当前分、阈值、成功/失败计数与完整交互历史直接可作为合规审计的原始材料。InMemoryTrustStore内置速率限制防刷分默认的内存信任存储 InMemoryTrustStore 有两个工程细节成功交互 5 分、失败交互 -10 分分数被钳制在 0–1000 区间set_trust_score中的max(0, min(1000, score))源码注释标注了 V33 修复MAX_UPDATES_PER_MINUTE 10即每个 DID 每分钟最多接受 10 次分数更新超出的更新被静默丢弃防止通过快速成功刷量人为抬高信任分。生产环境可替换为符合TrustStore协议L24要求实现get_trust_score与record_interaction两个方法的持久化后端。TrustAwareCrewkickoff 前的全员信任闸crew.py 中的TrustAwareCrew包装Crewverify_crew_trust()L58逐个检查成员信任分返回每个 DID 的score/threshold/trusted及汇总字段all_trustedkickoff()L80先做全员验证任一成员不达标即抛出TrustViolationError并列出全部不可信 DID全部达标后再透传 kwargs 构造真实 CrewAICrew并执行最终返回{trust_report: ..., result: ...}结构把信任验证报告与业务结果一并交还给调用方。这套先验证、后启动的行为有完整测试覆盖tests/test_crewai_integration.py 中的TestTrustAwareAgentVerifyPeer、TestTrustAwareAgentDelegation、TestTrustAwareAgentExecute、TestTrustAwareCrewVerify、TestTrustAwareCrewKickoff分别验证了阈值边界恰好等于阈值视为可信、低信任对端被拒委派并记录失败交互、以及未安装 crewai 时仍可正常导入并使用纯信任 stub的降级路径TestCrewAINotInstalled。独立能力层crewai-agentmesh 包的信任门选员与能力闸除上述集成模块外仓库还提供一个独立组件包 agentmesh-integrations/crewai-agentmesh聚焦任务分派前的选员与能力匹配核心实现在 crewai_agentmesh/trust.pyAgentProfileL20携带did、name、capabilities、trust_score0–1000默认 500、role、statusactive / suspended / revoked与metadata的身份档案提供has_capability/has_all_capabilities/has_any_capability查询CapabilityGateL73任务分派前的能力校验门。默认require_allTrue要求 Agent 具备任务所需的全部能力也可放宽为任一能力匹配Agent 状态非 active 时直接拒绝并返回原因如 Agent X is suspendedTrustTrackerL107跨 crew 运行周期维护信任分默认成功 10success_reward、失败 -50failure_penalty分数钳制在 0–1000并保留含新旧分数、任务描述与时间戳的完整变更历史TrustedCrewL166信任门控的选员器min_trust_score默认 100。select_for_task(required_capabilities, min_trust)返回同时满足active 信任分达标 能力匹配的候选并按信任分降序排列assign_task()返回结构化的TaskAssignmenttrust_sufficient、capability_match、allowed、reasonrecord_task_result()回写成功/失败到 TrustTrackerget_stats()汇总总 Agent 数、active 数、可信数以及被允许/被拒绝的分派计数。该包的 Quick Start 与测试用例可参考 README 和 tests/test_crewai_trust.pyfrom crewai_agentmesh import TrustedCrew, AgentProfile # Define trusted agents agents [ AgentProfile(diddid:mesh:researcher, nameResearcher, capabilities[research, analysis], trust_score800), AgentProfile(diddid:mesh:writer, nameWriter, capabilities[writing, editing], trust_score700), ] # Create trust-gated crew crew TrustedCrew(agentsagents, min_trust_score500) # Select agents for a task selected crew.select_for_task(required_capabilities[research]) assert len(selected) 1 assert selected[0].name Researcher版本提示从包入口 crewai_agentmesh/__init__.py 的显式DeprecationWarning可知该独立包已标记为弃用官方建议迁移到agent-governance-toolkit-integrations[crewai]extras。新项目建议直接使用 agent-mesh 内置的 integrations/crewai 模块AgentProfile/TrustedCrew这类能力门 选员思路仍可参考。实战示例内容创作 Crew文档给出的端到端示例覆盖supervisor 初始化 → 三个专职 Agent 委派 → 顺序流水线 → 合规报告完整继承如下from crewai import Agent, Task, Crew, Process from agentmesh import AgentIdentity, ScopeChain, PolicyEngine, AuditLog # Initialize AgentMesh supervisor AgentIdentity.create( namecontent-crew-supervisor, sponsormarketingcompany.com, capabilities[research, writing, seo, social_media] ) scope_chain ScopeChain(rootsupervisor) policy_engine PolicyEngine.from_file(policies/content-crew.yaml) audit_log AuditLog(agent_idsupervisor.did) # Create specialized agents seo_specialist Agent( roleSEO Specialist, goalOptimize content for search engines, agentmesh_identityscope_chain.delegate( nameseo-specialist, capabilities[research, seo] ) ) content_writer Agent( roleContent Writer, goalWrite engaging, SEO-optimized content, agentmesh_identityscope_chain.delegate( namecontent-writer, capabilities[writing] ) ) social_media_manager Agent( roleSocial Media Manager, goalCreate social media posts, agentmesh_identityscope_chain.delegate( namesocial-media-manager, capabilities[social_media, writing] ) ) # Define workflow tasks [ Task( descriptionResearch keywords for AI agent governance, agentseo_specialist ), Task( descriptionWrite a 1000-word blog post about AI agent governance, agentcontent_writer ), Task( descriptionCreate 5 social media posts to promote the blog, agentsocial_media_manager ), ] # Create governed crew crew Crew( agents[seo_specialist, content_writer, social_media_manager], taskstasks, processProcess.sequential, # Sequential execution verboseTrue ) # Execute with governance result crew.kickoff() # Generate compliance report print(\n Governance Report ) print(fSupervisor: {supervisor.did}) print(fCrew size: {len(crew.agents)}) print(fTasks completed: {len(tasks)}) print(fAudit entries: {len(audit_log.entries)})注意该示例体现的治理要点Process.sequential保证任务按研究→写作→社媒的顺序交接每一次交接都是信任握手与策略检查的天然插桩点AuditLog(agent_idsupervisor.did)以 supervisor DID 为锚点记录整条链路的审计条目supervisor.did本身即为可验证的加密身份。策略示例品牌安全与 API 限流文档提供了两个 crew 级策略 YAML分别覆盖输出内容风险审批与外部调用限流两类典型场景防止品牌风险——输出命中敏感主题时不直接放行而是转入人工审批流policies: - name: brand-safety rules: - condition: output contains controversial_topic action: require_approval approvers: [legalcompany.com]限制外部 API 调用速率——超出配额直接阻断policies: - name: api-rate-limit rules: - condition: action api_call limit: 1000/day action: block这两份策略文件分别落位到前面PolicyEngine.from_file(policies/crew.yaml)与from_file(policies/content-crew.yaml)加载的路径即可对 crew 的任务执行生效。最佳实践与生产就绪性文档最后给出的五条最佳实践与仓库实现一一呼应用范围链表达 crew 层级——委派即授权边界子 Agent 能力只能是父集的子集为专职 Agent 缩窄能力——示例中 SEO 专员拿不到social_media写手拿不到seo越权在签发时即被拒绝对协作启用信任握手——对应TrustHandshake.verify()或TrustAwareAgent.verify_peer()阈值建议 ≥700跨 crew 监控信任分——结合RewardEngine/TrustTracker的奖惩与衰减机制让长期低质协作的 Agent 自动降级审计每一次任务完成——AuditLog/InteractionRecord保留 DID、时间戳、成败与事件类型可直接导出为合规报告。关于生产就绪性原文档的结论是Production Ready: Yes, with monitoring and proper secret management.在生产环境需配套监控与密钥管理。此外仓库侧的 tests/test_crewai_integration.py 已经验证了信任阈值边界、委派拦截、交互留痕与无 crewai 环境的降级行为可作为集成改造后的回归基线。关键文件索引内容路径本文对应的集成文档agent-governance-python/agent-mesh/examples/integrations/crewai.mdTrustAwareAgent / InMemoryTrustStoreagent-governance-python/agent-mesh/src/agentmesh/integrations/crewai/agent.pyTrustAwareCrewagent-governance-python/agent-mesh/src/agentmesh/integrations/crewai/crew.py范围链与委派约束agent-governance-python/agent-mesh/src/agentmesh/identity/delegation.py信任握手agent-governance-python/agent-mesh/src/agentmesh/trust/handshake.py奖励引擎与信任衰减agent-governance-python/agent-mesh/src/agentmesh/reward/策略引擎 / 审计日志governance/policy.py、governance/audit.py集成测试agent-governance-python/agent-mesh/tests/test_crewai_integration.py独立能力层包已弃用agent-governance-python/agentmesh-integrations/crewai-agentmesh/【免费下载链接】agent-governance-toolkitAI Agent Governance Toolkit — Policy enforcement, zero-trust identity, execution sandboxing, and reliability engineering for autonomous AI agents. Covers 10/10 OWASP Agentic Top 10.项目地址: https://gitcode.com/GitHub_Trending/ag/agent-governance-toolkit创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考