403 lines
16 KiB
Python
403 lines
16 KiB
Python
"""Configuration and loaders for custom agents."""
|
|
|
|
import logging
|
|
import os
|
|
import re
|
|
import threading
|
|
from typing import Any
|
|
from urllib.parse import urlsplit
|
|
|
|
import yaml
|
|
from pydantic import BaseModel, Field, 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
|
|
|
|
|
|
_CHAT_LINK_URL_MAX = 2048
|
|
_CHAT_LINK_PARAM_MAX = 30
|
|
|
|
|
|
class AgentChatRouteParam(BaseModel):
|
|
"""One query parameter appended when an agent's gallery 「对话」 button opens its link.
|
|
|
|
Same shape as a light-app route param: login values are resolved in the
|
|
browser from the current user, theme values follow the active theme.
|
|
"""
|
|
|
|
key: str = ""
|
|
paramType: str = "login"
|
|
source: str | None = None
|
|
customValue: str | None = None
|
|
themeValue: str | None = None
|
|
|
|
|
|
class AgentChatLinkConfig(BaseModel):
|
|
"""Optional external (or in-app) destination for the gallery chat button."""
|
|
|
|
url: str = ""
|
|
open_mode: str = "blank"
|
|
route_params: list[AgentChatRouteParam] = Field(default_factory=list)
|
|
|
|
|
|
def _is_allowed_agent_chat_url(url: str) -> bool:
|
|
lowered = url.lower()
|
|
if lowered.startswith(("javascript:", "data:", "vbscript:", "file:")):
|
|
return False
|
|
if url.startswith("#/"):
|
|
return True
|
|
if url.startswith("/") and not url.startswith("//") and not url.startswith("/\\"):
|
|
return True
|
|
parsed = urlsplit(url)
|
|
return parsed.scheme in {"http", "https"} and bool(parsed.netloc)
|
|
|
|
|
|
def normalize_agent_chat_link(value: Any, *, strict: bool = False) -> dict[str, Any] | None:
|
|
"""Return a clean chat-link dict, or None when the link is unset.
|
|
|
|
``strict=True`` raises ``ValueError`` for a non-empty but illegal link so
|
|
the API can reject it. Disk loads stay lenient: a bad value is dropped
|
|
instead of making the whole agent unloadable.
|
|
"""
|
|
|
|
if value is None:
|
|
return None
|
|
if hasattr(value, "model_dump"):
|
|
value = value.model_dump()
|
|
if not isinstance(value, dict):
|
|
if strict:
|
|
raise ValueError("对话跳转配置格式不正确")
|
|
return None
|
|
|
|
url = str(value.get("url") or "").strip()
|
|
if not url:
|
|
return None
|
|
if len(url) > _CHAT_LINK_URL_MAX or not _is_allowed_agent_chat_url(url):
|
|
if strict:
|
|
raise ValueError("对话链接只支持 http(s) 地址，或以 /、#/ 开头的站内路径")
|
|
return None
|
|
|
|
mode = str(value.get("open_mode") or "blank").strip() or "blank"
|
|
if mode not in {"self", "blank"}:
|
|
if strict:
|
|
raise ValueError("打开方式只支持当前窗口或新窗口")
|
|
mode = "blank"
|
|
|
|
raw_params = value.get("route_params") or []
|
|
if raw_params is None:
|
|
raw_params = []
|
|
if not isinstance(raw_params, list):
|
|
if strict:
|
|
raise ValueError("路由参数格式不正确")
|
|
raw_params = []
|
|
if len(raw_params) > _CHAT_LINK_PARAM_MAX:
|
|
if strict:
|
|
raise ValueError("路由参数最多 30 个")
|
|
raw_params = raw_params[:_CHAT_LINK_PARAM_MAX]
|
|
|
|
params: list[dict[str, Any]] = []
|
|
for item in raw_params:
|
|
if hasattr(item, "model_dump"):
|
|
item = item.model_dump()
|
|
if not isinstance(item, dict):
|
|
if strict:
|
|
raise ValueError("路由参数格式不正确")
|
|
continue
|
|
key = str(item.get("key") or "").strip()
|
|
if not key:
|
|
if strict:
|
|
raise ValueError("路由参数键名不可为空")
|
|
continue
|
|
if len(key) > 128:
|
|
if strict:
|
|
raise ValueError("路由参数键名过长")
|
|
continue
|
|
param_type = str(item.get("paramType") or "login").strip() or "login"
|
|
if param_type not in {"login", "theme"}:
|
|
if strict:
|
|
raise ValueError("路由参数类型只支持登录参数或主题参数")
|
|
continue
|
|
source = None if item.get("source") is None else str(item.get("source"))[:64]
|
|
custom = None if item.get("customValue") is None else str(item.get("customValue"))[:2048]
|
|
theme = None if item.get("themeValue") is None else str(item.get("themeValue"))[:2048]
|
|
params.append(
|
|
{
|
|
"key": key,
|
|
"paramType": param_type,
|
|
"source": source if param_type == "login" else None,
|
|
"customValue": custom if param_type == "login" and source == "other" else None,
|
|
"themeValue": theme if param_type == "theme" else None,
|
|
}
|
|
)
|
|
return {"url": url, "open_mode": mode, "route_params": params}
|
|
|
|
|
|
class AgentConfig(BaseModel):
|
|
"""Configuration for a custom agent."""
|
|
|
|
id: str | None = None
|
|
name: str
|
|
description: str = ""
|
|
model: str | None = None
|
|
# Whether this agent inherits the administrator-configured prompt suffix
|
|
# for its selected model. Existing configs omit the field, so True keeps
|
|
# their current behavior without a migration.
|
|
model_prompt_suffix_enabled: bool = True
|
|
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
|
|
# Gallery 「对话」 button destination. None / empty url keeps the in-app chat.
|
|
# ``open_mode`` is ``blank`` (new window) or ``self`` (current window).
|
|
# ``route_params`` match light-app route params and are resolved in the browser.
|
|
chat_link: dict[str, Any] | None = None
|
|
|
|
@field_validator("chat_link", mode="before")
|
|
@classmethod
|
|
def _normalize_chat_link(cls, value: Any) -> dict[str, Any] | None:
|
|
return normalize_agent_chat_link(value, strict=False)
|
|
|
|
@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
|