从标准到代码:GB/Z 185合规的Agent Loop设计(附Python实现+合规Checklist)

一句话总结: GB/Z 185的7个核心条款不是"合规负担",而是Agent Loop的工程化需求规格——身份标识对应多租户、安全认证对应权限管理、可审计性对应运维监控、错误恢复对应系统稳定性,逐条映射为可运行的Python代码。

适合谁: 正在企业落地AI Agent、需要通过国标合规审查的开发者、架构师,以及把GB/Z 185视为"额外负担"想理解其工程价值的工程师。

验证环境: Python 3.10+, 本文代码基于GB/Z 185《人工智能 智能体互联》系列标准(2026年5月22日发布),兼容LangChain 0.3.x核心概念。



📋 阅读导航:本文约20分钟阅读。建议先浏览「四、合规Checklist」直接对照你的项目打勾,再按需深入具体条款的代码实现。如果只想看结论,跳到「六、总结」。


一、GB/Z 185中,哪些条款直接影响Agent Loop?

我在《GB/Z 185《人工智能 智能体互联》系列标准核心内容梳理》中详细解读过标准全文。但标准本身不直接告诉你"代码怎么写"——它只说"应该怎么做"。本文聚焦与Loop Engineering直接相关的7个条款,每条用一句话解释"对代码设计意味着什么":

标准分册 标准要求 对Agent Loop意味着什么 代码影响点
第2部分:身份码 每个智能体必须有唯一身份码,支持全生命周期追溯 Loop的每次运行必须绑定Agent身份 State中增加agent_id/tenant_id字段
第7部分:工具调用 定义工具描述标准、调用流程、数据格式 Loop中工具调用的输入输出必须遵循标准格式 Tool定义必须包含标准Schema校验
第3部分:身份管理 身份注册核验、凭证管理、双向身份鉴别 每次Tool调用前必须验证权限 在Verification Loop中加入SecurityGrader
第6部分:智能体交互 交互内容元素规范、消息结构、安全要求 记忆的读写必须按敏感度分级 MemoryItem增加sensitivity_level字段
第1部分:总体架构 功能参考架构、安全与合规要素、日志防篡改 每个决策步骤必须留下审计记录 Loop中集成AuditLogger模块
第6部分:智能体交互 交互容错机制、任务状态管理 异常时必须能回滚到安全状态 增加Checkpoint+Rollback机制
第6部分:智能体交互 点对点/群组/混合模式、协作方式 Multi-Agent场景下Loop必须支持协作模式 增加mode参数(独立/协作)

核心洞察:GB/Z 185不是"额外负担",而是Agent Loop的"需求规格说明书"——标准中的每一条要求,都对应代码中的一个具体设计决策。


二、标准条款 → 代码实现:逐条拆解

下面按照标准条款顺序,逐条给出Python实现。代码可以直接复制到项目中使用。

2.1 第2部分:身份码 → 合规Agent State

GB/Z 185.2—2026要求:每个智能体必须有唯一身份码,支持全生命周期追溯和多租户隔离。这意味着你的Loop State不能只存messagestool_calls,必须带上身份和租户信息。

from typing import TypedDict, Annotated, List
from dataclasses import dataclass, field
from datetime import datetime
import uuid
import operator


class CompliantAgentState(TypedDict):
    """
    符合GB/Z 185标准的Agent状态定义。
    
    标准条款:5.2 智能体身份标识
    - agent_id: 智能体唯一标识(UUID格式)
    - tenant_id: 租户ID,支持多租户隔离
    - session_id: 会话唯一标识,用于审计追踪
    """
    messages: Annotated[list, operator.add]
    tool_calls: list
    iteration: int
    is_complete: bool
    # ===== GB/Z 185 5.2 新增字段 =====
    agent_id: str       # 智能体唯一标识
    tenant_id: str      # 租户隔离
    session_id: str     # 会话追踪
    security_level: str  # 数据安全等级(public/internal/confidential)


