271 lines
10 KiB
Python
271 lines
10 KiB
Python
"""Configuration and loaders for custom agents."""
|
||
|
||
import logging
|
||
import os
|
||
import re
|
||
import threading
|
||
from typing import Any
|
||
|
||
import yaml
|
||
from pydantic import BaseModel, field_validator, model_validator
|
||
|
||
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_-]+$")
|
||
|
||
# Per-file mtime caches. These are read on every /api/agents list/detail request
|
||
# (one read per agent) — caching keyed by (path, mtime_ns, size) is correct
|
||
# because any edit to the file updates mtime, and an external tool re-writing
|
||
# the same bytes doesn't actually change the result we'd return. Cache size is
|
||
# unbounded but lookup is O(1) and entries are tiny (one parsed dict / string
|
||
# per agent file); for the order of N≤10⁴ agents the memory cost is negligible.
|
||
_config_cache: dict[str, tuple[tuple[int, int], "AgentConfig | None"]] = {}
|
||
_soul_cache: dict[str, tuple[tuple[int, int], str | None]] = {}
|
||
_cache_lock = threading.Lock()
|
||
|
||
|
||
def _file_stat_key(path) -> tuple[int, int] | None:
|
||
"""Return (mtime_ns, size) for cache invalidation, or None when missing."""
|
||
try:
|
||
st = os.stat(path)
|
||
except OSError:
|
||
return None
|
||
return (st.st_mtime_ns, st.st_size)
|
||
# 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 CommonQuestion(BaseModel):
|
||
"""A recommended question shown on the agent chat home / sidebar.
|
||
|
||
``title`` is the short label rendered on the question chip; ``prompt`` is the
|
||
full message actually sent when the user clicks it. Splitting the two lets an
|
||
admin keep the chip terse while still sending a richly-worded question.
|
||
|
||
Legacy configs stored questions as bare strings (content only). The
|
||
before-validator coerces such a string into ``{title, prompt}`` so old
|
||
config.yaml files keep loading without a migration — the title falls back to
|
||
a truncated copy of the content.
|
||
"""
|
||
|
||
title: str = ""
|
||
prompt: str = ""
|
||
|
||
@model_validator(mode="before")
|
||
@classmethod
|
||
def _coerce_legacy_string(cls, value: Any) -> Any:
|
||
if isinstance(value, str):
|
||
text = value.strip()
|
||
title = text if len(text) <= 40 else text[:40].rstrip() + "…"
|
||
return {"title": title, "prompt": text}
|
||
return value
|
||
|
||
|
||
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
|
||
# Frequently-asked questions ("常用问题") shown in the agent chat sidebar.
|
||
# Display/UX metadata only — the runtime never reads these. None = none set.
|
||
# Accepts both the legacy ``list[str]`` shape and the new ``{title, prompt}``
|
||
# objects (see CommonQuestion).
|
||
common_questions: list[CommonQuestion] | None = None
|
||
# 入库模板名（外部知识入库接口的 templateName）。配置后该智能体会话中的
|
||
# Markdown 产物可一键发布到外部入库接口；None/空 = 不支持入库。
|
||
template_name: str | None = None
|
||
# 会商席位「技能识别强化」开关（弱模型补偿）。作为会商席位运行时，是否把本智能体
|
||
# 配置的技能（名称 + SKILL.md 路径 + 必须先 read_file 再按其流程取信息的硬指令）显式
|
||
# 拼进本轮任务，避免弱模型无视系统提示里的技能块。None/False = 关（默认）；True = 开。
|
||
# 与「业务链条配置」里的同名开关取 OR：任一打开即对该席位生效。见
|
||
# ``app/gateway/roundtable_seat_skills.py``。
|
||
seat_skill_directive: bool | None = None
|
||
# DeerFlow mapping ids this agent may offer in its new-conversation
|
||
# knowledge selector. WeKnora ids and credentials never enter agent files.
|
||
llmwiki_knowledge_base_ids: list[str] | None = None
|
||
|
||
@field_validator("llmwiki_knowledge_base_ids")
|
||
@classmethod
|
||
def _normalize_llmwiki_ids(cls, value: list[str] | None) -> list[str] | None:
|
||
if value is None:
|
||
return None
|
||
normalized = list(dict.fromkeys(str(item).strip() for item in value if str(item).strip()))
|
||
if len(normalized) > 100:
|
||
raise ValueError("An agent can bind at most 100 LLMWiki knowledge bases")
|
||
return normalized
|
||
|
||
|
||
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, or ``None`` when the on-disk artifacts are missing.
|
||
|
||
Raises:
|
||
ValueError: If config.yaml exists but cannot be parsed.
|
||
|
||
Tolerance:
|
||
A missing agent directory or missing config.yaml is **not** an error —
|
||
the runtime simply falls back to default Lead Agent behavior for that
|
||
id. This keeps the chat session usable when the DB still has an agent
|
||
row but its files have been deleted (e.g. data dir moved between
|
||
machines, or filesystem wiped). The router layer is expected to
|
||
lazily restore the directory next time the agent is fetched via API.
|
||
"""
|
||
|
||
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():
|
||
logger.warning(
|
||
"Agent directory missing for id '%s' (%s); falling back to default config",
|
||
agent_id,
|
||
agent_dir,
|
||
)
|
||
return None
|
||
|
||
if not config_file.exists() and not (agent_dir / SOUL_FILENAME).exists():
|
||
logger.warning(
|
||
"Agent has no config.yaml or %s at %s; falling back to default config",
|
||
SOUL_FILENAME,
|
||
agent_dir,
|
||
)
|
||
return None
|
||
|
||
# mtime-based cache. The config file is rewritten every time the agent is
|
||
# edited via PUT /api/agents/{id}, which bumps mtime — so the next read
|
||
# picks up the change automatically. Missing config_file = key None below,
|
||
# bypassing the cache (cheap path: just return AgentConfig with id/name).
|
||
cache_path_key = str(config_file)
|
||
stat_key = _file_stat_key(config_file) if config_file.exists() else None
|
||
if stat_key is not None:
|
||
with _cache_lock:
|
||
cached = _config_cache.get(cache_path_key)
|
||
if cached is not None and cached[0] == stat_key:
|
||
return cached[1]
|
||
|
||
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}
|
||
|
||
result = AgentConfig(**data)
|
||
if stat_key is not None:
|
||
with _cache_lock:
|
||
_config_cache[cache_path_key] = (stat_key, result)
|
||
return result
|
||
|
||
|
||
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
|
||
# mtime-based cache, same rationale as load_agent_config: SOUL.md is
|
||
# rewritten on edit (PUT /api/agents/{id}), which bumps mtime.
|
||
cache_path_key = str(soul_path)
|
||
stat_key = _file_stat_key(soul_path)
|
||
if stat_key is not None:
|
||
with _cache_lock:
|
||
cached = _soul_cache.get(cache_path_key)
|
||
if cached is not None and cached[0] == stat_key:
|
||
return cached[1]
|
||
content = soul_path.read_text(encoding="utf-8").strip()
|
||
result = content or None
|
||
if stat_key is not None:
|
||
with _cache_lock:
|
||
_soul_cache[cache_path_key] = (stat_key, result)
|
||
return result
|
||
|
||
|
||
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
|