解决 agent failed before reply: no credentials found for profile 错误的技术实践

1次阅读

共计 3403 个字符，预计需要花费 9 分钟才能阅读完成。

最近在开发一个基于 agent 的自动化系统时，遇到了一个让人头疼的错误：agent failed before reply: no credentials found for profile "qwen-portal:def"。这个错误直接导致整个 agent 服务无法正常启动，影响了后续所有的业务流程。

这种错误在实际开发中并不少见，尤其是在分布式系统或微服务架构中，credentials（凭证）管理不当很容易导致类似问题。它不仅会影响系统的可用性，还可能引发安全风险。

经过仔细排查，发现这个错误主要有以下几个可能的根源：

配置文件确实缺失：系统找不到指定的 credentials 配置文件
文件路径配置错误：虽然文件存在，但系统查找的路径不正确
权限问题：运行 agent 的用户没有读取配置文件的权限
配置文件格式错误：文件存在但内容格式不符合要求
环境变量未正确设置：某些情况下 credentials 是通过环境变量传递的

首先，我们需要确保 credentials 配置文件正确存在并位于正确的位置。通常这类配置文件会放在以下位置之一：

/etc/agent/credentials.yaml
~/.agent/credentials.json
项目目录下的 config/credentials.yml

下面是一个标准的 YAML 格式 credentials 配置示例：

# credentials.yaml 示例
profiles:
  qwen-portal:
    def:
      access_key: AKIAxxxxxxxxxxxx
      secret_key: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
      region: us-west-2
    staging:
      access_key: AKIAyyyyyyyyyyyy
      secret_key: yyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyy
      region: eu-central-1

在 Python 中，我们可以这样实现动态加载：

import os
import yaml
from pathlib import Path

class CredentialsLoader:
    def __init__(self):
        self.config_locations = [Path('/etc/agent/credentials.yaml'),
            Path.home() / '.agent' / 'credentials.yaml',
            Path('config') / 'credentials.yaml'
        ]

    def load_credentials(self, profile_name):
        for config_path in self.config_locations:
            if config_path.exists():
                try:
                    with open(config_path, 'r') as f:
                        config = yaml.safe_load(f)
                        return config['profiles'].get(profile_name, {})
                except Exception as e:
                    print(f"Error loading config from {config_path}: {str(e)}")
        raise ValueError(f"No credentials found for profile {profile_name}")

下面是一个更完整的实现，包含错误处理和重试机制：

import yaml
import time
from pathlib import Path
from typing import Dict, Optional

class CredentialManager:
    """凭证管理器，负责加载和验证 credentials"""

    MAX_RETRIES = 3
    RETRY_DELAY = 1  # 秒

    def __init__(self, config_paths: list = None):
        self.config_paths = config_paths or [Path('/etc/agent/credentials.yaml'),
            Path.home() / '.agent' / 'credentials.yaml',
            Path('config') / 'credentials.yaml'
        ]

    def get_credentials(self, profile: str) -> Dict:
        """
        获取指定 profile 的 credentials
        :param profile: profile 名称，格式为 "service:env" 如 "qwen-portal:def"
        :return: credentials 字典
        :raises: ValueError 当无法找到有效的 credentials 时
        """service, env = profile.split(':')

        for attempt in range(self.MAX_RETRIES):
            try:
                config = self._load_config()
                if service in config.get('profiles', {}) and \
                   env in config['profiles'][service]:
                    return config['profiles'][service][env]

                # 如果找不到配置，稍等后重试
                if attempt < self.MAX_RETRIES - 1:
                    time.sleep(self.RETRY_DELAY)
                    continue

            except Exception as e:
                print(f"Attempt {attempt + 1} failed: {str(e)}")
                if attempt == self.MAX_RETRIES - 1:
                    raise ValueError(f"Failed to load credentials after {self.MAX_RETRIES} attempts")
                time.sleep(self.RETRY_DELAY)
                continue

        raise ValueError(f"No credentials found for profile {profile}")

    def _load_config(self) -> Dict:
        """从配置路径加载配置文件"""
        for config_path in self.config_paths:
            if config_path.exists():
                try:
                    with open(config_path, 'r') as f:
                        return yaml.safe_load(f) or {}
                except yaml.YAMLError as e:
                    print(f"YAML parsing error in {config_path}: {str(e)}")
                except Exception as e:
                    print(f"Error reading {config_path}: {str(e)}")
        return {}