Skip to content

4.3 实战:配置管理系统

🎯 小白理解:为什么需要"配置管理系统"?

想象你开了一家连锁奶茶店:

  • 没有配置系统:每家店的配方、价格、原料都手动管理,乱成一团
  • 有配置系统:统一管理所有店的设置,改一处全部生效

AI Agent 也是这样

Agent 配置包括:
├── 用哪个模型?(GPT-4、Claude...)
├── 温度是多少?(0.7、0.9...)
├── 开启哪些工具?(搜索、计算器...)
└── 系统提示词是什么?

配置管理系统帮你:

  1. 统一存储:所有配置放在一起
  2. 格式规范:使用 YAML/JSON,不会乱写
  3. 版本控制:改错了可以回滚
  4. 环境切换:开发/测试/生产用不同配置

Agent 配置管理系统

🎯 小白理解:什么是 @dataclass

Python 的 @dataclass 是一个"快捷方式",帮你自动生成类的常用方法。

不用 dataclass(麻烦):

python
class Person:
    def __init__(self, name, age):
        self.name = name
        self.age = age

    def __repr__(self):
        return f"Person(name={self.name}, age={self.age})"

用 dataclass(简洁):

python
@dataclass
class Person:
    name: str
    age: int
    # __init__ 和 __repr__ 自动生成!

类比@dataclass 就像一个"智能模板",你只需要声明数据字段,它帮你把杂活都干了!

python
from pathlib import Path
from typing import Dict, Any, Optional, List
import json
import yaml
from dataclasses import dataclass, field, asdict
from datetime import datetime

@dataclass
class ToolConfig:
    """工具配置"""
    name: str
    enabled: bool = True
    timeout: float = 30.0
    max_retries: int = 3
    settings: Dict[str, Any] = field(default_factory=dict)

@dataclass
class ModelConfig:
    """模型配置"""
    provider: str  # openai, anthropic, etc.
    model_name: str
    temperature: float = 0.7
    max_tokens: int = 2000
    top_p: float = 1.0
    frequency_penalty: float = 0.0
    presence_penalty: float = 0.0

@dataclass
class AgentConfig:
    """完整 Agent 配置"""
    name: str
    description: str
    version: str
    
    # 模型配置
    model: ModelConfig
    
    # 工具配置
    tools: List[ToolConfig] = field(default_factory=list)
    
    # 系统提示词
    system_prompt: str = "你是一个有用的AI助手。"
    
    # 运行时配置
    max_iterations: int = 10
    verbose: bool = False
    
    # 元数据
    created_at: str = field(default_factory=lambda: datetime.now().isoformat())
    updated_at: str = field(default_factory=lambda: datetime.now().isoformat())

class ConfigManager:
    """配置管理器"""
    
    def __init__(self, config_dir: str = "configs"):
        self.config_dir = Path(config_dir)
        self.config_dir.mkdir(exist_ok=True)
        self.cache: Dict[str, AgentConfig] = {}
    
    def save_config(self, config: AgentConfig, format: str = "yaml") -> Path:
        """保存配置到文件"""
        config.updated_at = datetime.now().isoformat()
        
        file_name = f"{config.name}.{format}"
        file_path = self.config_dir / file_name
        
        config_dict = asdict(config)
        
        if format == "json":
            with file_path.open("w", encoding="utf-8") as f:
                json.dump(config_dict, f, indent=2, ensure_ascii=False)
        
        elif format == "yaml":
            with file_path.open("w", encoding="utf-8") as f:
                yaml.dump(config_dict, f, allow_unicode=True, sort_keys=False)
        
        else:
            raise ValueError(f"不支持的格式: {format}")
        
        # 更新缓存
        self.cache[config.name] = config
        
        return file_path
    
    def load_config(self, name: str, format: str = "yaml") -> AgentConfig:
        """从文件加载配置"""
        # 检查缓存
        if name in self.cache:
            return self.cache[name]
        
        file_name = f"{name}.{format}"
        file_path = self.config_dir / file_name
        
        if not file_path.exists():
            raise FileNotFoundError(f"配置文件不存在: {file_path}")
        
        with file_path.open("r", encoding="utf-8") as f:
            if format == "json":
                data = json.load(f)
            elif format == "yaml":
                data = yaml.safe_load(f)
            else:
                raise ValueError(f"不支持的格式: {format}")
        
        # 转换为配置对象
        config = self._dict_to_config(data)
        
        # 缓存
        self.cache[name] = config
        
        return config
    
    def _dict_to_config(self, data: Dict[str, Any]) -> AgentConfig:
        """字典转配置对象"""
        # 转换 ModelConfig
        model_data = data.pop("model")
        model = ModelConfig(**model_data)
        
        # 转换 ToolConfig 列表
        tools_data = data.pop("tools", [])
        tools = [ToolConfig(**tool) for tool in tools_data]
        
        return AgentConfig(
            model=model,
            tools=tools,
            **data
        )
    
    def list_configs(self) -> List[str]:
        """列出所有配置"""
        configs = []
        for file_path in self.config_dir.glob("*"):
            if file_path.suffix in [".json", ".yaml", ".yml"]:
                configs.append(file_path.stem)
        return configs
    
    def delete_config(self, name: str):
        """删除配置"""
        # 删除缓存
        self.cache.pop(name, None)
        
        # 删除文件
        for ext in [".json", ".yaml", ".yml"]:
            file_path = self.config_dir / f"{name}{ext}"
            if file_path.exists():
                file_path.unlink()
    
    def update_config(
        self,
        name: str,
        updates: Dict[str, Any],
        format: str = "yaml"
    ) -> AgentConfig:
        """更新配置"""
        config = self.load_config(name, format)
        
        # 应用更新
        for key, value in updates.items():
            if hasattr(config, key):
                setattr(config, key, value)
        
        # 保存
        self.save_config(config, format)
        
        return config
    
    def validate_config(self, config: AgentConfig) -> List[str]:
        """验证配置"""
        errors = []
        
        # 验证模型配置
        if not config.model.provider:
            errors.append("模型提供商不能为空")
        
        if not config.model.model_name:
            errors.append("模型名称不能为空")
        
        if not 0 <= config.model.temperature <= 2:
            errors.append("temperature 必须在 0-2 之间")
        
        if config.model.max_tokens <= 0:
            errors.append("max_tokens 必须大于 0")
        
        # 验证工具配置
        for tool in config.tools:
            if not tool.name:
                errors.append("工具名称不能为空")
            
            if tool.timeout <= 0:
                errors.append(f"工具 {tool.name} 的超时时间必须大于 0")
        
        # 验证运行时配置
        if config.max_iterations <= 0:
            errors.append("max_iterations 必须大于 0")
        
        return errors