class CompliantAgentLoop:
    """
    符合GB/Z 185的Agent Loop实现。
    
    标准条款覆盖:5.2(身份标识)+ 7.1(可审计)+ 7.2(错误恢复)
    """
    
    def __init__(self, llm, tools, agent_id=None, tenant_id="default", max_iterations=10):
        """初始化符合GB/Z 185标准的Agent Loop。
        
        参数说明:
        - llm: 语言模型实例,需支持 invoke(messages) 接口
        - tools: 工具列表,每个工具需有 name 和 invoke(args) 方法
        - agent_id: 智能体唯一标识(默认自动生成UUID,生产建议显式传入便于追踪)
        - tenant_id: 租户ID,支持多租户隔离(默认"default",生产必须按租户分配)
        - max_iterations: 最大循环轮次(默认10,生产建议3-5)
        
        返回:无
        """
        self.llm = llm
        self.tools = {tool.name: tool for tool in tools}
        self.max_iterations = max_iterations
        # ===== 标准5.2:身份标识 =====
        self.agent_id = agent_id or str(uuid.uuid4())
        self.tenant_id = tenant_id
        self.session_id = str(uuid.uuid4())
    
    def create_initial_state(self, user_input: str) -> CompliantAgentState:
        """创建符合标准的初始状态。"""
        return CompliantAgentState(
            messages=[{"role": "user", "content": user_input}],
            tool_calls=[],
            iteration=0,
            is_complete=False,
            # 标准5.2字段
            agent_id=self.agent_id,
            tenant_id=self.tenant_id,
            session_id=self.session_id,
            security_level="public",  # 默认公开级别
        )

关键设计agent_idtenant_id在State初始化时就注入,确保后续所有操作(工具调用、日志记录)都能追溯到具体智能体和租户。这是多租户SaaS场景的基础要求。


2.2 第7部分:工具调用 → 标准消息格式校验

GB/Z 185.7—2026要求:工具描述、调用流程、数据格式必须遵循标准规范。对应到Loop中,就是Tool的输入输出必须遵循标准Schema。

from typing import Dict, Any
import jsonschema


class StandardTool:
    """
    符合GB/Z 185通信协议规范的工具定义。
    
    标准条款:5.3 通信协议规范
    - 每个Tool必须有标准Schema定义
    - 输入输出必须通过Schema校验
    """
    
    def __init__(self, name: str, description: str, input_schema: dict, output_schema: dict):
        self.name = name
        self.description = description
        self.input_schema = input_schema   # 标准输入格式
        self.output_schema = output_schema  # 标准输出格式
    
    def validate_input(self, params: dict) -> bool:
        """校验输入参数是否符合标准Schema。"""
        try:
            jsonschema.validate(instance=params, schema=self.input_schema)
            return True
        except jsonschema.ValidationError as e:
            print(f"❌ 输入参数不符合GB/Z 185标准格式: {e.message}")
            return False
    
    def validate_output(self, result: Any) -> bool:
        """校验输出结果是否符合标准Schema。"""
        try:
            jsonschema.validate(instance=result, schema=self.output_schema)
            return True
        except jsonschema.ValidationError as e:
            print(f"❌ 输出结果不符合GB/Z 185标准格式: {e.message}")
            return False


# ========== 使用示例:定义一个符合标准的天气查询工具 ==========
weather_tool = StandardTool(
    name="get_weather",
    description="查询指定城市的天气",
    input_schema={
        "type": "object",
        "properties": {
            "city": {"type": "string", "description": "城市名称"},
            "date": {"type": "string", "format": "date"}
        },
        "required": ["city"]
    },
    output_schema={
        "type": "object",
        "properties": {
            "temperature": {"type": "number"},
            "condition": {"type": "string"},
            "source": {"type": "string"}  # 数据来源(审计需要)
        },
        "required": ["temperature", "condition"]
    }
)

关键设计StandardTool强制要求每个工具定义输入Schema和输出Schema。任何不符合标准的参数或结果,都会在运行时被拦截。这是防止"脏数据"流入Agent循环的第一道防线。


2.3 第3部分:身份管理 → Verification Loop中的SecurityGrader

GB/Z 185.3—2026要求:身份注册核验、凭证管理、双向身份鉴别。对应到Loop中,就是在Verification Loop(质量验证层)增加一个安全评分器

from dataclasses import dataclass
from typing import Callable, List


@dataclass
class SecurityCheckResult:
    """安全检查结果。"""
    passed: bool
    violations: List[str]  # 违规项列表
    risk_level: str        # low/medium/high


