Agent 环境感知能力让智能体知道自己在哪个集群和命名空间一、Agent 把生产环境的日志输出到了开发环境的 Elasticsearch排查了半天才发现Agent 环境感知不是Agent 知道自己叫什么这么简单是 Agent 需要理解它运行在什么环境中——这是生产集群还是开发集群、这个命名空间里有哪些可用的工具和 API、当前用户有什么权限。Agent 的很多错误不是推理错误是环境误判——它尝试调用一个在生产集群不存在的 API、它查询了一个开发环境的数据库而不是生产环境的、它执行的操作超出了当前 ServiceAccount 的权限范围。Agent 环境感知通过两个渠道实现K8s 环境变量注入最可靠和向下 APIDownward API读取最灵活。Pod 在启动时携带着丰富的环境信息——namespace、pod name、node name、ServiceAccount——这些信息通过环境变量注入到 Agent 进程中。Agent 在做出任何外部操作之前都应该先自我感知这些环境信息。二、底层机制与原理剖析Agent 环境感知的四种信息源K8s Downward API通过 Pod 的env定义注入环境变量或者通过volume挂载。这些信息在 Pod 启动时由 K8s 自动填充不需要 Agent 代码做任何网络调用metadata.namespace→NAMESPACEmetadata.name→POD_NAMEspec.nodeName→NODE_NAMEspec.serviceAccountName→SERVICE_ACCOUNT自定义环境标签除了 K8s 自动提供的信息团队可以在 Pod 定义中添加自定义环境标签如DEPLOY_ENVproduction、TEAMai-platform。这些标签用于 Agent 的行为决策。K8s API 验证Agent 可以通过调用 K8s API/apis/authorization.k8s.io/v1/selfsubjectaccessreviews来验证自己当前有什么权限。例如Agent 在尝试创建一个 Pod 之前先检查自己是否能创建 Pod。如果不能返回明确的错误给用户而非操作失败。外部服务发现Agent 所在的命名空间可能有关联的外部服务如内部的向量数据库、模型注册中心。这些服务地址通过 ConfigMap 注入环境变量Agent 启动时自动发现。三、生产级代码实现# k8s/agent-deployment.yaml # Agent Pod 定义——注入环境感知信息 apiVersion: apps/v1 kind: Deployment metadata: name: agent-planner namespace: production labels: app: agent-planner env: production team: ai-platform spec: replicas: 2 selector: matchLabels: app: agent-planner template: metadata: labels: app: agent-planner env: production annotations: # 自定义元数据Agent 运行时读取 agent.workbuddy.tech/role: planner agent.workbuddy.tech/capabilities: k8s_query,db_read,log_search spec: serviceAccountName: agent-prod containers: - name: agent image: agent-planner:v1.5.0 env: # K8s Downward API 自动注入 - name: NAMESPACE valueFrom: fieldRef: fieldPath: metadata.namespace - name: POD_NAME valueFrom: fieldRef: fieldPath: metadata.name - name: POD_IP valueFrom: fieldRef: fieldPath: status.podIP - name: NODE_NAME valueFrom: fieldRef: fieldPath: spec.nodeName - name: SERVICE_ACCOUNT valueFrom: fieldRef: fieldPath: spec.serviceAccountName # 自定义环境标签 - name: DEPLOY_ENV value: production - name: CLUSTER_NAME value: prod-cluster-01 - name: LOG_ES_HOST valueFrom: configMapKeyRef: name: agent-config key: log_es_host - name: DB_HOST valueFrom: secretKeyRef: name: agent-db-credentials key: host resources: requests: memory: 512Mi cpu: 500m limits: memory: 2Gi cpu: 2 # 环境感知探活 livenessProbe: httpGet: path: /healthz port: 8080 initialDelaySeconds: 30 readinessProbe: httpGet: path: /readyz # 就绪探针包含环境验证 port: 8080 initialDelaySeconds: 10# agent/environment.py Agent 环境感知模块 在 Agent 启动时读取环境信息并在每次操作前验证环境兼容性 import os import logging from dataclasses import dataclass, field from typing import Dict, List, Optional from enum import Enum logger logging.getLogger(__name__) class DeployEnv(Enum): PRODUCTION production STAGING staging DEVELOPMENT development TEST test dataclass class EnvironmentContext: Agent 运行环境的完整上下文 由以下数据源合并 1. K8s Downward APInamespace, pod_name, service_account 2. 自定义环境变量deploy_env, cluster_name, team 3. K8s 注解agent capabilities, role namespace: str pod_name: str deploy_env: DeployEnv cluster_name: str service_account: str # 可选字段 node_name: str team: str capabilities: List[str] field(default_factorylist) pod_annotations: Dict[str, str] field(default_factorydict) classmethod def from_env(cls) - EnvironmentContext: 从环境变量构建上下文 为什么不用 dict 传递而是定义 dataclass - 类型安全编译时检查字段类型 - 强制必填字段deploy_env 为空时直接抛异常而不悄悄用默认值 deploy_env_str os.environ.get(DEPLOY_ENV, development) try: deploy_env DeployEnv(deploy_env_str) except ValueError: logger.error(Unknown DEPLOY_ENV: %s, defaulting to development, deploy_env_str) deploy_env DeployEnv.DEVELOPMENT # capabilities 以逗号分隔 caps [ c.strip() for c in os.environ.get(AGENT_CAPABILITIES, ).split(,) if c.strip() ] return cls( namespaceos.environ.get(NAMESPACE, default), pod_nameos.environ.get(POD_NAME, unknown), deploy_envdeploy_env, cluster_nameos.environ.get(CLUSTER_NAME, unknown), service_accountos.environ.get(SERVICE_ACCOUNT, default), node_nameos.environ.get(NODE_NAME, ), teamos.environ.get(TEAM, ), capabilitiescaps, ) def is_production(self) - bool: return self.deploy_env DeployEnv.PRODUCTION def can_perform(self, capability: str) - bool: 检查 Agent 是否具有某项能力 return capability in self.capabilities def require_confirmation_for(self, action: str) - bool: 判断某操作是否需要人工确认 规则 - 生产环境的所有删除操作需要确认 - Staging 的删除操作需要确认 - 开发环境不需要确认 destructive_verbs {delete, destroy, drop, purge} if self.deploy_env DeployEnv.DEVELOPMENT: return False if self.deploy_env DeployEnv.PRODUCTION: # 生产环境的任何写操作建议确认 return True # Staging只确认破坏性操作 for verb in destructive_verbs: if verb in action.lower(): return True return False class EnvironmentAwareAgent: 具有环境感知能力的 Agent 在每次执行外部操作前 1. 验证当前环境是否允许该操作 2. 如果是敏感操作要求确认 3. 在日志中标记环境信息便于排障追踪 def __init__(self, context: EnvironmentContext): self.context context logger.info( Agent initialized: env%s namespace%s cluster%s pod%s, self.context.deploy_env.value, self.context.namespace, self.context.cluster_name, self.context.pod_name, ) async def execute_action(self, action: str, target: str, **params) - Dict: 执行一个经过环境验证的操作 # 1. 环境验证 if not self._validate_environment(action, target): return { success: False, error: fAction {action} not allowed in {self.context.deploy_env.value} environment, reason: environment_restriction, } # 2. 确认检查 if self.context.require_confirmation_for(action): return { success: False, error: fAction {action} requires manual confirmation in {self.context.deploy_env.value}, reason: confirmation_required, action: action, target: target, } # 3. 执行带环境标签 logger.info( [%s/%s] Executing %s on %s, self.context.namespace, self.context.deploy_env.value, action, target, ) return {success: True, action: action, target: target} def _validate_environment(self, action: str, target: str) - bool: 验证当前环境是否允许该操作 # 规则 1开发环境不允许操作生产环境的目标 if (self.context.deploy_env ! DeployEnv.PRODUCTION and production in str(target).lower()): return False # 规则 2生产环境不允许执行 debug 级别的操作 debug_actions {debug_shell, debug_memory, debug_connections} if (self.context.deploy_env DeployEnv.PRODUCTION and action in debug_actions): return False return True # --------------------------------------------------------------------------- # 使用示例 # --------------------------------------------------------------------------- if __name__ __main__: context EnvironmentContext.from_env() agent EnvironmentAwareAgent(context) print(fAgent 环境: {context.deploy_env.value}) print(f命名空间: {context.namespace}) print(f集群: {context.cluster_name}) print(f能力: {context.capabilities}) print(f生产环境: {context.is_production()}) # 验证操作 import asyncio async def test(): result await agent.execute_action(delete, production-database) print(f结果: {result}) asyncio.run(test())四、边界分析与架构权衡环境感知的过度约束问题如果 Agent 严格按照 environment 限制行为可能在紧急排障时无法做必要的操作解决方案提供 override 机制——特定用户如 on-call 工程师可以临时提升 Agent 的权限。override 操作必须记录审计日志多集群场景下的环境一致性同一个 Agent 可能需要在不同集群中运行生产/灾备/灰度环境感知信息必须准确反映当前集群如果用同一个镜像部署不同集群环境信息完全靠环境变量——确保环境变量设置正确Security防止环境变量被篡改环境变量在容器内部可以被进程修改。不应该完全依赖环境变量做安全决策如权限判断权限判断应调用 K8s SelfSubjectAccessReview API 做二次验证五、总结Agent 环境感知的核心是通过 K8s Downward API 和环境变量注入环境信息让 Agent 在做出任何外部操作之前知道自己在哪里。关键规则生产环境需要操作确认delete 等破坏性操作、debug 接口在非 dev 环境不可用、操作目标不能跨环境dev Agent 不能碰 prod 资源。环境信息不应只依赖环境变量——搭配 K8s SelfSubjectAccessReview API 做权限验证防止环境变量被篡改。