# 使用示例
manager = ConfigManager()

# 创建配置
research_config = AgentConfig(
    name="research_agent",
    description="专业的研究助手",
    version="1.0.0",
    model=ModelConfig(
        provider="openai",
        model_name="gpt-4",
        temperature=0.7,
        max_tokens=2000
    ),
    tools=[
        ToolConfig(
            name="search",
            enabled=True,
            timeout=10.0,
            settings={"max_results": 5}
        ),
        ToolConfig(
            name="calculator",
            enabled=True,
            timeout=5.0
        )
    ],
    system_prompt="你是一个专业的研究助手,擅长查找和分析信息。",
    max_iterations=10,
    verbose=True
)

# 验证配置
errors = manager.validate_config(research_config)
if errors:
    print("配置错误:")
    for error in errors:
        print(f"  - {error}")
else:
    # 保存配置
    config_path = manager.save_config(research_config, format="yaml")
    print(f"配置已保存到: {config_path}")

# 加载配置
loaded_config = manager.load_config("research_agent", format="yaml")
print(f"\n加载的配置: {loaded_config.name}")
print(f"模型: {loaded_config.model.model_name}")
print(f"工具数量: {len(loaded_config.tools)}")

# 更新配置
manager.update_config(
    "research_agent",
    updates={
        "temperature": 0.8,
        "verbose": False
    }
)

# 列出所有配置
configs = manager.list_configs()
print(f"\n所有配置: {configs}")

配置模板系统

🎯 小白理解:什么是"配置模板"?

想象你是一个奶茶店老板:

  • 没有模板:每次开新店都要从零开始写配方
  • 有模板:有"奶茶模板"、"果茶模板",新店直接套用,稍微调整就行

配置模板就是这个道理

模板名预设配置适用场景
researchGPT-4 + 搜索工具 + 温度 0.7研究助手
writerClaude + 无工具 + 温度 0.9创意写作
coderGPT-4 + 代码解释器 + 温度 0.3编程助手

新建 Agent 时,选个模板,改改参数就好了!

python
from typing import Dict, Any
import re