class SecurityGrader:
    """
    符合GB/Z 185 6.1条款的安全认证评分器。
    
    在Verification Loop中,每次Tool调用前执行权限检查。
    """
    
    def __init__(self, agent_permissions: dict):
        """初始化安全认证评分器(GB/Z 185 6.1)。
        
        参数说明:
        - agent_permissions: 权限策略字典,格式 {tool_name: allowed_params}
          示例:{"read_file": {"allowed_paths": ["/data/public/*"], "allowed_ops": ["read"]}}
          - tool_name: 工具名称,与 StandardTool.name 对应
          - allowed_params: 该工具允许的参数限制,支持以下字段:
            * allowed_paths: 文件路径白名单列表(支持通配符 *)
            * allowed_ops: 允许的操作类型列表(如 ["read", "SELECT"])
        
        生产建议:权限配置建议存储在数据库或配置中心,支持热更新,不要硬编码在代码中
        """
        self.agent_permissions = agent_permissions
    
    def check(self, tool_name: str, params: dict, agent_id: str) -> SecurityCheckResult:
        """检查本次Tool调用是否合规。"""
        violations = []
        
        # 检查1:Agent是否有权限调用该工具
        if tool_name not in self.agent_permissions:
            violations.append(f"Agent '{agent_id}' 无权调用工具 '{tool_name}'")
        
        # 检查2:参数是否在允许范围内
        if tool_name in self.agent_permissions:
            allowed = self.agent_permissions[tool_name]
            
            # 检查文件路径白名单
            if "file_path" in params and "allowed_paths" in allowed:
                path = params["file_path"]
                if not any(path.startswith(p.rstrip("*")) for p in allowed["allowed_paths"]):
                    violations.append(f"路径 '{path}' 不在允许范围内")
            
            # 检查操作类型(只读 vs 读写)
            if "operation" in params and "allowed_ops" in allowed:
                if params["operation"] not in allowed["allowed_ops"]:
                    violations.append(f"操作 '{params['operation']}' 不被允许")
        
        # 检查3:敏感数据访问
        if any(k in str(params).lower() for k in ["password", "secret", "token", "key"]):
            violations.append("检测到敏感数据访问请求,需要二次确认")
        
        passed = len(violations) == 0
        risk_level = "high" if len(violations) > 1 else "medium" if violations else "low"
        
        return SecurityCheckResult(passed, violations, risk_level)


# ========== 使用示例 ==========
if __name__ == "__main__":
    # 定义Agent权限策略
    permissions = {
        "read_file": {
            "allowed_paths": ["/data/public/*", "/tmp/*"],
            "allowed_ops": ["read"]
        },
        "query_db": {
            "allowed_ops": ["SELECT"]
        },
        "send_email": {
            "allowed_ops": ["draft"],  # 只能存草稿,不能发送
        }
    }
    
    grader = SecurityGrader(permissions)
    
    # 场景1:合规调用
    result = grader.check("read_file", {"file_path": "/data/public/report.txt"}, "agent_001")
    print(f"合规检查: {'✅通过' if result.passed else '❌失败'}, 风险等级: {result.risk_level}")
    
    # 场景2:越权调用
    result = grader.check("read_file", {"file_path": "/etc/passwd"}, "agent_001")
    print(f"合规检查: {'✅通过' if result.passed else '❌失败'}, 违规: {result.violations}")
    
    # 场景3:敏感操作
    result = grader.check("send_email", {"to": "admin@company.com", "body": "password=123456"}, "agent_001")
    print(f"合规检查: {'✅通过' if result.passed else '❌失败'}, 违规: {result.violations}")

关键设计SecurityGraderVerification Loop的每次Tool调用前执行。它不依赖LLM的判断,而是基于明确的规则(白名单、操作类型、敏感词检测),确保权限检查是确定性的——不会因为LLM的"幻觉"而误放行。


2.4 第6部分:智能体交互 → 敏感数据分级存储

GB/Z 185.6—2026要求:交互内容元素规范、消息结构、安全要求。在Agent Loop中,这意味着记忆(Memory)的读写必须按敏感度分级

from enum import Enum

class DataSensitivity(Enum):
    """数据敏感度分级(GB/Z 185 6.2)。"""
    PUBLIC = "public"           # 公开数据
    INTERNAL = "internal"      # 内部数据
    CONFIDENTIAL = "confidential"  # 机密数据
    RESTRICTED = "restricted"    # 受限数据


