340 lines
12 KiB
Python
340 lines
12 KiB
Python
"""Agent 配置管理：从 JSON 文件加载/保存 agent、tool、env、model 配置。
|
||
|
||
系统定位:
|
||
提供统一接口读写 ``agent/core/`` 下的 JSON 配置文件，
|
||
使前端手动配置的 JSON 可以替代 core 底下的相应 .py 硬编码。
|
||
|
||
配置文件:
|
||
agent_config.json - Agent 运行时配置（名称、模型引用、工具列表、system_prompt）
|
||
tool_config.json - 工具权限与参数配置
|
||
env_config.json - 环境变量与 LLM 模型列表
|
||
|
||
可扩展性:
|
||
可增加版本号、校验 schema、多环境配置切换。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import uuid
|
||
from dataclasses import dataclass, field, asdict
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
# ---------- 配置文件路径 ----------
|
||
_CORE_DIR = Path(__file__).resolve().parent
|
||
AGENT_CONFIG_PATH = _CORE_DIR / "agent_config.json"
|
||
TOOL_CONFIG_PATH = _CORE_DIR / "tool_config.json"
|
||
ENV_CONFIG_PATH = _CORE_DIR / "env_config.json"
|
||
|
||
|
||
# ---------- 模型配置 ----------
|
||
@dataclass
|
||
class ModelConfig:
|
||
id: str = "" # 唯一标识
|
||
name: str = "" # 显示名称
|
||
api_key: str = ""
|
||
base_url: str = ""
|
||
model: str = "" # API 模型名
|
||
timeout: float = 60.0
|
||
max_retries: int = 0
|
||
|
||
|
||
# ---------- LLM 配置结构（保留兼容） ----------
|
||
@dataclass
|
||
class LLMConfig:
|
||
model: str = ""
|
||
api_key: str = ""
|
||
base_url: str = ""
|
||
timeout: float = 60.0
|
||
max_retries: int = 0
|
||
|
||
|
||
@dataclass
|
||
class AgentConfig:
|
||
name: str = ""
|
||
system_prompt: str = ""
|
||
max_iterations: int = 200
|
||
llm: LLMConfig = field(default_factory=LLMConfig) # 保留兼容
|
||
llm_model_id: str = "" # 新：引用 models 列表中的模型 ID
|
||
tools: list[str] = field(default_factory=list)
|
||
enabled: bool = True
|
||
role: str = "sub" # "main" | "sub"
|
||
|
||
|
||
@dataclass
|
||
class ToolConfig:
|
||
name: str = ""
|
||
display_name: str = ""
|
||
description: str = ""
|
||
permission: str = "confirm" # "direct" | "confirm"
|
||
auto_execute_rules: list[dict] = field(default_factory=list) # [{"parameter":"action","operator":"equals","value":"open"}, ...]
|
||
enabled: bool = True
|
||
|
||
|
||
# ---------- 通用 JSON 读写 ----------
|
||
def _read_json(path: Path) -> dict | list | None:
|
||
"""读取 JSON 文件，不存在或格式错误返回 None。"""
|
||
if not path.is_file():
|
||
return None
|
||
try:
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
except (json.JSONDecodeError, OSError):
|
||
return None
|
||
|
||
|
||
def _write_json(path: Path, data: Any) -> bool:
|
||
"""写入 JSON 文件，成功返回 True。"""
|
||
try:
|
||
with open(path, "w", encoding="utf-8") as f:
|
||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||
return True
|
||
except OSError:
|
||
return False
|
||
|
||
|
||
# ---------- Agent 配置 ----------
|
||
def _default_agent_config() -> list[dict]:
|
||
"""生成默认 agent 配置（单 agent: main）。"""
|
||
from agent.memory.system_prompt import MAIN_SYSTEM_PROMPT
|
||
from agent.tools import get_all_available_tool_names
|
||
models = load_models()
|
||
default_model_id = models[0]["id"] if models else ""
|
||
return [
|
||
{
|
||
"name": "main",
|
||
"system_prompt": MAIN_SYSTEM_PROMPT,
|
||
"max_iterations": 200,
|
||
"llm_model_id": default_model_id,
|
||
"tools": get_all_available_tool_names(),
|
||
"enabled": True,
|
||
"role": "main",
|
||
}
|
||
]
|
||
|
||
|
||
def load_agent_configs() -> list[AgentConfig]:
|
||
"""从 agent_config.json 加载 agent 配置列表，无文件时返回默认。"""
|
||
raw = _read_json(AGENT_CONFIG_PATH)
|
||
if not raw or not isinstance(raw, list):
|
||
raw = _default_agent_config()
|
||
_write_json(AGENT_CONFIG_PATH, raw)
|
||
|
||
configs = []
|
||
for item in raw:
|
||
llm_raw = item.get("llm", {})
|
||
configs.append(AgentConfig(
|
||
name=item.get("name", ""),
|
||
system_prompt=item.get("system_prompt", ""),
|
||
max_iterations=item.get("max_iterations", 200),
|
||
llm=LLMConfig(
|
||
model=llm_raw.get("model", ""),
|
||
api_key=llm_raw.get("api_key", ""),
|
||
base_url=llm_raw.get("base_url", ""),
|
||
timeout=llm_raw.get("timeout", 60.0),
|
||
max_retries=llm_raw.get("max_retries", 0),
|
||
),
|
||
llm_model_id=item.get("llm_model_id", ""),
|
||
tools=item.get("tools", []),
|
||
enabled=item.get("enabled", True),
|
||
role=item.get("role", "sub"),
|
||
))
|
||
return configs
|
||
|
||
|
||
def save_agent_configs(configs: list[dict]) -> bool:
|
||
"""将 agent 配置列表写入 agent_config.json。"""
|
||
return _write_json(AGENT_CONFIG_PATH, configs)
|
||
|
||
|
||
# ---------- Tool 配置 ----------
|
||
|
||
# 各工具的默认 auto_execute_rules —— 与 tool_policy.py 的硬编码规则对应
|
||
_TOOL_DEFAULT_AUTO_RULES: dict[str, list[dict]] = {
|
||
"browser_control": [
|
||
{"parameter": "action", "operator": "equals", "value": "open"},
|
||
],
|
||
"rag_tool": [
|
||
{"parameter": "action_type", "operator": "equals", "value": "query"},
|
||
],
|
||
}
|
||
|
||
|
||
def _get_tool_parameters() -> dict[str, list[dict]]:
|
||
"""获取每个工具的参数名和类型信息，供前端生成 auto_execute_rules 提示。
|
||
|
||
返回:
|
||
{tool_name: [{"name": "action", "type": "Literal[open,close]"}, ...], ...}
|
||
"""
|
||
from agent.tools import _ALL_AVAILABLE_TOOL_CLASSES
|
||
import typing
|
||
result: dict[str, list[dict]] = {}
|
||
for name, cls in _ALL_AVAILABLE_TOOL_CLASSES.items():
|
||
try:
|
||
inst = cls()
|
||
schema = getattr(inst, "args_schema", None)
|
||
if schema is None:
|
||
result[name] = []
|
||
continue
|
||
fields_info = getattr(schema, "model_fields", {})
|
||
params = []
|
||
for fname, finfo in fields_info.items():
|
||
annotation = finfo.annotation
|
||
type_str = str(annotation) if annotation is not None else "Any"
|
||
# 简化 Literal 显示
|
||
import re
|
||
m = re.search(r"Literal\[(.+?)\]", type_str)
|
||
if m:
|
||
type_str = "Literal[" + m.group(1) + "]"
|
||
params.append({"name": fname, "type": type_str})
|
||
result[name] = params
|
||
except Exception:
|
||
result[name] = []
|
||
return result
|
||
|
||
|
||
def _default_tool_config() -> list[dict]:
|
||
"""根据所有可用工具生成默认 tool 配置。"""
|
||
from agent.tools import get_all_available_tool_names, get_all_available_tool_descriptions
|
||
from agent.core.tool_policy import classify_tool_execution
|
||
all_names = get_all_available_tool_names()
|
||
all_descriptions = get_all_available_tool_descriptions()
|
||
result = []
|
||
for tname in all_names:
|
||
display_name, desc = all_descriptions.get(tname, (tname, ""))
|
||
perm = classify_tool_execution(tname, {})
|
||
default_rules = _TOOL_DEFAULT_AUTO_RULES.get(tname, [])
|
||
result.append({
|
||
"name": tname,
|
||
"display_name": display_name,
|
||
"description": desc,
|
||
"permission": perm,
|
||
"auto_execute_rules": default_rules,
|
||
"enabled": True,
|
||
})
|
||
return result
|
||
|
||
|
||
def load_tool_configs() -> list[ToolConfig]:
|
||
"""从 tool_config.json 加载工具配置，无文件时返回默认。"""
|
||
raw = _read_json(TOOL_CONFIG_PATH)
|
||
if not raw or not isinstance(raw, list):
|
||
raw = _default_tool_config()
|
||
_write_json(TOOL_CONFIG_PATH, raw)
|
||
|
||
configs = []
|
||
for item in raw:
|
||
configs.append(ToolConfig(
|
||
name=item.get("name", ""),
|
||
display_name=item.get("display_name", ""),
|
||
description=item.get("description", ""),
|
||
permission=item.get("permission", "confirm"),
|
||
auto_execute_rules=item.get("auto_execute_rules", []),
|
||
enabled=item.get("enabled", True),
|
||
))
|
||
return configs
|
||
|
||
|
||
def save_tool_configs(configs: list[dict]) -> bool:
|
||
"""将 tool 配置列表写入 tool_config.json。"""
|
||
return _write_json(TOOL_CONFIG_PATH, configs)
|
||
|
||
|
||
# ---------- 强制权限工具（前端锁死，不可修改） ----------
|
||
# 仅必须 confirm 的工具锁死；safe_tools 默认 direct 但用户可改为 confirm
|
||
_FORCED_CONFIRM_TOOLS = {"human_interaction"}
|
||
|
||
|
||
def get_tool_forced_permissions() -> dict[str, str]:
|
||
"""返回强制权限的工具映射 {tool_name: "confirm"}，
|
||
前端应对这些工具禁用权限下拉框和 auto_execute_rules 编辑。"""
|
||
return {name: "confirm" for name in _FORCED_CONFIRM_TOOLS}
|
||
|
||
|
||
# ---------- 模型管理 ----------
|
||
def _default_models() -> list[dict]:
|
||
"""生成默认模型列表，首次启动时从 .env 继承已有 LLM 配置。"""
|
||
import os
|
||
api_key = os.environ.get("LLM_API_KEY", "")
|
||
base_url = os.environ.get("LLM_BASE_URL", "")
|
||
model = os.environ.get("LLM_MODEL", "")
|
||
timeout = os.environ.get("LLM_TIMEOUT", "60")
|
||
if api_key and base_url and model:
|
||
return [
|
||
{
|
||
"id": str(uuid.uuid4())[:8],
|
||
"name": "默认模型",
|
||
"api_key": api_key,
|
||
"base_url": base_url,
|
||
"model": model,
|
||
"timeout": float(timeout) if timeout else 60.0,
|
||
"max_retries": 0,
|
||
}
|
||
]
|
||
return []
|
||
|
||
|
||
def load_models() -> list[dict]:
|
||
"""从 env_config.json 的 models 字段加载模型列表。"""
|
||
raw = _read_json(ENV_CONFIG_PATH)
|
||
if isinstance(raw, dict) and "models" in raw and isinstance(raw["models"], list) and raw["models"]:
|
||
return raw["models"]
|
||
return _default_models()
|
||
|
||
|
||
def save_models(models: list[dict]) -> bool:
|
||
"""将模型列表写入 env_config.json（合并其他字段）。"""
|
||
current = load_env_config()
|
||
current["models"] = models
|
||
return _write_json(ENV_CONFIG_PATH, current)
|
||
|
||
|
||
# ---------- Env 配置 ----------
|
||
def _default_env_config() -> dict:
|
||
"""生成默认 env 配置（不含 models，models 由 load_models 独立处理）。"""
|
||
return {
|
||
"TTS_BASE_URL": "",
|
||
"ASR_BASE_URL": "",
|
||
"RAG_BASE_URL": "",
|
||
"LLM_CONTEXT_WINDOW": "131072",
|
||
"MINI_COMPRESS_RATE": "0.5",
|
||
"HARD_COMPRESS_RATE": "0.75",
|
||
"CURRENT_TURNS": "4",
|
||
"IMG_SIZE": "768",
|
||
"RAG_CHUNK_SIZE": "500",
|
||
"RAG_CHUNK_OVERLAP": "50",
|
||
"GROUNDING_WIDTH": "1000",
|
||
"GROUNDING_HEIGHT": "1000",
|
||
}
|
||
|
||
|
||
def load_env_config() -> dict:
|
||
"""从 env_config.json 加载环境变量配置，无文件时返回默认。"""
|
||
raw = _read_json(ENV_CONFIG_PATH)
|
||
if not raw or not isinstance(raw, dict):
|
||
raw = _default_env_config()
|
||
raw["models"] = _default_models()
|
||
_write_json(ENV_CONFIG_PATH, raw)
|
||
return raw
|
||
# 确保有 models 字段（向后兼容旧配置文件）
|
||
if "models" not in raw or not isinstance(raw.get("models"), list):
|
||
raw["models"] = _default_models()
|
||
return raw
|
||
|
||
|
||
def save_env_config(config: dict) -> bool:
|
||
"""将 env 配置写入 env_config.json（保留已有 models）。"""
|
||
# 保留已有的 models 列表（由 models API 单独管理）
|
||
existing = _read_json(ENV_CONFIG_PATH)
|
||
if isinstance(existing, dict) and "models" in existing:
|
||
config["models"] = existing["models"]
|
||
return _write_json(ENV_CONFIG_PATH, config)
|
||
|
||
|
||
# ---------- 获取所有已注册工具的基础信息 ----------
|
||
def get_registered_tool_names() -> list[str]:
|
||
"""返回所有可用工具名称（含数据库工具）。"""
|
||
from agent.tools import get_all_available_tool_names
|
||
return get_all_available_tool_names()
|