class ConfigTemplate:
    """配置模板系统"""
    
    def __init__(self):
        self.templates: Dict[str, Dict[str, Any]] = {}
        self._load_default_templates()
    
    def _load_default_templates(self):
        """加载默认模板"""
        self.templates = {
            "research": {
                "description": "研究助手模板",
                "model": {
                    "provider": "openai",
                    "model_name": "gpt-4",
                    "temperature": 0.7
                },
                "tools": [
                    {"name": "search", "enabled": True},
                    {"name": "calculator", "enabled": True},
                    {"name": "wikipedia", "enabled": True}
                ],
                "system_prompt": "你是一个专业的研究助手。",
                "max_iterations": 10
            },
            
            "writer": {
                "description": "写作助手模板",
                "model": {
                    "provider": "anthropic",
                    "model_name": "claude-3-5-sonnet-20241022",
                    "temperature": 0.9
                },
                "tools": [],
                "system_prompt": "你是一个创意写作助手,风格优美简洁。",
                "max_iterations": 5
            },
            
            "coder": {
                "description": "编程助手模板",
                "model": {
                    "provider": "openai",
                    "model_name": "gpt-4",
                    "temperature": 0.3
                },
                "tools": [
                    {"name": "code_interpreter", "enabled": True},
                    {"name": "file_search", "enabled": True}
                ],
                "system_prompt": "你是一个专业的编程助手。",
                "max_iterations": 15
            }
        }
    
    def create_from_template(
        self,
        template_name: str,
        agent_name: str,
        overrides: Optional[Dict[str, Any]] = None
    ) -> AgentConfig:
        """从模板创建配置"""
        if template_name not in self.templates:
            raise ValueError(f"模板不存在: {template_name}")
        
        # 深拷贝模板
        template = self.templates[template_name].copy()
        
        # 应用覆盖
        if overrides:
            self._apply_overrides(template, overrides)
        
        # 创建配置
        model = ModelConfig(**template.pop("model"))
        tools_data = template.pop("tools")
        tools = [ToolConfig(**tool) for tool in tools_data]
        
        return AgentConfig(
            name=agent_name,
            model=model,
            tools=tools,
            version="1.0.0",
            **template
        )
    
    def _apply_overrides(self, template: Dict[str, Any], overrides: Dict[str, Any]):
        """应用配置覆盖"""
        for key, value in overrides.items():
            if isinstance(value, dict) and key in template:
                # 递归更新嵌套字典
                template[key].update(value)
            else:
                template[key] = value
    
    def list_templates(self) -> List[str]:
        """列出所有模板"""
        return list(self.templates.keys())
    
    def add_template(self, name: str, template: Dict[str, Any]):
        """添加自定义模板"""
        self.templates[name] = template

# 使用示例
template_manager = ConfigTemplate()

# 列出模板
print("可用模板:")
for name in template_manager.list_templates():
    template = template_manager.templates[name]
    print(f"  - {name}: {template['description']}")

# 从模板创建配置
config = template_manager.create_from_template(
    template_name="research",
    agent_name="my_research_bot",
    overrides={
        "model": {
            "temperature": 0.5
        },
        "system_prompt": "你是一个学术研究助手。"
    }
)

print(f"\n创建的配置: {config.name}")
print(f"模型: {config.model.model_name}")
print(f"温度: {config.model.temperature}")
print(f"工具: {[tool.name for tool in config.tools]}")

环境感知配置

🎯 小白理解:为什么需要"环境感知"?

软件开发通常有 3 个环境:

环境用途配置特点
开发 (dev)程序员本地调试详细日志、测试 API Key
测试 (staging)上线前测试模拟生产、可能出错
生产 (production)用户使用稳定、正式 API Key

同一个 Agent,不同环境用不同配置

configs/
├── development/
│   └── research_agent.yaml  # 开发环境:详细日志
├── staging/
│   └── research_agent.yaml  # 测试环境:测试 API
└── production/
    └── research_agent.yaml  # 生产环境:正式 API

程序自动检测当前环境(通过 ENV 环境变量),加载对应配置!

python
import os
from enum import Enum

class Environment(Enum):
    """环境类型"""
    DEVELOPMENT = "development"
    STAGING = "staging"
    PRODUCTION = "production"

class EnvironmentAwareConfig:
    """环境感知配置管理"""
    
    def __init__(self, base_config_dir: str = "configs"):
        self.base_dir = Path(base_config_dir)
        self.env = self._detect_environment()
        self.manager = ConfigManager(str(self.base_dir / self.env.value))
    
    def _detect_environment(self) -> Environment:
        """检测当前环境"""
        env_name = os.getenv("ENV", "development").lower()
        
        if env_name in ["prod", "production"]:
            return Environment.PRODUCTION
        elif env_name in ["stage", "staging"]:
            return Environment.STAGING
        else:
            return Environment.DEVELOPMENT
    
    def load_config(self, name: str) -> AgentConfig:
        """加载环境特定的配置"""
        return self.manager.load_config(name)
    
    def get_environment(self) -> Environment:
        """获取当前环境"""
        return self.env

# 使用示例
# 目录结构:
# configs/
#   development/
#     research_agent.yaml
#   staging/
#     research_agent.yaml
#   production/
#     research_agent.yaml

env_manager = EnvironmentAwareConfig()
print(f"当前环境: {env_manager.get_environment().value}")