class CompliantMemoryItem:
    """
    符合GB/Z 185 6.2的数据隐私保护要求的记忆项。
    """
    
    def __init__(self, content: str, memory_type: str, sensitivity: DataSensitivity, metadata: dict = None):
        self.content = content
        self.memory_type = memory_type
        self.sensitivity = sensitivity
        self.metadata = metadata or {}
        self.created_at = datetime.now().isoformat()
        self.access_count = 0
    
    def can_access(self, requester_security_level: str) -> bool:
        """
        检查请求者是否有权限访问本记忆项。
        
        权限矩阵:
        - public: 任何人可访问
        - internal: 需认证用户
        - confidential: 需特定角色
        - restricted: 需显式授权
        """
        access_matrix = {
            DataSensitivity.PUBLIC: ["public", "internal", "confidential", "restricted"],
            DataSensitivity.INTERNAL: ["internal", "confidential", "restricted"],
            DataSensitivity.CONFIDENTIAL: ["confidential", "restricted"],
            DataSensitivity.RESTRICTED: ["restricted"],
        }
        return requester_security_level in access_matrix.get(self.sensitivity, [])
    
    def sanitize_for_output(self, target_level: str) -> str:
        """
        根据目标安全级别,对记忆内容进行脱敏。
        
        例如:将机密数据输出给public级别时,自动脱敏。
        """
        if self.sensitivity == DataSensitivity.CONFIDENTIAL and target_level == "public":
            return "[机密信息,已脱敏]"
        return self.content


# ========== 使用示例 ==========
if __name__ == "__main__":
    # 创建一个机密记忆
    secret = CompliantMemoryItem(
        content="公司Q3营收1.2亿元,利润率15%",
        memory_type="financial",
        sensitivity=DataSensitivity.CONFIDENTIAL
    )
    
    # 场景1:内部用户访问 → 允许
    print(f"内部用户访问: {secret.can_access('internal')}")  # True
    
    # 场景2:公开输出时自动脱敏
    print(f"公开输出: {secret.sanitize_for_output('public')}")  # [机密信息,已脱敏]
    
    # 场景3:内部用户查看 → 完整内容
    print(f"内部输出: {secret.sanitize_for_output('internal')}")  # 公司Q3营收1.2亿元...

关键设计DataSensitivity分级 + can_access()权限检查 + sanitize_for_output()自动脱敏,三层防护确保数据隐私。这在企业客服Agent、内部助手等场景中至关重要。


2.5 第1部分:总体架构 → 全链路审计日志

GB/Z 185.1—2026要求:功能参考架构中的安全与合规要素、日志防篡改。对应到Loop中,就是在每个关键步骤记录审计日志。

import json
from datetime import datetime

class AuditLogger:
    """
    符合GB/Z 185 7.1可审计性要求的审计日志模块。
    
    记录内容:
    - 决策日志:每次LLM调用的prompt、response、模型版本
    - 工具日志:每次Tool调用的参数、返回、执行耗时
    - 状态日志:State的每次变更
    """
    
    def __init__(self, agent_id: str, session_id: str, log_file: str = "agent_audit.log"):
        """初始化审计日志模块(GB/Z 185 7.1)。
        
        参数说明:
        - agent_id: 智能体唯一标识,与 CompliantAgentState.agent_id 对应,用于审计追踪
        - session_id: 会话唯一标识,与 CompliantAgentState.session_id 对应,支持按会话查询完整轨迹
        - log_file: 日志文件路径(默认"agent_audit.log",生产建议按日期分片如"audit/2026-08-09.log")
        
        生产注意:
        - 日志文件建议定期归档,单文件超过100MB时切换新文件
        - 敏感数据(如密码、token)在写入日志前必须脱敏
        - 生产环境建议接入ELK/Splunk等日志分析平台,便于合规审查时快速检索
        """
        self.agent_id = agent_id
        self.session_id = session_id
        self.log_file = log_file
    
    def log_decision(self, iteration: int, prompt: str, response: str, model_version: str):
        """记录决策日志。"""
        entry = {
            "timestamp": datetime.now().isoformat(),
            "agent_id": self.agent_id,
            "session_id": self.session_id,
            "type": "decision",
            "iteration": iteration,
            "model_version": model_version,
            "prompt_length": len(prompt),
            "response_length": len(response),
        }
        self._write(entry)
    
    def log_tool_call(self, iteration: int, tool_name: str, params: dict, result: Any, duration_ms: float):
        """记录工具调用日志。"""
        entry = {
            "timestamp": datetime.now().isoformat(),
            "agent_id": self.agent_id,
            "session_id": self.session_id,
            "type": "tool_call",
            "iteration": iteration,
            "tool_name": tool_name,
            "params": params,
            "result_preview": str(result)[:200],  # 只存前200字符,防止日志过大
            "duration_ms": duration_ms,
        }
        self._write(entry)
    
    def log_state_change(self, iteration: int, field: str, old_value: Any, new_value: Any):
        """记录状态变更日志。"""
        entry = {
            "timestamp": datetime.now().isoformat(),
            "agent_id": self.agent_id,
            "session_id": self.session_id,
            "type": "state_change",
            "iteration": iteration,
            "field": field,
        }
        self._write(entry)
    
    def _write(self, entry: dict):
        """写入日志文件。"""
        with open(self.log_file, "a", encoding="utf-8") as f:
            f.write(json.dumps(entry, ensure_ascii=False) + "\n")
    
    def query(self, session_id: str = None, tool_name: str = None) -> list:
        """查询审计日志(用于合规审查)。"""
        results = []
        with open(self.log_file, "r", encoding="utf-8") as f:
            for line in f:
                entry = json.loads(line.strip())
                if session_id and entry.get("session_id") != session_id:
                    continue
                if tool_name and entry.get("tool_name") != tool_name:
                    continue
                results.append(entry)
        return results


