151 lines
4.8 KiB
Python
151 lines
4.8 KiB
Python
"""Configuration and loaders for custom agents."""
|
|
|
|
import logging
|
|
import re
|
|
from typing import Any
|
|
|
|
import yaml
|
|
from pydantic import BaseModel
|
|
|
|
from deerflow.config.paths import get_paths
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
SOUL_FILENAME = "SOUL.md"
|
|
AGENT_ID_PATTERN = re.compile(r"^[A-Za-z0-9_-]+$")
|
|
# Backward-compatible alias. Historically this validated the display name;
|
|
# after the DB-backed agent table, display names can be Chinese and duplicated.
|
|
# The pattern now applies only to storage/runtime ids.
|
|
AGENT_NAME_PATTERN = AGENT_ID_PATTERN
|
|
|
|
|
|
def validate_agent_id(agent_id: str | None) -> str | None:
|
|
"""Validate a custom agent id before using it in filesystem paths."""
|
|
if agent_id is None:
|
|
return None
|
|
if not isinstance(agent_id, str):
|
|
raise ValueError("Invalid agent id. Expected a string or None.")
|
|
if not AGENT_ID_PATTERN.fullmatch(agent_id):
|
|
raise ValueError(f"Invalid agent id '{agent_id}'. Must match pattern: {AGENT_ID_PATTERN.pattern}")
|
|
return agent_id
|
|
|
|
|
|
def validate_agent_name(name: str | None) -> str | None:
|
|
"""Backward-compatible alias for validating a storage/runtime agent id."""
|
|
return validate_agent_id(name)
|
|
|
|
|
|
class AgentConfig(BaseModel):
|
|
"""Configuration for a custom agent."""
|
|
|
|
id: str | None = None
|
|
name: str
|
|
description: str = ""
|
|
model: str | None = None
|
|
tool_groups: list[str] | None = None
|
|
# skills controls which skills are loaded into a custom agent's prompt:
|
|
# - None (or omitted): no skills (custom agents require explicit selection)
|
|
# - [] (explicit empty list): no skills
|
|
# - ["skill1", "skill2"]: load only the specified skills
|
|
skills: list[str] | None = None
|
|
|
|
|
|
def load_agent_config(agent_id: str | None) -> AgentConfig | None:
|
|
"""Load the custom or default agent's config from its directory.
|
|
|
|
Args:
|
|
agent_id: The agent storage/runtime id.
|
|
|
|
Returns:
|
|
AgentConfig instance.
|
|
|
|
Raises:
|
|
FileNotFoundError: If the agent directory or config.yaml does not exist.
|
|
ValueError: If config.yaml cannot be parsed.
|
|
"""
|
|
|
|
if agent_id is None:
|
|
return None
|
|
|
|
agent_id = validate_agent_id(agent_id)
|
|
agent_dir = get_paths().agent_dir(agent_id)
|
|
config_file = agent_dir / "config.yaml"
|
|
|
|
if not agent_dir.exists():
|
|
raise FileNotFoundError(f"Agent directory not found: {agent_dir}")
|
|
|
|
if not config_file.exists() and not (agent_dir / SOUL_FILENAME).exists():
|
|
raise FileNotFoundError(f"Agent config not found: {config_file}")
|
|
|
|
data: dict[str, Any] = {}
|
|
if config_file.exists():
|
|
try:
|
|
with open(config_file, encoding="utf-8") as f:
|
|
data = yaml.safe_load(f) or {}
|
|
except yaml.YAMLError as e:
|
|
raise ValueError(f"Failed to parse agent config {config_file}: {e}") from e
|
|
|
|
# Ensure id/name are set from directory id if not in file.
|
|
if "id" not in data:
|
|
data["id"] = agent_id
|
|
if "name" not in data:
|
|
data["name"] = agent_id
|
|
|
|
# Strip unknown fields before passing to Pydantic (e.g. legacy prompt_file)
|
|
known_fields = set(AgentConfig.model_fields.keys())
|
|
data = {k: v for k, v in data.items() if k in known_fields}
|
|
|
|
return AgentConfig(**data)
|
|
|
|
|
|
def load_agent_soul(agent_id: str | None) -> str | None:
|
|
"""Read the SOUL.md file for a custom agent, if it exists.
|
|
|
|
SOUL.md defines the agent's personality, values, and behavioral guardrails.
|
|
It is injected into the lead agent's system prompt as additional context.
|
|
|
|
Args:
|
|
agent_id: The storage/runtime id of the agent or None for the default agent.
|
|
|
|
Returns:
|
|
The SOUL.md content as a string, or None if the file does not exist.
|
|
"""
|
|
agent_dir = get_paths().agent_dir(validate_agent_id(agent_id)) if agent_id else get_paths().base_dir
|
|
soul_path = agent_dir / SOUL_FILENAME
|
|
if not soul_path.exists():
|
|
return None
|
|
content = soul_path.read_text(encoding="utf-8").strip()
|
|
return content or None
|
|
|
|
|
|
def list_custom_agents() -> list[AgentConfig]:
|
|
"""Scan the agents directory and return all valid custom agents.
|
|
|
|
Returns:
|
|
List of AgentConfig for each valid agent directory found.
|
|
"""
|
|
agents_dir = get_paths().agents_dir
|
|
|
|
if not agents_dir.exists():
|
|
return []
|
|
|
|
agents: list[AgentConfig] = []
|
|
|
|
for entry in sorted(agents_dir.iterdir()):
|
|
if not entry.is_dir():
|
|
continue
|
|
|
|
config_file = entry / "config.yaml"
|
|
soul_file = entry / SOUL_FILENAME
|
|
if not config_file.exists() and not soul_file.exists():
|
|
logger.debug(f"Skipping {entry.name}: no config.yaml or SOUL.md")
|
|
continue
|
|
|
|
try:
|
|
agent_cfg = load_agent_config(entry.name)
|
|
agents.append(agent_cfg)
|
|
except Exception as e:
|
|
logger.warning(f"Skipping agent '{entry.name}': {e}")
|
|
|
|
return agents
|