380 lines
15 KiB
Python
380 lines
15 KiB
Python
"""
|
|
config.py — AIPA Unified Configuration
|
|
|
|
Loads and merges configuration from two sources:
|
|
1. config/app.yaml — application structure, paths, infrastructure defaults
|
|
2. config/agents.yaml — provider definitions and agent assignments
|
|
|
|
Environment variables override config file values using the AIPA_ prefix.
|
|
Secrets (API keys) are loaded from .env and never stored in config files.
|
|
|
|
Path resolution order:
|
|
1. If AIPA_ROOT is set, relative paths resolve against it.
|
|
2. Otherwise, relative paths resolve against the repository root.
|
|
3. Absolute paths are always used as-is.
|
|
|
|
Docker pattern:
|
|
ENV AIPA_ROOT=/app/data
|
|
VOLUME /app/data # mounts standing_brief, archive, logs, data
|
|
COPY config/ /app/config/
|
|
COPY agents/ /app/agents/
|
|
"""
|
|
|
|
import os
|
|
import warnings
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
from dotenv import load_dotenv
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Bootstrap paths (before config is loaded)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_REPO_ROOT = Path(__file__).parent.parent
|
|
load_dotenv(_REPO_ROOT / ".env")
|
|
|
|
# AIPA_ROOT relocates all relative paths — set this in Docker.
|
|
_AIPA_ROOT = Path(os.getenv("AIPA_ROOT", "")) if os.getenv("AIPA_ROOT") else _REPO_ROOT
|
|
|
|
|
|
def _resolve(raw: str) -> Path:
|
|
"""Resolve a path from app.yaml: absolute paths as-is, relative to _AIPA_ROOT."""
|
|
p = Path(raw)
|
|
return p if p.is_absolute() else _AIPA_ROOT / p
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Load app.yaml
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_APP_CONFIG_PATH = Path(os.getenv("AIPA_APP_CONFIG", str(_REPO_ROOT / "config" / "app.yaml")))
|
|
|
|
if not _APP_CONFIG_PATH.exists():
|
|
raise FileNotFoundError(
|
|
f"Application config not found: {_APP_CONFIG_PATH}\n"
|
|
f"Expected at config/app.yaml relative to the repo root."
|
|
)
|
|
|
|
with open(_APP_CONFIG_PATH, encoding="utf-8") as _f:
|
|
_app: dict = yaml.safe_load(_f)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Load agents.yaml
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_AGENTS_CONFIG_PATH = Path(
|
|
os.getenv("AIPA_AGENTS_CONFIG",
|
|
str(_resolve(_app["paths"].get("agents_config", "config/agents.yaml"))))
|
|
)
|
|
|
|
if not _AGENTS_CONFIG_PATH.exists():
|
|
raise FileNotFoundError(
|
|
f"Agents config not found: {_AGENTS_CONFIG_PATH}\n"
|
|
f"Expected at config/agents.yaml relative to the repo root."
|
|
)
|
|
|
|
with open(_AGENTS_CONFIG_PATH, encoding="utf-8") as _f:
|
|
_agents_cfg: dict = yaml.safe_load(_f)
|
|
|
|
_providers: dict = _agents_cfg.get("providers", {})
|
|
_agents: dict = _agents_cfg.get("agents", {})
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# App identity
|
|
# ---------------------------------------------------------------------------
|
|
|
|
APP_NAME: str = os.getenv("AIPA_APP_NAME", _app["app"]["name"])
|
|
APP_VERSION: str = _app["app"]["version"]
|
|
APP_ENV: str = os.getenv("AIPA_ENV", _app["app"]["environment"])
|
|
LOG_LEVEL: str = os.getenv("AIPA_LOG_LEVEL", _app["app"]["log_level"])
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Resolved paths
|
|
# ---------------------------------------------------------------------------
|
|
|
|
PROMPTS_DIR: Path = _resolve(os.getenv("AIPA_PROMPTS_DIR",
|
|
_app["paths"]["prompts_dir"]))
|
|
|
|
REGISTRY_PATH: Path = _resolve(os.getenv("AIPA_REGISTRY_PATH",
|
|
_app["paths"]["registry"]))
|
|
|
|
STANDING_BRIEF_PATH: Path = _resolve(os.getenv("AIPA_STANDING_BRIEF_PATH",
|
|
_app["paths"]["standing_brief"]))
|
|
|
|
BRIEF_TEMPLATE_PATH: Path = _resolve(os.getenv("AIPA_BRIEF_TEMPLATE_PATH",
|
|
_app["paths"]["brief_template"]))
|
|
|
|
ARCHIVE_DIR: Path = _resolve(os.getenv("AIPA_ARCHIVE_DIR",
|
|
_app["paths"]["archive_dir"]))
|
|
|
|
LOGS_DIR: Path = _resolve(os.getenv("AIPA_LOGS_DIR",
|
|
_app["paths"]["logs_dir"]))
|
|
|
|
DATA_DIR: Path = _resolve(os.getenv("AIPA_DATA_DIR",
|
|
_app["paths"]["data_dir"]))
|
|
|
|
TASKS_DIR: Path = _resolve(os.getenv("AIPA_TASKS_DIR",
|
|
_app["paths"].get("tasks_dir", "data/tasks")))
|
|
|
|
# Derived convenience path — standing brief archive file
|
|
ARCHIVE_PATH: Path = ARCHIVE_DIR / "standing_brief_archive.md"
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Logging
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_log_cfg = _app.get("logging", {})
|
|
|
|
LOG_FORMAT: str = os.getenv("AIPA_LOG_FORMAT", _log_cfg.get("format", "rich"))
|
|
|
|
_log_file_cfg = _log_cfg.get("file", {})
|
|
LOG_FILE_ENABLED: bool = os.getenv("AIPA_LOG_FILE_ENABLED", str(_log_file_cfg.get("enabled", False))).lower() == "true"
|
|
LOG_FILE_PATH: Path = _resolve(os.getenv("AIPA_LOG_FILE_PATH", _log_file_cfg.get("path", "logs/aipa.log")))
|
|
LOG_ROTATION: str = os.getenv("AIPA_LOG_ROTATION", _log_file_cfg.get("rotation", "daily"))
|
|
LOG_RETENTION_DAYS: int = int(os.getenv("AIPA_LOG_RETENTION_DAYS", str(_log_file_cfg.get("retention_days", 30))))
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Session defaults
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_sess = _app.get("session", {})
|
|
|
|
SESSION_LOG_RETENTION: int = int(os.getenv("AIPA_SESSION_LOG_RETENTION",
|
|
str(_sess.get("log_retention", 10))))
|
|
LEAD_TIMEOUT_SECONDS: int = int(os.getenv("AIPA_LEAD_TIMEOUT",
|
|
str(_sess.get("lead_timeout", 120))))
|
|
VERA_AUTO_AUDIT: bool = os.getenv("AIPA_AUTO_AUDIT",
|
|
str(_sess.get("auto_audit", True))).lower() != "false"
|
|
DEBUG_PRINT_AGENT_OUTPUTS: bool = os.getenv("AIPA_DEBUG",
|
|
str(_sess.get("debug_output", False))).lower() == "true"
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Memory
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_mem = _app.get("memory", {})
|
|
_mem_store = _mem.get("store", {})
|
|
|
|
MEMORY_BRIEF_ENABLED: bool = os.getenv("AIPA_MEMORY_BRIEF_ENABLED",
|
|
str(_mem.get("standing_brief", {}).get("enabled", True))).lower() == "true"
|
|
MEMORY_STORE_ENABLED: bool = os.getenv("AIPA_MEMORY_STORE_ENABLED",
|
|
str(_mem_store.get("enabled", False))).lower() == "true"
|
|
MEMORY_STORE_BACKEND: str = os.getenv("AIPA_MEMORY_STORE_BACKEND",
|
|
_mem_store.get("backend", "none"))
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Infrastructure
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_infra = _app.get("infrastructure", {})
|
|
_api = _infra.get("api", {})
|
|
_cache = _infra.get("cache", {})
|
|
|
|
API_ENABLED: bool = os.getenv("AIPA_API_ENABLED", str(_api.get("enabled", False))).lower() == "true"
|
|
API_HOST: str = os.getenv("AIPA_API_HOST", _api.get("host", "0.0.0.0"))
|
|
API_PORT: int = int(os.getenv("AIPA_API_PORT", str(_api.get("port", 8080))))
|
|
API_WORKERS: int = int(os.getenv("AIPA_API_WORKERS", str(_api.get("workers", 1))))
|
|
|
|
CACHE_ENABLED: bool = os.getenv("AIPA_CACHE_ENABLED", str(_cache.get("enabled", False))).lower() == "true"
|
|
CACHE_BACKEND: str = os.getenv("AIPA_CACHE_BACKEND", _cache.get("backend", "memory"))
|
|
CACHE_TTL: int = int(os.getenv("AIPA_CACHE_TTL", str(_cache.get("ttl_seconds", 3600))))
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Provider resolution (from agents.yaml)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _provider_def(provider_name: str) -> dict:
|
|
"""Return the provider definition for a named provider."""
|
|
if not provider_name or provider_name == "none":
|
|
return {"type": "none"}
|
|
pdef = _providers.get(provider_name)
|
|
if pdef is None:
|
|
raise ValueError(
|
|
f"Unknown provider: {provider_name!r}. "
|
|
f"Define it in config/agents.yaml under 'providers'."
|
|
)
|
|
return pdef
|
|
|
|
|
|
def agent_provider_name(agent: str) -> str:
|
|
"""Named provider assigned to an agent (e.g. 'anthropic', 'lmstudio')."""
|
|
adef = _agents.get(agent, {})
|
|
per_agent = adef.get("provider", "") or ""
|
|
# Treat empty string or the placeholder "none" as "not configured — use default"
|
|
if per_agent and per_agent != "none":
|
|
return per_agent
|
|
return _agents_cfg.get("default_provider", "none")
|
|
|
|
|
|
def agent_provider_type(agent: str) -> str:
|
|
"""Provider type: anthropic | openai | openai_compatible | ollama | none."""
|
|
return _provider_def(agent_provider_name(agent)).get("type", "none")
|
|
|
|
|
|
def agent_api_key(agent: str) -> str:
|
|
"""Resolve the API key for an agent's provider from .env."""
|
|
pdef = _provider_def(agent_provider_name(agent))
|
|
if "api_key_env" in pdef:
|
|
key = os.getenv(pdef["api_key_env"], "")
|
|
if not key:
|
|
warnings.warn(
|
|
f"Environment variable {pdef['api_key_env']!r} is not set "
|
|
f"(required for agent '{agent}'s provider)."
|
|
)
|
|
return key
|
|
return pdef.get("api_key", "local")
|
|
|
|
|
|
def agent_base_url(agent: str) -> str | None:
|
|
"""Resolve the base URL for an agent's provider."""
|
|
pname = agent_provider_name(agent)
|
|
pdef = _provider_def(pname)
|
|
url = pdef.get("base_url")
|
|
if not url and pdef.get("type") == "openai_compatible":
|
|
raise ValueError(
|
|
f"Provider '{pname}' is type openai_compatible but has no base_url. "
|
|
f"Add a base_url to its entry in config/agents.yaml."
|
|
)
|
|
return url or None
|
|
|
|
|
|
def agent_extra_body(agent: str) -> dict:
|
|
"""Extra request body fields for an agent's provider (e.g. enable_thinking: false)."""
|
|
pdef = _provider_def(agent_provider_name(agent))
|
|
return pdef.get("extra_body") or {}
|
|
|
|
|
|
def agent_model(agent: str) -> str:
|
|
"""
|
|
Model name for an agent.
|
|
Uses the agent's own model if set; falls back to the provider's default_model.
|
|
"""
|
|
per_agent = _agents.get(agent, {}).get("model", "") or ""
|
|
if per_agent:
|
|
return per_agent
|
|
pdef = _provider_def(agent_provider_name(agent))
|
|
return pdef.get("default_model", "")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Role resolution (from agents.yaml `roles` section)
|
|
# The orchestration code references roles only — never agent names directly.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_roles: dict = _agents_cfg.get("roles", {})
|
|
|
|
ORCHESTRATOR_AGENT: str = _roles.get("orchestrator", "")
|
|
AUDITOR_AGENT: str = _roles.get("auditor", "")
|
|
RECRUITER_AGENT: str = _roles.get("recruiter", "")
|
|
SUMMARIZER_AGENT: str = _roles.get("summarizer", "")
|
|
|
|
if not ORCHESTRATOR_AGENT:
|
|
raise ValueError("No 'orchestrator' role defined in config/agents.yaml under 'roles'.")
|
|
if not AUDITOR_AGENT:
|
|
raise ValueError("No 'auditor' role defined in config/agents.yaml under 'roles'.")
|
|
|
|
# Only the orchestrator, auditor, and summarizer are non-dispatchable. All other
|
|
# roles (including recruiter) are leads and can receive TASK BRIEFs.
|
|
_SYSTEM_AGENTS: set[str] = {ORCHESTRATOR_AGENT, AUDITOR_AGENT, SUMMARIZER_AGENT} - {""}
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Agent and lead definitions (derived from agents.yaml)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def agent_title(agent: str) -> str:
|
|
"""Human-readable title for an agent, from agents.yaml."""
|
|
return _agents.get(agent, {}).get("title", agent.capitalize())
|
|
|
|
|
|
def agent_color(agent: str) -> str:
|
|
"""Display color for an agent, from agents.yaml. Defaults to white."""
|
|
return _agents.get(agent, {}).get("color", "white")
|
|
|
|
|
|
# Full agent config dict used by build_agent() and /agents display.
|
|
AGENT_CONFIGS: dict[str, dict] = {
|
|
name: {
|
|
"prompt_file": adef["prompt_file"],
|
|
"temperature": adef["temperature"],
|
|
"max_tokens": adef["max_tokens"],
|
|
"stateful": adef["stateful"],
|
|
"tools": list(adef.get("tools") or []),
|
|
}
|
|
for name, adef in _agents.items()
|
|
}
|
|
|
|
|
|
def agent_tools(agent: str) -> list[str]:
|
|
"""List of tool names assigned to an agent (empty list if none)."""
|
|
return AGENT_CONFIGS.get(agent, {}).get("tools", [])
|
|
|
|
|
|
def reload_agents_config() -> None:
|
|
"""
|
|
Re-read agents.yaml and refresh all derived module-level values.
|
|
Call after upsert_agent_definition() to pick up changes in the running session.
|
|
"""
|
|
global _agents_cfg, _providers, _agents
|
|
global AGENT_CONFIGS, LEAD_AGENT_NAMES, ACTIVE_PROVIDER
|
|
global ORCHESTRATOR_AGENT, AUDITOR_AGENT, RECRUITER_AGENT
|
|
global _roles, _SYSTEM_AGENTS
|
|
|
|
with open(_AGENTS_CONFIG_PATH, encoding="utf-8") as fh:
|
|
_agents_cfg = yaml.safe_load(fh)
|
|
|
|
_providers = _agents_cfg.get("providers", {})
|
|
_agents = _agents_cfg.get("agents", {})
|
|
|
|
AGENT_CONFIGS = {
|
|
name: {
|
|
"prompt_file": adef["prompt_file"],
|
|
"temperature": adef["temperature"],
|
|
"max_tokens": adef["max_tokens"],
|
|
"stateful": adef["stateful"],
|
|
"tools": list(adef.get("tools") or []),
|
|
}
|
|
for name, adef in _agents.items()
|
|
}
|
|
|
|
_roles = _agents_cfg.get("roles", {})
|
|
ORCHESTRATOR_AGENT = _roles.get("orchestrator", "")
|
|
AUDITOR_AGENT = _roles.get("auditor", "")
|
|
RECRUITER_AGENT = _roles.get("recruiter", "")
|
|
SUMMARIZER_AGENT = _roles.get("summarizer", "")
|
|
_SYSTEM_AGENTS = {ORCHESTRATOR_AGENT, AUDITOR_AGENT, SUMMARIZER_AGENT} - {""}
|
|
LEAD_AGENT_NAMES = {name for name in _agents if name not in _SYSTEM_AGENTS}
|
|
ACTIVE_PROVIDER = _agents_cfg.get("default_provider", "none")
|
|
|
|
# Agents the orchestrator can dispatch work to.
|
|
# Everything in agents.yaml that is not a system-role agent.
|
|
LEAD_AGENT_NAMES: set[str] = {
|
|
name for name in _agents if name not in _SYSTEM_AGENTS
|
|
}
|
|
|
|
# Global fallback provider name (for /status display).
|
|
ACTIVE_PROVIDER: str = _agents_cfg.get("default_provider", "none")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Knowledge Base
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_kb = _app.get("knowledge_base", {})
|
|
|
|
KB_ENABLED: bool = os.getenv("AIPA_KB_ENABLED",
|
|
str(_kb.get("enabled", True))).lower() == "true"
|
|
KB_PATH: Path = _resolve(os.getenv("AIPA_KB_PATH",
|
|
_kb.get("path", "data/knowledge_base")))
|
|
KB_DEFAULT_COLLECTION: str = os.getenv("AIPA_KB_DEFAULT_COLLECTION",
|
|
_kb.get("default_collection", "general"))
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Ensure writable runtime directories exist
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def ensure_runtime_dirs():
|
|
"""Create runtime directories that may not exist yet (logs, data, archive, tasks)."""
|
|
for d in (LOGS_DIR, DATA_DIR, ARCHIVE_DIR, TASKS_DIR):
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
if KB_ENABLED:
|
|
KB_PATH.mkdir(parents=True, exist_ok=True)
|