# ========== 使用示例 ==========
if __name__ == "__main__":
    logger = AuditLogger(agent_id="agent_001", session_id="sess_abc123")
    
    # 记录一次决策
    logger.log_decision(
        iteration=1,
        prompt="查询北京天气",
        response="调用get_weather工具",
        model_version="gpt-4o-2024-08-06"
    )
    
    # 记录一次工具调用
    logger.log_tool_call(
        iteration=1,
        tool_name="get_weather",
        params={"city": "北京"},
        result={"temperature": 28, "condition": "晴"},
        duration_ms=145.2
    )
    
    # 合规审查时查询
    logs = logger.query(session_id="sess_abc123")
    print(f"本次会话共记录 {len(logs)} 条审计日志")

关键设计AuditLogger在Loop的三个关键节点(决策、工具调用、状态变更)自动记录日志。日志文件可以直接用于合规审查——审查人员可以回放任意一次Agent运行的完整轨迹。


2.6 第6部分:智能体交互 → Checkpoint + Rollback

GB/Z 185.6—2026要求:交互容错机制、任务状态管理。对应到Loop中,就是Checkpoint机制——定期保存状态快照,异常时回滚。

import copy
import pickle

class CheckpointManager:
    """
    符合GB/Z 185 7.2错误恢复机制的Checkpoint管理器。
    
    功能:
    - 定期保存State快照
    - 异常时自动回滚到最近的安全状态
    - 支持手动回滚到任意历史版本
    """
    
    def __init__(self, max_checkpoints: int = 5):
        """初始化Checkpoint管理器(GB/Z 185 7.2)。
        
        参数说明:
        - max_checkpoints: 保留的最大快照数量(默认5,生产建议10-20)
          * 值过小:异常时回滚选择少,可能回滚到不够"安全"的状态
          * 值过大:内存占用增加,快照列表查询变慢
        
        生产注意:
        - Checkpoint只保存在内存中,进程重启会丢失——生产环境建议定期持久化到Redis/Postgres
        - 每个Checkpoint是State的深拷贝,频繁保存对大State有性能开销
        - 建议每3-5次迭代保存一次,不要每次迭代都保存
        """
        self.checkpoints = []  # State快照列表
        self.max_checkpoints = max_checkpoints
    
    def save(self, state: CompliantAgentState):
        """保存当前State快照。"""
        snapshot = copy.deepcopy(state)
        self.checkpoints.append({
            "timestamp": datetime.now().isoformat(),
            "iteration": state["iteration"],
            "state": snapshot,
        })
        
        # 只保留最近N个快照
        if len(self.checkpoints) > self.max_checkpoints:
            self.checkpoints.pop(0)
    
    def rollback(self, iteration: int = None) -> CompliantAgentState:
        """
        回滚到指定迭代或最近的安全状态。
        
        如果iteration为None,回滚到最近一个快照。
        """
        if not self.checkpoints:
            raise RuntimeError("没有可用的Checkpoint,无法回滚")
        
        if iteration is not None:
            for cp in reversed(self.checkpoints):
                if cp["iteration"] <= iteration:
                    return cp["state"]
        
        return self.checkpoints[-1]["state"]
    
    def list_checkpoints(self) -> list:
        """列出所有可用快照(用于人工干预)。"""
        return [
            {"iteration": cp["iteration"], "timestamp": cp["timestamp"]}
            for cp in self.checkpoints
        ]

