4.3 实战:配置管理系统
🎯 小白理解:为什么需要"配置管理系统"?
想象你开了一家连锁奶茶店:
- 没有配置系统:每家店的配方、价格、原料都手动管理,乱成一团
- 有配置系统:统一管理所有店的设置,改一处全部生效
AI Agent 也是这样:
Agent 配置包括: ├── 用哪个模型?(GPT-4、Claude...) ├── 温度是多少?(0.7、0.9...) ├── 开启哪些工具?(搜索、计算器...) └── 系统提示词是什么?配置管理系统帮你:
- 统一存储:所有配置放在一起
- 格式规范:使用 YAML/JSON,不会乱写
- 版本控制:改错了可以回滚
- 环境切换:开发/测试/生产用不同配置
Agent 配置管理系统
🎯 小白理解:什么是
@dataclass?Python 的
@dataclass是一个"快捷方式",帮你自动生成类的常用方法。不用 dataclass(麻烦):
pythonclass 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 修改)好处:
- 可追溯:谁改了什么?什么时候改的?
- 可回滚:新版本有 bug?切回旧版本
- 可对比:两个版本有什么区别?
类似 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}")关键要点
- 结构化配置:使用 dataclass 定义清晰的配置结构
- 多格式支持:支持 JSON 和 YAML 两种格式
- 配置验证:在保存和加载时验证配置
- 模板系统:使用模板快速创建标准配置
- 版本控制:跟踪配置变更历史
下一节:4.4 小结和复习