夜雨聆风前言
配置系统是QwenPaw灵活性的基础，支持用户自定义模型、渠道、记忆等各个方面。本篇文章将深入解析QwenPaw的配置系统设计。
配置系统架构
QwenPaw的配置系统采用分层设计：
┌─────────────────────────────────────────────────────┐
│ 配置层级 │
├─────────────────────────────────────────────────────┤
│ Level 1: 全局配置 (config.yaml) │
│ Level 2: Agent配置 (agent_profile.yaml) │
│ Level 3: 环境变量 (.env / working.secret) │
│ Level 4: 运行时覆盖 (CLI参数) │
└─────────────────────────────────────────────────────┘
AgentProfileConfig：智能体配置
配置结构
@dataclass
class AgentProfileConfig:
"""智能体配置文件"""
id: str # 智能体唯一标识
name: str # 智能体名称
# 模型配置
active_model: ModelConfig # 当前使用的模型
available_models: list[ModelConfig] # 可用模型列表
# 运行配置
running: RunningConfig # 运行参数
language: str = "zh" # 语言设置
# 工具配置
tools: ToolsConfig | None = None # 工具启用状态
# 安全配置
security: SecurityConfig | None = None # 安全设置
# 渠道配置
channels: list[str] = field(default_factory=list) # 启用的渠道
# 记忆配置
memory: MemoryConfig | None = None # 记忆配置
# 心跳配置
heartbeat: HeartbeatConfig | None = None # 心跳设置
子配置详解
@dataclass
class RunningConfig:
"""运行配置"""
max_iters: int = 20 # 最大迭代次数
max_input_length: int = 8000 # 最大输入长度
memory_compact_threshold: float = 0.8 # 记忆压缩阈值
auto_continue_on_text_only: bool = True # 自动继续
shell_command_timeout: float = 30.0 # Shell超时时间
@dataclass
class ModelConfig:
"""模型配置"""
provider_id: str # 提供商ID (如 "openai", "dashscope")
model: str # 模型名称 (如 "gpt-4", "qwen-max")
api_key: str | None = None # API密钥
base_url: str | None = None # 自定义API地址
temperature: float = 0.7 # 温度参数
max_tokens: int = 4096 # 最大token数
@dataclass
class ToolsConfig:
"""工具配置"""
builtin_tools: dict[str, ToolSetting] # 内置工具设置
@dataclass
class ToolSetting:
"""工具设置"""
enabled: bool = True # 是否启用
async_execution: bool = False # 是否支持异步执行
配置加载流程
工作目录结构
~/.qwenpaw/
├── config.yaml # 全局配置
├── agents/ # 智能体配置
│ ├── default.yaml # 默认智能体
│ └── assistant.yaml # 自定义智能体
├── channels/ # 渠道配置
│ ├── dingtalk.yaml
│ ├── feishu.yaml
│ └── discord.yaml
├── memories/ # 记忆存储
├── skills/ # 技能目录
├── skills_config.yaml # 技能启用配置
├── .secret # 敏感信息（API密钥等）
└── .secret.backups/ # 备份
配置加载器
class ConfigLoader:
"""配置加载器"""
def __init__(self, working_dir: Path):
self.working_dir = working_dir
self.config_dir = working_dir / "config"
self.secrets_file = working_dir / ".secret"
async def load_agent_config(self, agent_id: str) -> AgentProfileConfig:
"""加载智能体配置"""
config_file = self.config_dir / "agents" / f"{agent_id}.yaml"
if not config_file.exists():
# 加载默认配置
config_file = self.config_dir / "agents" / "default.yaml"
# 读取配置
config_dict = yaml.safe_load(config_file.read_text())
# 合并Secrets
secrets = self._load_secrets()
config_dict = self._merge_secrets(config_dict, secrets)
# 验证配置
return self._validate_and_parse(config_dict)
def _load_secrets(self) -> dict:
"""加载敏感信息"""
if not self.secrets_file.exists():
return {}
# .secret文件支持多格式
if self.secrets_file.suffix == ".json":
return json.loads(self.secrets_file.read_text())
elif self.secrets_file.suffix in [".env", ""]:
return self._parse_env_file(self.secrets_file.read_text())
else:
return {}
环境变量管理
.env文件解析
class EnvManager:
"""环境变量管理器"""
def __init__(self, secrets_file: Path):
self.secrets_file = secrets_file
self.envs: dict[str, str] = {}
self._load()
def _load(self) -> None:
"""加载环境变量"""
if not self.secrets_file.exists():
return
content = self.secrets_file.read_text()
for line in content.splitlines():
line = line.strip()
# 跳过注释和空行
if not line or line.startswith("#"):
continue
# 解析 KEY=VALUE
if "=" in line:
key, value = line.split("=", 1)
key = key.strip()
value = value.strip()
# 支持引号
if value.startswith('"') and value.endswith('"'):
value = value[1:-1]
elif value.startswith("'") and value.endswith("'"):
value = value[1:-1]
self.envs[key] = value
def get(self, key: str, default: str | None = None) -> str | None:
"""获取环境变量"""
return self.envs.get(key, default)
def set(self, key: str, value: str) -> None:
"""设置环境变量"""
self.envs[key] = value
self._save()
def _save(self) -> None:
"""保存环境变量"""
lines = [f"{k}={v}" for k, v in self.envs.items()]
self.secrets_file.write_text("\n".join(lines))
环境变量加载到进程
# envs.py
def load_envs_into_environ() -> None:
"""将持久化的环境变量加载到os.environ"""
# 获取工作目录
working_dir = os.environ.get("QWENPAW_WORKING_DIR")
if not working_dir:
return
secrets_file = Path(working_dir) / ".secret"
if not secrets_file.exists():
return
# 解析.env格式
if secrets_file.suffix == ".env" or not secrets_file.suffix:
content = secrets_file.read_text()
for line in content.splitlines():
line = line.strip()
if not line or line.startswith("#"):
continue
if "=" in line:
key, value = line.split("=", 1)
key = key.strip()
value = value.strip().strip('"').strip("'")
os.environ[key] = value
配置验证
Pydantic验证
from pydantic import BaseModel, validator, Field
class AgentProfileConfig(BaseModel):
"""使用Pydantic验证的智能体配置"""
id: str = Field(..., min_length=1, max_length=100)
name: str = Field(..., min_length=1, max_length=50)
# 模型配置
active_model: ModelConfig
# 运行配置
running: RunningConfig = Field(default_factory=RunningConfig)
@validator("id")
def validate_id(cls, v):
"""验证ID格式"""
if not re.match(r"^[a-zA-Z0-9_-]+$", v):
raise ValueError(
"Agent ID must contain only alphanumeric characters, "
"underscores, and hyphens"
)
return v
@validator("running")
def validate_running(cls, v):
"""验证运行配置"""
if v.max_iters <= 0:
raise ValueError("max_iters must be positive")
if v.max_iters > 100:
raise ValueError("max_iters cannot exceed 100")
return v
class ModelConfig(BaseModel):
"""模型配置验证"""
provider_id: str
model: str
@validator("provider_id")
def validate_provider(cls, v):
"""验证提供商"""
valid_providers = ["openai", "dashscope", "anthropic", "gemini"]
if v not in valid_providers:
raise ValueError(
f"Invalid provider: {v}. "
f"Valid providers: {valid_providers}"
)
return v
渠道配置
@dataclass
class ChannelConfig:
"""渠道配置基类"""
channel_id: str
enabled: bool = True
name: str = ""
@dataclass
class DingTalkConfig(ChannelConfig):
"""钉钉配置"""
app_key: str = ""
app_secret: str = ""
agent_id: str = ""
webhook_url: str | None = None
@dataclass
class FeishuConfig(ChannelConfig):
"""飞书配置"""
app_id: str = ""
app_secret: str = ""
bot_id: str = ""
@dataclass
class DiscordConfig(ChannelConfig):
"""Discord配置"""
bot_token: str = ""
guild_id: str = ""
channel_ids: list[str] = field(default_factory=list)
渠道加载器
class ChannelConfigLoader:
"""渠道配置加载器"""
def __init__(self, working_dir: Path):
self.working_dir = working_dir
self.channel_dir = working_dir / "channels"
async def load_all_channels(self) -> dict[str, ChannelConfig]:
"""加载所有渠道配置"""
channels = {}
for config_file in self.channel_dir.glob("*.yaml"):
channel_type = config_file.stem
try:
config = await self._load_channel_config(channel_type, config_file)
channels[channel_type] = config
except Exception as e:
logger.warning(f"Failed to load channel {channel_type}: {e}")
return channels
async def _load_channel_config(
self,
channel_type: str,
config_file: Path,
) -> ChannelConfig:
"""加载单个渠道配置"""
config_dict = yaml.safe_load(config_file.read_text())
# 根据渠道类型创建配置对象
if channel_type == "dingtalk":
return DingTalkConfig(**config_dict)
elif channel_type == "feishu":
return FeishuConfig(**config_dict)
elif channel_type == "discord":
return DiscordConfig(**config_dict)
else:
return ChannelConfig(**config_dict)
配置热更新
class ConfigWatcher:
"""配置文件监视器，支持热更新"""
def __init__(self, working_dir: Path):
self.working_dir = working_dir
self._watchers: dict[Path, asyncio.Task] = {}
self._callbacks: dict[str, Callable] = {}
async def watch(
self,
config_path: Path,
callback: Callable,
) -> None:
"""监视配置文件变化"""
async with aiofiles.watch(config_path) as changes:
self._callbacks[str(config_path)] = callback
async for change in changes:
if change.type == "modified":
logger.info(f"Config file changed: {config_path}")
await self._handle_change(config_path)
async def _handle_change(self, config_path: Path) -> None:
"""处理配置变化"""
callback = self._callbacks.get(str(config_path))
if not callback:
return
try:
# 重新加载配置
new_config = self._load_config(config_path)
# 触发回调
await callback(new_config)
except Exception as e:
logger.error(f"Failed to handle config change: {e}")
配置备份与恢复
class ConfigBackupManager:
"""配置备份管理器"""
def __init__(self, working_dir: Path):
self.working_dir = working_dir
self.backup_dir = working_dir / ".secret.backups"
async def create_backup(self, description: str = "") -> str:
"""创建配置备份"""
if not self.backup_dir.exists():
self.backup_dir.mkdir(parents=True)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_name = f"backup_{timestamp}"
backup_path = self.backup_dir / backup_name
# 复制配置文件
shutil.copytree(self.working_dir / ".secret", backup_path)
# 记录备份元数据
metadata = {
"timestamp": timestamp,
"description": description,
"files": list(self.backup_dir.iterdir()),
}
metadata_file = backup_path / "metadata.json"
metadata_file.write_text(json.dumps(metadata, indent=2))
return backup_name
async def restore_backup(self, backup_name: str) -> None:
"""恢复配置备份"""
backup_path = self.backup_dir / backup_name
if not backup_path.exists():
raise ValueError(f"Backup not found: {backup_name}")
# 备份当前配置
await self.create_backup("Auto-backup before restore")
# 恢复配置
current_secret = self.working_dir / ".secret"
if current_secret.exists():
shutil.rmtree(current_secret)
shutil.copytree(backup_path, current_secret)
logger.info(f"Restored backup: {backup_name}")
多智能体配置
@dataclass
class MultiAgentConfig:
"""多智能体配置"""
agents: list[AgentProfileConfig] # 智能体列表
collaboration: CollaborationConfig # 协作配置
@classmethod
def from_directory(cls, config_dir: Path) -> "MultiAgentConfig":
"""从目录加载多智能体配置"""
agents = []
for config_file in config_dir.glob("agents/*.yaml"):
agent_config = cls._load_agent_config(config_file)
agents.append(agent_config)
collaboration_file = config_dir / "collaboration.yaml"
if collaboration_file.exists():
collaboration = cls._load_collaboration_config(collaboration_file)
else:
collaboration = CollaborationConfig()
return cls(agents=agents, collaboration=collaboration)
@dataclass
class CollaborationConfig:
"""智能体协作配置"""
enabled: bool = True
protocols: list[str] = field(default_factory=lambda: ["acp"])
max_concurrent_tasks: int = 5
总结
QwenPaw的配置系统设计原则：
通过这套配置系统，QwenPaw能够适应不同的使用场景和用户需求。
往期回顾：
下期预告：
如果对你有帮助，欢迎点赞、在看！