关键设计CheckpointManager在每次迭代后保存State快照。当异常发生时(如工具调用超时、权限校验失败),可以回滚到最近的安全状态,而不是让Agent在错误状态上继续运行。


三、完整合规Agent Loop:把所有条款整合到一起

下面是整合了所有GB/Z 185条款的完整合规Agent Loop代码:

class GBZ185CompliantAgent:
    """
    完全符合GB/Z 185的Agent Loop实现。
    
    覆盖标准:第2部分(身份码) + 第7部分(工具调用) + 第3部分(身份管理) + 第6部分(智能体交互) + 第1部分(总体架构)
    """
    
    def __init__(self, llm, tools, permissions: dict, agent_id: str = None, tenant_id: str = "default"):
        self.llm = llm
        self.tools = {t.name: t for t in tools}  # StandardTool对象
        self.security_grader = SecurityGrader(permissions)
        self.checkpoint_manager = CheckpointManager()
        self.max_iterations = 10
        
        # 标准5.2:身份标识
        self.agent_id = agent_id or str(uuid.uuid4())
        self.tenant_id = tenant_id
        self.session_id = str(uuid.uuid4())
        
        # 标准7.1:审计日志
        self.audit_logger = AuditLogger(self.agent_id, self.session_id)
    
    def run(self, user_input: str) -> dict:
        """运行符合GB/Z 185的Agent Loop。"""
        state = self._create_initial_state(user_input)
        
        while not state["is_complete"] and state["iteration"] < self.max_iterations:
            try:
                # 1. Checkpoint保存(标准7.2:错误恢复)
                self.checkpoint_manager.save(state)
                
                # 2. LLM决策(标准7.1:审计)
                response = self.llm.invoke(state["messages"])
                self.audit_logger.log_decision(
                    state["iteration"],
                    str(state["messages"]),
                    response.content,
                    "gpt-4o"
                )
                state["messages"].append({"role": "assistant", "content": response.content})
                
                # 3. 检查工具调用
                if hasattr(response, 'tool_calls') and response.tool_calls:
                    for tool_call in response.tool_calls:
                        tool_name = tool_call["name"]
                        tool_params = tool_call["args"]
                        
                        # 4. Schema校验(标准5.3:通信规范)
                        tool = self.tools.get(tool_name)
                        if tool and not tool.validate_input(tool_params):
                            raise ValueError(f"工具参数不符合标准格式: {tool_name}")
                        
                        # 5. 安全校验(标准6.1:安全认证)
                        security_result = self.security_grader.check(
                            tool_name, tool_params, state["agent_id"]
                        )
                        if not security_result.passed:
                            raise PermissionError(f"安全校验失败: {security_result.violations}")
                        
                        # 6. 执行工具
                        import time
                        start = time.time()
                        result = tool.invoke(tool_params) if hasattr(tool, 'invoke') else "模拟结果"
                        duration = (time.time() - start) * 1000
                        
                        # 7. 输出校验(标准5.3)
                        if tool and not tool.validate_output(result):
                            raise ValueError(f"工具输出不符合标准格式: {tool_name}")
                        
                        # 8. 审计日志(标准7.1)
                        self.audit_logger.log_tool_call(
                            state["iteration"], tool_name, tool_params, result, duration
                        )
                        
                        state["messages"].append({
                            "role": "tool", "name": tool_name, "content": str(result)
                        })
                        state["tool_calls"].append({"name": tool_name, "result": result})
                else:
                    state["is_complete"] = True
                
                state["iteration"] += 1
                
            except Exception as e:
                # 标准7.2:错误恢复
                print(f"❌ 异常发生: {e}")
                print(f"🔄 回滚到最近Checkpoint...")
                state = self.checkpoint_manager.rollback()
                state["messages"].append({
                    "role": "system",
                    "content": f"[系统恢复] 发生异常,已回滚到迭代{state['iteration']}。错误: {str(e)}"
                })
        
        return {
            "output": state["messages"][-1]["content"],
            "session_id": state["session_id"],
            "iterations": state["iteration"],
            "audit_log_file": self.audit_logger.log_file,
        }
    
    def _create_initial_state(self, user_input: str) -> CompliantAgentState:
        """创建符合标准的初始状态。"""
        return CompliantAgentState(
            messages=[{"role": "user", "content": user_input}],
            tool_calls=[],
            iteration=0,
            is_complete=False,
            agent_id=self.agent_id,
            tenant_id=self.tenant_id,
            session_id=self.session_id,
            security_level="public",
        )