config = env_manager.load_config("research_agent")
print(f"加载配置: {config.name}")

配置版本控制

🎯 小白理解:为什么需要"版本控制"?

想象你在写论文:

  • 没有版本控制:改错了?完了,原来的版本没了
  • 有版本控制:每次修改都保存一个版本,改错了可以"后悔"

配置版本控制的作用

config_versions/
└── research_agent/
    ├── 1.0.0.json  ← 初始版本(Alice 创建)
    ├── 1.0.1.json  ← 调整温度(Bob 修改)
    └── 1.0.2.json  ← 添加新工具(Alice 修改)

好处

  1. 可追溯:谁改了什么?什么时候改的?
  2. 可回滚:新版本有 bug?切回旧版本
  3. 可对比:两个版本有什么区别?

类似 Git 对代码的管理,这是对配置的管理!

python
from dataclasses import dataclass
from datetime import datetime
from typing import List, Optional

@dataclass
class ConfigVersion:
    """配置版本"""
    version: str
    config: AgentConfig
    created_at: datetime
    author: str
    comment: str

class ConfigVersionControl:
    """配置版本控制"""
    
    def __init__(self, storage_dir: str = "config_versions"):
        self.storage_dir = Path(storage_dir)
        self.storage_dir.mkdir(exist_ok=True)
    
    def save_version(
        self,
        config: AgentConfig,
        author: str,
        comment: str = ""
    ) -> str:
        """保存配置版本"""
        # 生成版本号
        version = self._generate_version(config.name)
        
        # 创建版本对象
        config_version = ConfigVersion(
            version=version,
            config=config,
            created_at=datetime.now(),
            author=author,
            comment=comment
        )
        
        # 保存
        version_dir = self.storage_dir / config.name
        version_dir.mkdir(exist_ok=True)
        
        file_path = version_dir / f"{version}.json"
        
        with file_path.open("w", encoding="utf-8") as f:
            data = {
                "version": version,
                "created_at": config_version.created_at.isoformat(),
                "author": author,
                "comment": comment,
                "config": asdict(config)
            }
            json.dump(data, f, indent=2, ensure_ascii=False)
        
        return version
    
    def _generate_version(self, config_name: str) -> str:
        """生成版本号"""
        version_dir = self.storage_dir / config_name
        
        if not version_dir.exists():
            return "1.0.0"
        
        # 获取最新版本
        versions = [f.stem for f in version_dir.glob("*.json")]
        
        if not versions:
            return "1.0.0"
        
        # 解析版本号并递增
        latest = sorted(versions, key=lambda v: list(map(int, v.split("."))))[-1]
        major, minor, patch = map(int, latest.split("."))
        
        return f"{major}.{minor}.{patch + 1}"
    
    def load_version(self, config_name: str, version: str) -> AgentConfig:
        """加载指定版本"""
        file_path = self.storage_dir / config_name / f"{version}.json"
        
        if not file_path.exists():
            raise FileNotFoundError(f"版本不存在: {version}")
        
        with file_path.open("r", encoding="utf-8") as f:
            data = json.load(f)
        
        # 转换配置
        config_data = data["config"]
        model = ModelConfig(**config_data.pop("model"))
        tools = [ToolConfig(**t) for t in config_data.pop("tools")]
        
        return AgentConfig(model=model, tools=tools, **config_data)
    
    def list_versions(self, config_name: str) -> List[Dict[str, Any]]:
        """列出所有版本"""
        version_dir = self.storage_dir / config_name
        
        if not version_dir.exists():
            return []
        
        versions = []
        for file_path in sorted(version_dir.glob("*.json")):
            with file_path.open("r", encoding="utf-8") as f:
                data = json.load(f)
                versions.append({
                    "version": data["version"],
                    "created_at": data["created_at"],
                    "author": data["author"],
                    "comment": data["comment"]
                })
        
        return versions

# 使用示例
vc = ConfigVersionControl()

# 保存版本
version = vc.save_version(
    config=research_config,
    author="Alice",
    comment="初始版本"
)
print(f"保存版本: {version}")

# 列出版本
versions = vc.list_versions("research_agent")
print("\n版本历史:")
for v in versions:
    print(f"  {v['version']} - {v['author']} - {v['comment']}")

# 加载版本
loaded = vc.load_version("research_agent", "1.0.0")
print(f"\n加载版本: {loaded.name}")

关键要点

  1. 结构化配置:使用 dataclass 定义清晰的配置结构
  2. 多格式支持:支持 JSON 和 YAML 两种格式
  3. 配置验证:在保存和加载时验证配置
  4. 模板系统:使用模板快速创建标准配置
  5. 版本控制:跟踪配置变更历史

下一节:4.4 小结和复习

基于 MIT 许可证发布。内容版权归作者所有。