四、合规Checklist:可直接用于项目评审

下面这份Checklist可以直接复制到企业的合规文档或项目评审中。每个检查项对应GB/Z 185的一个条款,验证方法明确可执行:

序号 标准条款 检查项 工程实现 验证方法 是否通过
1 5.2 身份标识 Agent是否有唯一标识 State中持久化agent_id 检查日志中是否有agent_id字段 [ ]
2 5.2 身份标识 是否支持多租户隔离 State中包含tenant_id 检查不同租户的数据是否隔离 [ ]
3 5.3 通信规范 工具输入是否有Schema校验 StandardTool.validate_input() 传入非法参数,检查是否被拦截 [ ]
4 5.3 通信规范 工具输出是否有Schema校验 StandardTool.validate_output() 模拟非法输出,检查是否被拦截 [ ]
5 6.1 安全认证 是否有权限检查机制 SecurityGrader.check() 模拟越权调用,检查是否被拦截 [ ]
6 6.1 安全认证 是否遵循最小权限原则 权限配置中只开放必要工具 检查权限配置是否包含冗余工具 [ ]
7 6.2 数据隐私 敏感数据是否有分级 DataSensitivity枚举 检查MemoryItem是否包含敏感度字段 [ ]
8 6.2 数据隐私 低权限用户是否能访问高敏感数据 can_access()权限检查 模拟低权限访问,检查是否被拒绝 [ ]
9 6.2 数据隐私 公开输出时是否自动脱敏 sanitize_for_output() 检查机密数据输出给public时是否脱敏 [ ]
10 7.1 可审计性 是否有决策日志 AuditLogger.log_decision() 检查日志中是否有LLM调用记录 [ ]
11 7.1 可审计性 是否有工具调用日志 AuditLogger.log_tool_call() 检查日志中是否有Tool调用记录 [ ]
12 7.1 可审计性 是否能按会话查询日志 AuditLogger.query() 输入session_id,检查能否返回完整轨迹 [ ]
13 7.2 错误恢复 是否有Checkpoint机制 CheckpointManager.save() 检查运行过程中是否定期保存快照 [ ]
14 7.2 错误恢复 异常时是否能回滚 CheckpointManager.rollback() 模拟异常,检查是否回滚到安全状态 [ ]
15 7.2 错误恢复 回滚后是否能继续运行 异常处理后的循环继续逻辑 检查回滚后Agent是否继续执行 [ ]

五、生产环境警告:合规不是"全量上线",而是"成本与收益的权衡"

⚠️ 重要提示:GB/Z 185合规在生产环境中会引入额外的性能开销、存储成本和系统复杂度。不是每个Agent都需要"全量合规",根据场景选择合规等级。

合规等级与适用场景

合规等级 适用场景 需要实现条款 性能开销 存储成本
基础合规 内部工具、个人项目 5.2(身份标识)+ 7.2(错误恢复) 低(仅增加字段+Checkpoint)
标准合规 企业内网应用、SaaS产品 基础合规 + 5.3(通信规范)+ 6.1(安全认证) 中(Schema校验+权限检查)
完整合规 金融、政务、医疗等强监管行业 全部7个条款 高(审计日志+数据分级+全链路追踪)

生产环境建议

  1. 渐进式合规:先用基础合规跑通业务,再按监管要求逐步升级,不要一次引入全部条款
  2. 审计日志存储成本AuditLogger每条日志约500字节-2KB,日均1万次运行=5-20MB/天,月存储150-600MB——建议设置日志保留策略(如保留90天)
  3. 性能监控指标:Schema校验增加约5-10ms延迟、权限检查增加约2-5ms延迟、Checkpoint深拷贝增加约10-50ms(取决于State大小)
  4. 敏感数据最小化AuditLogger中只记录参数预览(前200字符),不要记录完整敏感数据——日志文件本身也需要加密存储
  5. 权限配置热更新SecurityGrader的权限策略不要硬编码,建议从数据库/配置中心加载,支持运行时更新

一句话总结:合规是"成本与风险"的权衡,不是"全量"或"不做"的二选一。内部工具基础合规即可,强监管行业才需要完整合规。


六、总结:国家标准不是束缚,而是工程化的"需求规格说明书"

本文将GB/Z 185系列标准的7个关键要求,逐条映射到了Agent Loop的具体代码实现:

标准分册 解决的问题 提供的代码
第2部分 身份码 Agent"是谁" CompliantAgentState(含agent_id/tenant_id)
第7部分 工具调用 工具调用"格式对不对" StandardTool(Schema校验)
第3部分 身份管理 “能不能调用这个工具” SecurityGrader(权限检查)
第6部分 智能体交互 “敏感数据怎么保护” DataSensitivity(分级+脱敏)
第1部分 总体架构 “做了什么,怎么证明” AuditLogger(全链路日志)
第6部分 智能体交互 “坏了怎么恢复” CheckpointManager(快照+回滚)
第6部分 智能体交互 “多个Agent怎么协调” 预留mode参数(系列下篇展开)

核心观点:很多开发者把国家标准视为"合规负担",但实际上——GB/Z 185的每一条要求,都对应着Agent工程中一个真实存在的痛点。身份标识对应多租户、安全认证对应权限管理、可审计性对应运维监控、错误恢复对应系统稳定性。标准不是"额外加的东西",而是"帮你把工程化做扎实的设计指南"。



七、速查卡:七条款代码对照表

快速对照表:根据你的合规需求,直接查这张表找到对应的代码和Checklist。

序号 标准分册 解决的问题 核心类 关键方法 合规Checklist序号 生产注意
1 第2部分 身份码 Agent"是谁" CompliantAgentState agent_id/tenant_id字段 1-2 生产必须显式传入agent_id,不要依赖UUID自动生成
2 第7部分 工具调用 工具调用"格式对不对" StandardTool validate_input()/validate_output() 3-4 每个工具必须定义input/output Schema,否则无法校验
3 第3部分 身份管理 “能不能调用这个工具” SecurityGrader check(tool_name, params, agent_id) 5-6 权限配置建议从数据库加载,支持热更新
4 第6部分 智能体交互 “敏感数据怎么保护” DataSensitivity can_access()/sanitize_for_output() 7-9 审计日志中只记录参数预览(前200字符),不要记录完整敏感数据
5 第1部分 总体架构 “做了什么,怎么证明” AuditLogger log_decision()/log_tool_call()/query() 10-12 日志存储成本约5-20MB/天(1万次运行),建议保留90天
6 第6部分 智能体交互 “坏了怎么恢复” CheckpointManager save()/rollback() 13-15 每3-5次迭代保存一次,State过大时考虑持久化到Redis
7 第6部分 智能体交互 “多个Agent怎么协调” CompliantAgentLoop 预留mode参数 系列下篇 预留扩展点,当前版本不展开

使用建议

  • 内部工具 → 实现第2部分+第6部分(基础合规)
  • 企业SaaS → 实现第2/7/3/6/1部分(标准合规)
  • 金融/政务/医疗 → 实现全部标准分册(完整合规)

八、更新日志

日期 版本 更新内容
2026-07-09 v1.0 初稿:从标准到代码:GB/Z 185合规的Agent Loop设计(附Python实现+合规Checklist)

相关阅读:


你的Agent项目需要过国标合规审查吗? 评论区说说你遇到的合规难点——是权限管理太复杂、审计日志不知道怎么设计,还是多租户隔离没思路?我会针对高频问题单独出篇深度解答。

收藏这篇"标准+代码"对照指南,下次合规审查时直接翻出来对照Checklist打勾。觉得有用的话点赞+收藏,让更多智能体开发者看到这篇合规工程化指南。

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