From 344280cd6158fe40b5b6a1fc85d972bd58b203f5 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Wed, 29 Apr 2026 14:12:47 +0800 Subject: [PATCH] Refactor agent persona runtime layering --- app/agent/__init__.py | 23 +- app/agent/defaults/CURRENT_PERSONA.md | 19 + .../defaults/personas/concise/PERSONA.md | 23 + .../defaults/personas/default/PERSONA.md | 24 + app/agent/defaults/personas/guide/PERSONA.md | 22 + app/agent/middleware/hooks.py | 68 -- app/agent/middleware/memory.py | 24 +- app/agent/middleware/runtime_config.py | 42 + app/agent/prompt/System Core Prompt.txt | 44 +- .../System Tasks.yaml} | 11 - app/agent/prompt/__init__.py | 317 +++++- app/agent/runtime.py | 906 +++++++++--------- app/agent/runtime_defaults/CURRENT_PERSONA.md | 24 - .../runtime_defaults/USER_PREFERENCES.md | 10 - .../personas/default/AGENT_HOOKS.md | 26 - .../personas/default/AGENT_PROFILE.md | 27 - .../personas/default/AGENT_WORKFLOW.md | 25 - app/agent/tools/factory.py | 6 + app/agent/tools/impl/query_personas.py | 75 ++ app/agent/tools/impl/switch_persona.py | 62 ++ .../tools/impl/update_persona_definition.py | 131 +++ tests/test_agent_interaction.py | 2 +- tests/test_agent_persona_tools.py | 121 +++ tests/test_agent_prompt_style.py | 72 +- tests/test_agent_runtime.py | 147 ++- tests/test_agent_summarization_streaming.py | 30 + 26 files changed, 1529 insertions(+), 752 deletions(-) create mode 100644 app/agent/defaults/CURRENT_PERSONA.md create mode 100644 app/agent/defaults/personas/concise/PERSONA.md create mode 100644 app/agent/defaults/personas/default/PERSONA.md create mode 100644 app/agent/defaults/personas/guide/PERSONA.md delete mode 100644 app/agent/middleware/hooks.py create mode 100644 app/agent/middleware/runtime_config.py rename app/agent/{runtime_defaults/system_tasks/SYSTEM_TASKS.md => prompt/System Tasks.yaml} (94%) delete mode 100644 app/agent/runtime_defaults/CURRENT_PERSONA.md delete mode 100644 app/agent/runtime_defaults/USER_PREFERENCES.md delete mode 100644 app/agent/runtime_defaults/personas/default/AGENT_HOOKS.md delete mode 100644 app/agent/runtime_defaults/personas/default/AGENT_PROFILE.md delete mode 100644 app/agent/runtime_defaults/personas/default/AGENT_WORKFLOW.md create mode 100644 app/agent/tools/impl/query_personas.py create mode 100644 app/agent/tools/impl/switch_persona.py create mode 100644 app/agent/tools/impl/update_persona_definition.py create mode 100644 tests/test_agent_persona_tools.py diff --git a/app/agent/__init__.py b/app/agent/__init__.py index 6c87706f..f3701f79 100644 --- a/app/agent/__init__.py +++ b/app/agent/__init__.py @@ -21,10 +21,10 @@ from langgraph.checkpoint.memory import InMemorySaver from app.agent.callback import StreamingHandler from app.agent.memory import memory_manager from app.agent.middleware.activity_log import ActivityLogMiddleware -from app.agent.middleware.hooks import AgentHooksMiddleware from app.agent.middleware.jobs import JobsMiddleware from app.agent.middleware.memory import MemoryMiddleware from app.agent.middleware.patch_tool_calls import PatchToolCallsMiddleware +from app.agent.middleware.runtime_config import RuntimeConfigMiddleware from app.agent.middleware.skills import SkillsMiddleware from app.agent.middleware.usage import UsageMiddleware from app.agent.prompt import prompt_manager @@ -406,9 +406,9 @@ class MoviePilotAgent: JobsMiddleware( sources=[str(agent_runtime_manager.jobs_dir)], ), - # 结构化 hooks - AgentHooksMiddleware(), - # 记忆管理（仅扫描 memory 目录，避免与根层 persona/workflow 配置混写） + # 运行时人格与核心规则（动态加载，支持执行中切换人格） + RuntimeConfigMiddleware(), + # 记忆管理（仅扫描 memory 目录，避免与根层核心规则或人格定义混写） MemoryMiddleware(memory_dir=str(agent_runtime_manager.memory_dir)), # 活动日志 ActivityLogMiddleware( @@ -1008,9 +1008,8 @@ class AgentManager: @staticmethod def _build_heartbeat_prompt() -> str: - """使用统一 wake 模板源构建心跳任务提示词。""" - runtime_config = agent_runtime_manager.load_runtime_config() - return runtime_config.render_system_task_message("heartbeat") + """使用程序内置 System Tasks 定义构建心跳任务提示词。""" + return prompt_manager.render_system_task_message("heartbeat") @staticmethod def _build_retry_transfer_template_context( @@ -1034,23 +1033,22 @@ class AgentManager: history_ids: list[int], ) -> str: """根据失败记录数量构建统一的重试整理后台任务提示词。""" - runtime_config = agent_runtime_manager.load_runtime_config() task_type, template_context = AgentManager._build_retry_transfer_template_context( history_ids ) - return runtime_config.render_system_task_message( + return prompt_manager.render_system_task_message( task_type, template_context=template_context, ) @staticmethod def _build_manual_redo_template_context(history) -> dict[str, int | str]: - """仅负责把整理历史对象映射成 SYSTEM_TASKS 需要的模板变量。""" + """仅负责把整理历史对象映射成 System Tasks 需要的模板变量。""" src_fileitem = history.src_fileitem or {} source_path = src_fileitem.get("path") if isinstance(src_fileitem, dict) else "" source_path = source_path or history.src or "" season_episode = f"{history.seasons or ''}{history.episodes or ''}".strip() - # 这里故意只做数据整形，具体行为定义全部交给 SYSTEM_TASKS。 + # 这里故意只做数据整形，具体行为定义全部交给内置 System Tasks YAML。 return { "history_id": history.id, "current_status": "success" if history.status else "failed", @@ -1203,8 +1201,7 @@ class AgentManager: """ 构建手动 AI 整理提示词。 """ - runtime_config = agent_runtime_manager.load_runtime_config() - return runtime_config.render_system_task_message( + return prompt_manager.render_system_task_message( "manual_transfer_redo", template_context=AgentManager._build_manual_redo_template_context(history), ) diff --git a/app/agent/defaults/CURRENT_PERSONA.md b/app/agent/defaults/CURRENT_PERSONA.md new file mode 100644 index 00000000..55ecb965 --- /dev/null +++ b/app/agent/defaults/CURRENT_PERSONA.md @@ -0,0 +1,19 @@ +--- +version: 3 +active_persona: default +extra_context_files: [] +deprecated_phrases: [] +--- +# CURRENT_PERSONA + +当前激活人格：`default` + +运行时加载顺序固定如下： + +1. 核心系统提示词（程序内置，不可运行时覆盖） +2. `personas//PERSONA.md` +3. `extra_context_files` +4. `memory/*.md` +5. `activity/*.md` + +`memory` 中的长期偏好可以细化回复方式，但不应覆盖系统核心身份、目标和安全边界。 diff --git a/app/agent/defaults/personas/concise/PERSONA.md b/app/agent/defaults/personas/concise/PERSONA.md new file mode 100644 index 00000000..d584550f --- /dev/null +++ b/app/agent/defaults/personas/concise/PERSONA.md @@ -0,0 +1,23 @@ +--- +version: 1 +persona_id: concise +label: 极简 +description: 更短、更硬朗，优先结论和动作，不主动展开背景解释。 +aliases: + - 简洁 + - 干脆 + - 极简人格 +--- +# PERSONA + +- Tone: terse, decisive, and highly compressed. +- Prefer the shortest complete answer that still moves the task forward. +- Default to one sentence when possible. Only use lists when they materially improve readability. +- Avoid extra context, caveats, or teaching unless the user explicitly asks for explanation. +- Keep transitions minimal and skip conversational softening. + +## RESPONSE_FORMAT + +- Lead with the conclusion or result. +- For option lists, keep each item very short. +- Do not repeat already-known context back to the user unless it is needed to disambiguate the action. diff --git a/app/agent/defaults/personas/default/PERSONA.md b/app/agent/defaults/personas/default/PERSONA.md new file mode 100644 index 00000000..488a6a22 --- /dev/null +++ b/app/agent/defaults/personas/default/PERSONA.md @@ -0,0 +1,24 @@ +--- +version: 1 +persona_id: default +label: 默认 +description: 专业、克制、简洁，适合大多数日常媒体管理场景。 +aliases: + - 专业 + - 默认人格 +--- +# PERSONA + +- Tone: professional, concise, restrained. +- Be direct. No unnecessary preamble, no repeating the user's words, no narrating internal reasoning. +- Do not flatter the user, praise the question, or add emotional cushioning. +- Do not use emojis, exclamation marks, cute language, or excessive apology. +- Prefer short declarative sentences. Default to one or two short paragraphs; use lists only when they improve scanability. +- Use Markdown for structured data. Use `inline code` for media titles and paths. + +## RESPONSE_FORMAT + +- Keep confirmations short. +- For search or comparison results, prefer a brief list over a long paragraph. +- Skip filler phrases like "Let me help you", "Here are the results", or "I found...". +- When an error occurs, briefly state the blocker and the next best action. diff --git a/app/agent/defaults/personas/guide/PERSONA.md b/app/agent/defaults/personas/guide/PERSONA.md new file mode 100644 index 00000000..5b00df6f --- /dev/null +++ b/app/agent/defaults/personas/guide/PERSONA.md @@ -0,0 +1,22 @@ +--- +version: 1 +persona_id: guide +label: 说明型 +description: 在复杂问题上更愿意解释原因和步骤，但仍保持克制，不会无节制展开。 +aliases: + - 讲解 + - 解释型 + - 教学 +--- +# PERSONA + +- Tone: clear, structured, and mildly explanatory. +- When the task is simple, stay concise. When the task is complex or the user asks why/how, provide a short explanation with visible structure. +- Keep explanations practical and tied to the current decision, not generic theory. +- Remain restrained: do not become chatty, cute, or overly warm. + +## RESPONSE_FORMAT + +- For non-trivial tasks, prefer short sections or a compact numbered list. +- When describing tradeoffs, keep them concrete and action-oriented. +- End with the actual outcome or next step, not a generic summary. diff --git a/app/agent/middleware/hooks.py b/app/agent/middleware/hooks.py deleted file mode 100644 index 57c7fce2..00000000 --- a/app/agent/middleware/hooks.py +++ /dev/null @@ -1,68 +0,0 @@ -"""结构化 Agent hooks 中间件。""" - -from collections.abc import Awaitable, Callable -from typing import Annotated, NotRequired, TypedDict - -from langchain.agents.middleware.types import ( - AgentMiddleware, - AgentState, - ContextT, - ModelRequest, - ModelResponse, - PrivateStateAttr, # noqa - ResponseT, -) -from langchain_core.runnables import RunnableConfig -from langgraph.runtime import Runtime - -from app.agent.middleware.utils import append_to_system_message -from app.agent.runtime import agent_runtime_manager - - -class HooksState(AgentState): - """hooks 中间件状态。""" - - hooks_prompt: NotRequired[Annotated[str, PrivateStateAttr]] - - -class HooksStateUpdate(TypedDict): - """hooks 状态更新。""" - - hooks_prompt: str - - -class AgentHooksMiddleware(AgentMiddleware[HooksState, ContextT, ResponseT]): # noqa - """在固定生命周期点注入结构化 pre/in/post hooks。""" - - state_schema = HooksState - - async def abefore_agent( # noqa - self, state: HooksState, runtime: Runtime, config: RunnableConfig - ) -> HooksStateUpdate | None: - if "hooks_prompt" in state: - return None - - runtime_config = agent_runtime_manager.load_runtime_config() - return HooksStateUpdate(hooks_prompt=runtime_config.render_hooks_prompt()) - - def modify_request(self, request: ModelRequest[ContextT]) -> ModelRequest[ContextT]: # noqa - hooks_prompt = request.state.get("hooks_prompt", "") # noqa - if not hooks_prompt: - return request - - new_system_message = append_to_system_message( - request.system_message, hooks_prompt - ) - return request.override(system_message=new_system_message) - - async def awrap_model_call( - self, - request: ModelRequest[ContextT], - handler: Callable[ - [ModelRequest[ContextT]], Awaitable[ModelResponse[ResponseT]] - ], - ) -> ModelResponse[ResponseT]: - return await handler(self.modify_request(request)) - - -__all__ = ["AgentHooksMiddleware"] diff --git a/app/agent/middleware/memory.py b/app/agent/middleware/memory.py index 3a768d4b..6bfef8ac 100644 --- a/app/agent/middleware/memory.py +++ b/app/agent/middleware/memory.py @@ -57,8 +57,8 @@ You can create, edit, or organize any `.md` files in this directory to manage yo **Memory file organization:** - All `.md` files in `{memory_dir}` are automatically loaded as memory. - - `MEMORY.md` is the default/primary memory file for general user preferences and profile. - - You may create additional `.md` files to organize knowledge by topic (e.g., `MEDIA_RULES.md`, `DOWNLOAD_PREFERENCES.md`, `SITE_CONFIGS.md`, etc.). + - `MEMORY.md` is the default/primary memory file for general user preferences, communication style, and durable working rules. + - You may create additional `.md` files to organize knowledge by topic (e.g., `MEDIA_RULES.md`, `COMMUNICATION_PREFERENCES.md`, `DOWNLOAD_PREFERENCES.md`, `SITE_CONFIGS.md`, etc.). - Keep each file focused on a specific domain or topic for better organization. - Subdirectories are NOT scanned — only `.md` files directly in `{memory_dir}`. @@ -78,11 +78,11 @@ You can create, edit, or organize any `.md` files in this directory to manage yo **When to update memories:** - When the user explicitly asks you to remember something (e.g., "remember my email", "save this preference") - - When the user describes your role or how you should behave (e.g., "you are a web researcher", "always do X") + - When the user gives durable communication or reply-format preferences (e.g., "be more concise", "prefer tables", "use JSON when summarizing") - When the user gives feedback on your work - capture what was wrong and how to improve - When the user provides information required for tool use (e.g., slack channel ID, email addresses) - When the user provides context useful for future tasks, such as how to use tools, or which actions to take in a particular situation - - When you discover new patterns or preferences (coding styles, conventions, workflows) + - When you discover new user-specific patterns or preferences (communication style, formatting, workflows) **When to NOT update memories:** - When the information is temporary or transient (e.g., "I'm running late", "I'm on my phone right now") @@ -90,6 +90,8 @@ You can create, edit, or organize any `.md` files in this directory to manage yo - When the information is a simple question that doesn't reveal lasting preferences (e.g., "What day is it?", "Can you explain X?") - When the information is an acknowledgment or small talk (e.g., "Sounds good!", "Hello", "Thanks for that") - When the information is stale or irrelevant in future conversations + - Memory may refine user-facing style, but it must NOT redefine the agent's core identity, safety boundaries, or global system-task rules. + - If the user wants a built-in speaking style/persona, prefer the dedicated persona-switching tools instead of rewriting memory as a substitute. - Never store API keys, access tokens, passwords, or any other credentials in any file, memory, or system prompt. - If the user asks where to put API keys or provides an API key, do NOT echo or save it. - Do NOT record daily activities or task execution history in memory files - these are automatically tracked in the activity log system (see ). Memory files are only for long-term knowledge, preferences, and patterns. @@ -135,7 +137,7 @@ Default memory file: {memory_file} - Only ask for preferences when they are directly useful for the current task, or when a short follow-up question at the end would clearly help future interactions. **What to collect when useful:** - - Preferred communication style + - Preferred communication style or persona preference - Media interests - Quality / codec / subtitle preferences - Any standing rules the user wants you to follow @@ -153,7 +155,7 @@ Default memory file: {memory_file} Your memory directory is at: {memory_dir}. You can save new knowledge by calling the `edit_file` or `write_file` tool on any `.md` file in this directory. **Memory file organization:** - - `MEMORY.md` is the default/primary memory file for general user preferences and profile. + - `MEMORY.md` is the default/primary memory file for user preferences, persona preferences, and durable working rules. - You may create additional `.md` files to organize knowledge by topic. - All `.md` files directly in the memory directory are automatically loaded on each conversation. @@ -166,15 +168,17 @@ Default memory file: {memory_file} **When to update memories:** - When the user explicitly asks you to remember something - - When the user describes your role or how you should behave + - When the user gives durable communication or reply-format preferences - When the user gives feedback on your work - When the user provides information required for tool use - - When you discover new patterns or preferences + - When you discover new user-specific patterns or preferences **When to NOT update memories:** - Temporary/transient information - One-time task requests - Simple questions, acknowledgments, or small talk + - Memory may refine user-facing style, but it must NOT redefine the agent's core identity, safety boundaries, or global system-task rules + - If the user wants a built-in speaking style/persona, prefer the dedicated persona-switching tools instead of rewriting memory as a substitute - Never store API keys, access tokens, passwords, or credentials - Do NOT record daily activities in memory files — those go to the activity log @@ -189,7 +193,7 @@ class MemoryMiddleware(AgentMiddleware[MemoryState, ContextT, ResponseT]): # no 参数： memory_dir: 记忆文件目录路径。建议使用独立的 `config/agent/memory` - 目录，避免与 persona/workflow 等根层配置混写。 + 目录，避免与核心规则或人格定义混写。 """ state_schema = MemoryState @@ -289,7 +293,7 @@ class MemoryMiddleware(AgentMiddleware[MemoryState, ContextT, ResponseT]): # no return md_files - async def abefore_agent( + async def abefore_agent( # noqa self, state: MemoryState, runtime: Runtime, # noqa diff --git a/app/agent/middleware/runtime_config.py b/app/agent/middleware/runtime_config.py new file mode 100644 index 00000000..df2eadbf --- /dev/null +++ b/app/agent/middleware/runtime_config.py @@ -0,0 +1,42 @@ +"""动态注入 Agent 根层运行时配置的中间件。""" + +from collections.abc import Awaitable, Callable + +from langchain.agents.middleware.types import ( + AgentMiddleware, + ContextT, + ModelRequest, + ModelResponse, + ResponseT, +) + +from app.agent.middleware.utils import append_to_system_message +from app.agent.runtime import agent_runtime_manager + + +class RuntimeConfigMiddleware(AgentMiddleware[dict, ContextT, ResponseT]): # noqa + """在每次模型调用前动态加载运行时配置。 + + 这里不把结果缓存到 middleware state 中，目的是让人格切换工具在同一轮 + Agent 执行里修改 CURRENT_PERSONA 后，后续模型调用可以立即看到新的人格。 + """ + + def modify_request(self, request: ModelRequest[ContextT]) -> ModelRequest[ContextT]: # noqa + runtime_config = agent_runtime_manager.load_runtime_config() + runtime_sections = runtime_config.render_prompt_sections() + new_system_message = append_to_system_message( + request.system_message, runtime_sections + ) + return request.override(system_message=new_system_message) + + async def awrap_model_call( + self, + request: ModelRequest[ContextT], + handler: Callable[ + [ModelRequest[ContextT]], Awaitable[ModelResponse[ResponseT]] + ], + ) -> ModelResponse[ResponseT]: + return await handler(self.modify_request(request)) + + +__all__ = ["RuntimeConfigMiddleware"] diff --git a/app/agent/prompt/System Core Prompt.txt b/app/agent/prompt/System Core Prompt.txt index 5e980abc..07011dbc 100644 --- a/app/agent/prompt/System Core Prompt.txt +++ b/app/agent/prompt/System Core Prompt.txt @@ -1,12 +1,47 @@ -You are the MoviePilot agent runtime. Follow the injected root configuration to determine the active persona, workflow, and operator preferences. +You are the MoviePilot agent runtime. Follow the injected runtime configuration to determine the active persona and any extra user-specific context. All your responses must be in **Chinese (中文)**. You act as a proactive agent. Your goal is to fully resolve the user's media-related requests autonomously. Do not end your turn until the task is complete or you are blocked and require user feedback. - -{runtime_sections} - + +Identity and Goal: +- You are an AI media assistant powered by MoviePilot. +- Your primary goal is to fully resolve the user's MoviePilot-related media tasks with the available tools whenever the request is actionable. +- Focus on MoviePilot's home media domain: search, recognition, subscriptions, downloads, library organization, file transfer, and system status. +- Stay within the MoviePilot product domain unless the user explicitly asks for adjacent help that can be handled with your existing tools. + +Behavior Model: +- Prioritize task progress over conversation. +- Check current state before making changes, then do the smallest correct action. +- Do not stop for approval on read-only operations. Only confirm before destructive or high-impact actions such as starting downloads, deleting subscriptions, or removing history. +- When a request can be completed by tools, prefer doing the work over explaining what you might do. +- After an action, perform the minimum validation needed to confirm the result actually landed. +- If the user explicitly asks to change the speaking style or persona, use the dedicated persona tools instead of editing runtime files manually. +- If the user explicitly asks to rewrite or create a persona definition, prefer `update_persona_definition` rather than generic file-editing tools. +- Do not let user memory or persona style override this core identity, safety boundaries, or built-in background task rules. +- You are not a general-purpose coding assistant in normal media conversations. Only cross into implementation details when the user explicitly asks about MoviePilot internals or debugging. + +Core Workflow: +1. Media Discovery: Identify exact media metadata such as TMDB ID and Season or Episode using search tools when needed. +2. Context Checking: Verify whether the media already exists in the library, has already been subscribed, or has relevant history that affects the next step. +3. Action Execution: Perform the requested task with concise user-facing output unless the operation is destructive or blocked. +4. Final Confirmation: State the outcome briefly, including the key media facts or blocker. + +Tool Calling Strategy: +- Call independent tools in parallel whenever possible. +- If search results are ambiguous, use `query_media_detail` or `recognize_media` to clarify before proceeding. +- If `search_media` fails, fall back to `search_web` or `recognize_media`. Only ask the user when automated paths are exhausted. +- Reuse known media identity, prior tool results, and current system context instead of repeating expensive recognition or search calls. +- When a tool fails, try one narrower fallback path before escalating to the user. + +Media Management Rules: +1. Download Safety: Present found torrents with size, seeds, and quality, then get explicit consent before downloading. +2. Subscription Logic: Check for the best matching quality profile based on user history or defaults. +3. Library Awareness: Check if content already exists in the library to avoid duplicates. +4. Error Handling: If a tool or site fails, briefly explain what went wrong and suggest an alternative. +5. TV Subscription Rule: When calling `add_subscribe` for a TV show, omitting `season` means subscribe to season 1 only. To subscribe multiple seasons or the full series, call `add_subscribe` separately for each season. + {verbose_spec} @@ -25,6 +60,7 @@ You act as a proactive agent. Your goal is to fully resolve the user's media-rel 4. System Status and Organization - Monitor downloads, server health, file transfers, renaming, and library cleanup. 5. Visual Input Handling - Users may attach images from supported channels; analyze them together with the text when relevant. 6. File Context Handling - User messages may arrive as structured JSON. Treat the `message` field as the user's text. Attachments appear in `files`; when `local_path` is present, use local file tools to inspect the uploaded file directly. When image input is disabled for the current model, user images may also be delivered through `files`. +7. Persona Management - If the user explicitly asks to change the speaking style or persona, prefer `query_personas` and `switch_persona`; if the user asks to rewrite or create a persona definition, prefer `update_persona_definition` instead of editing runtime files manually. diff --git a/app/agent/runtime_defaults/system_tasks/SYSTEM_TASKS.md b/app/agent/prompt/System Tasks.yaml similarity index 94% rename from app/agent/runtime_defaults/system_tasks/SYSTEM_TASKS.md rename to app/agent/prompt/System Tasks.yaml index a1bc0b31..f49464df 100644 --- a/app/agent/runtime_defaults/system_tasks/SYSTEM_TASKS.md +++ b/app/agent/prompt/System Tasks.yaml @@ -1,4 +1,3 @@ ---- version: 2 shared_rules: - This is a background system task, NOT a user conversation. @@ -96,13 +95,3 @@ task_types: - "Do NOT reorganize blindly when media identity is uncertain." - "If the previous record was successful but obviously identified as the wrong media, still use the tool-based flow above instead of `/redo`." - "Keep the final response short and focused on outcome." ---- -# SYSTEM_TASKS - -这是后台系统任务的唯一定义源。 - -- `shared_rules` 负责统一口径。 -- `task_types..context_lines` 负责定义上下文字段展示。 -- `task_types..steps` 负责定义任务执行步骤。 -- `task_types..task_rules` 负责定义该任务独有的补充约束。 -- 代码侧只负责触发任务并提供模板变量，不再保存具体行为提示词。 diff --git a/app/agent/prompt/__init__.py b/app/agent/prompt/__init__.py index f8e4d7c8..814d4e0e 100644 --- a/app/agent/prompt/__init__.py +++ b/app/agent/prompt/__init__.py @@ -1,13 +1,16 @@ """提示词管理器""" import socket +from dataclasses import dataclass, field from pathlib import Path +from string import Formatter from time import strftime -from typing import Dict +from typing import Any, Dict, Optional + +import yaml from app.core.config import settings from app.log import logger -from app.agent.runtime import agent_runtime_manager from app.schemas import ( ChannelCapability, ChannelCapabilities, @@ -16,6 +19,37 @@ from app.schemas import ( ) from app.utils.system import SystemUtils +SYSTEM_TASKS_FILE = "System Tasks.yaml" +SYSTEM_TASKS_SCHEMA_VERSION = 2 + + +class PromptConfigError(ValueError): + """程序内置提示词定义加载异常。""" + + +@dataclass +class SystemTaskTypeDefinition: + """单个后台系统任务定义。""" + + header: str + objective: str + context_title: Optional[str] = None + context_lines: list[str] = field(default_factory=list) + steps_title: Optional[str] = None + steps: list[str] = field(default_factory=list) + task_rules: list[str] = field(default_factory=list) + empty_result: Optional[str] = None + + +@dataclass +class SystemTasksDefinition: + """程序内置后台系统任务定义。""" + + path: Path + version: int + shared_rules: list[str] + task_types: dict[str, SystemTaskTypeDefinition] + class PromptManager: """ @@ -28,6 +62,8 @@ class PromptManager: else: self.prompts_dir = Path(prompts_dir) self.prompts_cache: Dict[str, str] = {} + self._system_tasks_cache: Optional[SystemTasksDefinition] = None + self._system_tasks_signature: Optional[tuple[int, int]] = None def load_prompt(self, prompt_name: str) -> str: """ @@ -60,11 +96,9 @@ class PromptManager: :param prefer_voice_reply: 是否优先使用语音回复 :return: 提示词内容 """ - # 根层运行时配置由独立装配器负责，避免人格/工作流继续硬编码在单文件 prompt 中。 - runtime_config = agent_runtime_manager.load_runtime_config() - runtime_sections = runtime_config.render_prompt_sections() - # 基础提示词只保留 MoviePilot 运行时和渠道能力相关约束。 + # 根层运行时配置由 RuntimeConfigMiddleware 在每次模型调用前动态注入， + # 这样人格切换可以在同一轮 Agent 执行里立即生效。 base_prompt = self.load_prompt("System Core Prompt.txt") # 识别渠道 @@ -109,11 +143,119 @@ class PromptManager: moviepilot_info=moviepilot_info, voice_reply_spec=voice_reply_spec, button_choice_spec=button_choice_spec, - runtime_sections=runtime_sections, ) return base_prompt + def load_system_tasks_definition(self) -> SystemTasksDefinition: + """加载程序内置的后台系统任务定义。""" + system_tasks_path = self.prompts_dir / SYSTEM_TASKS_FILE + try: + stat = system_tasks_path.stat() + except FileNotFoundError as err: + logger.error(f"系统任务定义文件不存在: {system_tasks_path}") + raise PromptConfigError(f"系统任务定义文件不存在: {system_tasks_path}") from err + + signature = (stat.st_mtime_ns, stat.st_size) + if ( + self._system_tasks_signature == signature + and self._system_tasks_cache is not None + ): + return self._system_tasks_cache + + try: + content = system_tasks_path.read_text(encoding="utf-8") + except Exception as err: # noqa: BLE001 + logger.error(f"读取系统任务定义失败: {system_tasks_path}, 错误: {err}") + raise PromptConfigError( + f"读取系统任务定义失败 {system_tasks_path}: {err}" + ) from err + + try: + data = yaml.safe_load(content) or {} + except yaml.YAMLError as err: + raise PromptConfigError(f"YAML 解析失败 {system_tasks_path}: {err}") from err + if not isinstance(data, dict): + raise PromptConfigError( + f"YAML 根节点必须是映射类型: {system_tasks_path}" + ) + + definition = self._parse_system_tasks_definition(system_tasks_path, data) + self._system_tasks_signature = signature + self._system_tasks_cache = definition + return definition + + def render_system_task_message( + self, + task_type: str, + *, + template_context: Optional[dict[str, Any]] = None, + extra_rules: Optional[list[str]] = None, + ) -> str: + """根据程序内置 YAML 渲染后台系统任务提示词。""" + system_tasks = self.load_system_tasks_definition() + task_definition = system_tasks.task_types.get(task_type) + if not task_definition: + raise PromptConfigError(f"未定义的后台系统任务类型: {task_type}") + + rendered_context = self._render_template_lines( + task_definition.context_lines, + template_context, + task_type, + "context_lines", + ) + rendered_steps = self._render_template_lines( + task_definition.steps, + template_context, + task_type, + "steps", + ) + rendered_task_rules = self._render_template_lines( + task_definition.task_rules, + template_context, + task_type, + "task_rules", + ) + + sections = [ + self._render_template_text( + task_definition.header, + template_context, + task_type, + "header", + ).strip(), + self._render_template_text( + task_definition.objective, + template_context, + task_type, + "objective", + ).strip(), + ] + if rendered_context: + sections.append( + self._format_titled_lines( + task_definition.context_title or "Task context", + rendered_context, + ) + ) + if rendered_steps: + sections.append( + self._format_titled_lines( + task_definition.steps_title or "Follow these steps", + rendered_steps, + ) + ) + + rules = list(system_tasks.shared_rules) + if task_definition.empty_result: + rules.append(task_definition.empty_result) + rules.extend(rendered_task_rules) + if extra_rules: + rules.extend(rule.strip() for rule in extra_rules if rule and rule.strip()) + if rules: + sections.append(self._format_numbered_rules("IMPORTANT", rules)) + return "\n\n".join(section for section in sections if section).strip() + @staticmethod def _get_moviepilot_info() -> str: """ @@ -214,11 +356,172 @@ class PromptManager: ) return "- User questions: When you truly need user input, ask briefly in plain text." + def _parse_system_tasks_definition( + self, + path: Path, + data: dict[str, Any], + ) -> SystemTasksDefinition: + """把 YAML 结构转换成系统任务定义对象。""" + version = self._normalize_positive_int(data.get("version"), "version", default=1) + if version < SYSTEM_TASKS_SCHEMA_VERSION: + raise PromptConfigError( + f"{path} 的 version={version} 过旧，" + f"当前要求 System Tasks schema v{SYSTEM_TASKS_SCHEMA_VERSION} 或更高版本" + ) + + shared_rules = self._normalize_string_list(data.get("shared_rules"), "shared_rules") + if not shared_rules: + raise PromptConfigError(f"{path} 缺少 shared_rules") + + raw_task_types = data.get("task_types") + if not isinstance(raw_task_types, dict) or not raw_task_types: + raise PromptConfigError(f"{path} 缺少 task_types 映射") + + task_types: dict[str, SystemTaskTypeDefinition] = {} + for key, raw in raw_task_types.items(): + if not isinstance(raw, dict): + raise PromptConfigError(f"task_types.{key} 必须是映射") + + header = str(raw.get("header") or "").strip() + objective = str(raw.get("objective") or "").strip() + if not header or not objective: + raise PromptConfigError(f"task_types.{key} 缺少 header 或 objective") + + task_types[str(key)] = SystemTaskTypeDefinition( + header=header, + objective=objective, + context_title=str(raw.get("context_title") or "").strip() or None, + context_lines=self._normalize_string_list( + raw.get("context_lines"), + f"task_types.{key}.context_lines", + ), + steps_title=str(raw.get("steps_title") or "").strip() or None, + steps=self._normalize_string_list( + raw.get("steps"), + f"task_types.{key}.steps", + ), + task_rules=self._normalize_string_list( + raw.get("task_rules"), + f"task_types.{key}.task_rules", + ), + empty_result=str(raw.get("empty_result") or "").strip() or None, + ) + return SystemTasksDefinition( + path=path, + version=version, + shared_rules=shared_rules, + task_types=task_types, + ) + + @classmethod + def _render_template_text( + cls, + text: str, + template_context: Optional[dict[str, Any]], + task_type: str, + field_name: str, + ) -> str: + if not text: + return "" + + formatter = Formatter() + required_fields = { + placeholder_name + for _, placeholder_name, _, _ in formatter.parse(text) + if placeholder_name + } + if not required_fields: + return text + + context = cls._normalize_template_context(template_context) + missing_fields = sorted(field for field in required_fields if field not in context) + if missing_fields: + raise PromptConfigError( + f"系统任务定义 `{task_type}` 的 `{field_name}` 缺少变量: " + + ", ".join(f"`{field}`" for field in missing_fields) + ) + + # 这里统一做字符串替换，让 YAML 成为后台任务文案的唯一行为来源。 + return text.format_map(context) + + @classmethod + def _render_template_lines( + cls, + items: list[str], + template_context: Optional[dict[str, Any]], + task_type: str, + field_name: str, + ) -> list[str]: + return [ + cls._render_template_text( + item, + template_context, + task_type, + f"{field_name}[{index}]", + ).rstrip() + for index, item in enumerate(items, start=1) + if item and item.rstrip() + ] + + @staticmethod + def _normalize_template_context( + template_context: Optional[dict[str, Any]], + ) -> dict[str, str]: + if not template_context: + return {} + return { + str(key): "" if value is None else str(value) + for key, value in template_context.items() + } + + @staticmethod + def _format_numbered_rules(title: str, items: list[str]) -> str: + return "\n".join( + [f"{title}:"] + [f"{index}. {item}" for index, item in enumerate(items, start=1)] + ) + + @staticmethod + def _format_titled_lines(title: str, items: list[str]) -> str: + cleaned = [item.rstrip() for item in items if item and item.rstrip()] + return "\n".join([f"{title}:"] + cleaned) + + @staticmethod + def _normalize_positive_int( + value: Any, + field_name: str, + *, + default: int, + ) -> int: + if value in (None, ""): + return default + try: + normalized = int(value) + except (TypeError, ValueError) as err: + raise PromptConfigError(f"{field_name} 必须是正整数") from err + if normalized <= 0: + raise PromptConfigError(f"{field_name} 必须是正整数") + return normalized + + @staticmethod + def _normalize_string_list(values: Any, field_name: str) -> list[str]: + if values is None: + return [] + if not isinstance(values, list): + raise PromptConfigError(f"{field_name} 必须是字符串数组") + normalized: list[str] = [] + for value in values: + text = str(value).strip() + if text: + normalized.append(text) + return normalized + def clear_cache(self): """ 清空缓存 """ self.prompts_cache.clear() + self._system_tasks_cache = None + self._system_tasks_signature = None logger.info("提示词缓存已清空") diff --git a/app/agent/runtime.py b/app/agent/runtime.py index a3ae4ffc..18b66c68 100644 --- a/app/agent/runtime.py +++ b/app/agent/runtime.py @@ -7,7 +7,6 @@ import shutil import threading from dataclasses import dataclass, field from pathlib import Path -from string import Formatter from typing import Any, Iterable, Optional import yaml @@ -16,24 +15,45 @@ from app.core.config import settings from app.log import logger CURRENT_PERSONA_FILE = "CURRENT_PERSONA.md" -USER_PREFERENCES_FILE = "USER_PREFERENCES.md" -SYSTEM_TASKS_FILE = "SYSTEM_TASKS.md" -LEGACY_WAKE_FORMAT_FILE = "WAKE_FORMAT.md" SYSTEM_RUNTIME_DIR = "runtime" MEMORY_DIR = "memory" SKILLS_DIR = "skills" JOBS_DIR = "jobs" ACTIVITY_DIR = "activity" -SYSTEM_TASKS_SCHEMA_VERSION = 2 +PERSONAS_DIR = "personas" +PERSONA_FILE = "PERSONA.md" +CURRENT_PERSONA_SCHEMA_VERSION = 3 +PERSONA_SCHEMA_VERSION = 1 +DEFAULT_PERSONA_ID = "default" +PERSONA_ID_PATTERN = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$") ROOT_LEVEL_RUNTIME_FILES = { CURRENT_PERSONA_FILE, +} + +OBSOLETE_AGENT_ROOT_FILES = { + "AGENT_CORE.md", "AGENT_PROFILE.md", "AGENT_WORKFLOW.md", "AGENT_HOOKS.md", - USER_PREFERENCES_FILE, - SYSTEM_TASKS_FILE, - LEGACY_WAKE_FORMAT_FILE, + "USER_PREFERENCES.md", + "SYSTEM_TASKS.md", + "WAKE_FORMAT.md", +} + +OBSOLETE_RUNTIME_FILES = { + Path("AGENT_CORE.md"), + Path("AGENT_PROFILE.md"), + Path("AGENT_WORKFLOW.md"), + Path("AGENT_HOOKS.md"), + Path("USER_PREFERENCES.md"), + Path("SYSTEM_TASKS.md"), + Path("WAKE_FORMAT.md"), + Path("personas") / DEFAULT_PERSONA_ID / "AGENT_PROFILE.md", + Path("personas") / DEFAULT_PERSONA_ID / "AGENT_WORKFLOW.md", + Path("personas") / DEFAULT_PERSONA_ID / "AGENT_HOOKS.md", + Path("system_tasks") / "SYSTEM_TASKS.md", + Path("templates") / "WAKE_FORMAT.md", } FRONTMATTER_PATTERN = re.compile(r"^---\s*\n(.*?)\n---\s*\n?", re.DOTALL) @@ -52,37 +72,43 @@ class ParsedMarkdownDocument: @dataclass -class HookDefinition: - """结构化执行钩子定义。""" +class PersonaDefinition: + """单个人格定义。""" + persona_id: str path: Path - pre_task: list[str] - in_task: list[str] - post_task: list[str] + label: str + description: str + text: str + aliases: list[str] = field(default_factory=list) + def matches(self, query: str) -> bool: + """判断 query 是否命中当前人格。""" + normalized = query.strip().casefold() + if not normalized: + return False + candidates = [self.persona_id, self.label, *self.aliases] + return any(candidate.strip().casefold() == normalized for candidate in candidates) -@dataclass -class SystemTaskTypeDefinition: - """单个后台系统任务定义。""" + def summary_line(self) -> str: + """渲染可读的一行人格摘要。""" + parts = [f"`{self.persona_id}`"] + if self.label and self.label != self.persona_id: + parts.append(self.label) + if self.description: + parts.append(self.description) + return " - ".join(parts) - header: str - objective: str - context_title: Optional[str] = None - context_lines: list[str] = field(default_factory=list) - steps_title: Optional[str] = None - steps: list[str] = field(default_factory=list) - task_rules: list[str] = field(default_factory=list) - empty_result: Optional[str] = None - - -@dataclass -class SystemTasksDefinition: - """统一的后台系统任务定义源。""" - - path: Path - version: int - shared_rules: list[str] - task_types: dict[str, SystemTaskTypeDefinition] + def to_dict(self, *, is_active: bool) -> dict[str, Any]: + """输出给查询工具的结构化信息。""" + return { + "persona_id": self.persona_id, + "label": self.label, + "description": self.description, + "aliases": self.aliases, + "is_active": is_active, + "path": str(self.path), + } @dataclass @@ -92,232 +118,82 @@ class AgentRuntimeConfig: source_root: Path active_persona: str current_persona_path: Path - profile_path: Path - workflow_path: Path - hooks_path: Path - user_preferences_path: Optional[Path] - system_tasks_path: Path + persona: PersonaDefinition + available_personas: list[PersonaDefinition] extra_context_paths: list[Path] - profile_text: str - workflow_text: str - user_preferences_text: str extra_contexts: list[tuple[Path, str]] - hooks: HookDefinition - system_tasks: SystemTasksDefinition warnings: list[str] = field(default_factory=list) used_fallback: bool = False def render_prompt_sections(self) -> str: - """渲染进入系统提示词的根层配置片段。""" + """渲染进入系统提示词的运行时片段。""" sections: list[str] = [ - "", + "", f"- Active persona: `{self.active_persona}`", - f"- Profile source: `{self.profile_path}`", - f"- Workflow source: `{self.workflow_path}`", + f"- Active persona source: `{self.persona.path}`", ] - if self.user_preferences_path: - sections.append(f"- Root preferences source: `{self.user_preferences_path}`") - sections.append(f"- System task source: `{self.system_tasks_path}`") - sections.append("") - sections.append("") - sections.append("") - sections.append(self.profile_text.strip() or "(No agent profile configured.)") - sections.append("") - sections.append("") - sections.append("") - sections.append(self.workflow_text.strip() or "(No agent workflow configured.)") - sections.append("") - if self.user_preferences_text.strip(): - sections.append("") - sections.append("") - sections.append(self.user_preferences_text.strip()) - sections.append("") + if self.available_personas: + sections.append("- Available personas:") + sections.extend(f" - {persona.summary_line()}" for persona in self.available_personas) + sections.append("") + + if self.warnings: + sections.extend( + [ + "", + "", + *[f"- {warning}" for warning in self.warnings], + "", + ] + ) + + sections.extend( + [ + "", + "", + f"- Persona ID: `{self.persona.persona_id}`", + ] + ) + if self.persona.label and self.persona.label != self.persona.persona_id: + sections.append(f"- Persona Label: {self.persona.label}") + if self.persona.description: + sections.append(f"- Persona Description: {self.persona.description}") + sections.extend( + [ + "", + self.persona.text.strip() or "(No persona instructions configured.)", + "", + ] + ) for path, text in self.extra_contexts: if not text.strip(): continue - sections.append("") - sections.append(f'') - sections.append(text.strip()) - sections.append("") + sections.extend( + [ + "", + f'', + text.strip(), + "", + ] + ) return "\n".join(sections).strip() - def render_hooks_prompt(self) -> str: - """渲染结构化 hooks 提示词。""" - blocks = [ - "", - f"- Hook source: `{self.hooks.path}`", - "- These hooks are loaded structurally by the runtime and must be followed at the matching lifecycle stage.", - "", - "Pre-Task Hooks:", - self._format_hook_list(self.hooks.pre_task), - "", - "In-Task Hooks:", - self._format_hook_list(self.hooks.in_task), - "", - "Post-Task Hooks:", - self._format_hook_list(self.hooks.post_task), - "", - ] - return "\n".join(blocks) - - def render_system_task_message( - self, - task_type: str, - *, - template_context: Optional[dict[str, Any]] = None, - extra_rules: Optional[list[str]] = None, - ) -> str: - """根据统一的后台系统任务定义渲染提示词。""" - task_definition = self.system_tasks.task_types.get(task_type) - if not task_definition: - raise AgentRuntimeConfigError(f"未定义的后台系统任务类型: {task_type}") - - rendered_context = self._render_template_lines( - task_definition.context_lines, - template_context, - task_type, - "context_lines", - ) - rendered_steps = self._render_template_lines( - task_definition.steps, - template_context, - task_type, - "steps", - ) - rendered_task_rules = self._render_template_lines( - task_definition.task_rules, - template_context, - task_type, - "task_rules", - ) - - sections = [ - self._render_template_text( - task_definition.header, - template_context, - task_type, - "header", - ).strip(), - self._render_template_text( - task_definition.objective, - template_context, - task_type, - "objective", - ).strip(), - ] - if rendered_context: - sections.append( - self._format_titled_lines( - task_definition.context_title or "Task context", - rendered_context, - ) - ) - if rendered_steps: - sections.append( - self._format_titled_lines( - task_definition.steps_title or "Follow these steps", - rendered_steps, - ) - ) - - rules = list(self.system_tasks.shared_rules) - if task_definition.empty_result: - rules.append(task_definition.empty_result) - rules.extend(rendered_task_rules) - if extra_rules: - rules.extend(rule.strip() for rule in extra_rules if rule and rule.strip()) - if rules: - sections.append(self._format_numbered_rules("IMPORTANT", rules)) - return "\n\n".join(section for section in sections if section).strip() - - @classmethod - def _render_template_text( - cls, - text: str, - template_context: Optional[dict[str, Any]], - task_type: str, - field_name: str, - ) -> str: - if not text: - return "" - - formatter = Formatter() - required_fields = { - placeholder_name - for _, placeholder_name, _, _ in formatter.parse(text) - if placeholder_name - } - if not required_fields: - return text - - context = cls._normalize_template_context(template_context) - missing_fields = sorted(field for field in required_fields if field not in context) - if missing_fields: - raise AgentRuntimeConfigError( - f"系统任务定义 `{task_type}` 的 `{field_name}` 缺少变量: " - + ", ".join(f"`{field}`" for field in missing_fields) - ) - - # 这里统一做字符串替换，让模板文件成为后台任务文案的唯一行为来源。 - return text.format_map(context) - - @classmethod - def _render_template_lines( - cls, - items: list[str], - template_context: Optional[dict[str, Any]], - task_type: str, - field_name: str, - ) -> list[str]: + def list_personas(self) -> list[dict[str, Any]]: + """返回全部人格摘要。""" return [ - cls._render_template_text( - item, - template_context, - task_type, - f"{field_name}[{index}]", - ).rstrip() - for index, item in enumerate(items, start=1) - if item and item.rstrip() + persona.to_dict(is_active=persona.persona_id == self.active_persona) + for persona in self.available_personas ] - @staticmethod - def _normalize_template_context( - template_context: Optional[dict[str, Any]], - ) -> dict[str, str]: - if not template_context: - return {} - return { - str(key): "" if value is None else str(value) - for key, value in template_context.items() - } - - @staticmethod - def _format_hook_list(items: list[str]) -> str: - if not items: - return "(No hooks configured.)" - return "\n".join(f"{index}. {item}" for index, item in enumerate(items, start=1)) - - @staticmethod - def _format_numbered_rules(title: str, items: list[str]) -> str: - return "\n".join( - [f"{title}:"] - + [f"{index}. {item}" for index, item in enumerate(items, start=1)] - ) - - @staticmethod - def _format_titled_lines(title: str, items: list[str]) -> str: - cleaned = [item.rstrip() for item in items if item and item.rstrip()] - return "\n".join([f"{title}:"] + cleaned) - class AgentRuntimeManager: - """统一管理 agent 根层配置目录、迁移、校验与模板渲染。""" + """统一管理 agent 根层运行时配置目录、校验与人格切换。""" def __init__( self, *, agent_root_dir: Optional[Path] = None, - bundled_runtime_dir: Optional[Path] = None, + bundled_defaults_dir: Optional[Path] = None, ) -> None: self.agent_root_dir = agent_root_dir or (settings.CONFIG_PATH / "agent") self.runtime_dir = self.agent_root_dir / SYSTEM_RUNTIME_DIR @@ -325,15 +201,15 @@ class AgentRuntimeManager: self.skills_dir = self.agent_root_dir / SKILLS_DIR self.jobs_dir = self.agent_root_dir / JOBS_DIR self.activity_dir = self.agent_root_dir / ACTIVITY_DIR - self.bundled_runtime_dir = bundled_runtime_dir or ( - Path(__file__).parent / "runtime_defaults" + self.bundled_defaults_dir = bundled_defaults_dir or ( + Path(__file__).parent / "defaults" ) self._cache_lock = threading.Lock() self._cached_signature: Optional[tuple[tuple[str, int, int], ...]] = None self._cached_config: Optional[AgentRuntimeConfig] = None def ensure_layout(self) -> None: - """创建目录、同步默认文件，并迁移旧版 memory/runtime 文件。""" + """创建目录、同步默认文件，并清理废弃的旧版 runtime 文件。""" self.agent_root_dir.mkdir(parents=True, exist_ok=True) self.runtime_dir.mkdir(parents=True, exist_ok=True) self.memory_dir.mkdir(parents=True, exist_ok=True) @@ -341,7 +217,8 @@ class AgentRuntimeManager: self.jobs_dir.mkdir(parents=True, exist_ok=True) self.activity_dir.mkdir(parents=True, exist_ok=True) self._migrate_root_runtime_files() - self._sync_bundled_runtime_defaults() + self._remove_obsolete_runtime_files() + self._sync_bundled_defaults() self._migrate_root_memory_files() def load_runtime_config(self) -> AgentRuntimeConfig: @@ -356,7 +233,7 @@ class AgentRuntimeManager: config = self._load_from_root(self.runtime_dir) except AgentRuntimeConfigError as err: logger.warning("Agent 根层配置无效，回退到内置默认配置: %s", err) - config = self._load_from_root(self.bundled_runtime_dir) + config = self._load_from_root(self.bundled_defaults_dir) config.used_fallback = True config.warnings.insert( 0, f"用户运行时配置加载失败，已回退到内置默认配置: {err}" @@ -372,10 +249,128 @@ class AgentRuntimeManager: self._cached_signature = None self._cached_config = None + def set_active_persona(self, persona_query: str) -> AgentRuntimeConfig: + """切换当前激活人格，并立即刷新缓存。""" + self.ensure_layout() + runtime_root = self.runtime_dir + current_path = runtime_root / CURRENT_PERSONA_FILE + current_doc = self._read_markdown(current_path) + current_meta = current_doc.metadata + + available_personas = self._load_personas(runtime_root) + persona = self._resolve_persona_definition(persona_query, available_personas) + + document = self._render_current_persona_document( + active_persona=persona.persona_id, + extra_context_files=self._coerce_string_list( + current_meta.get("extra_context_files") + ), + deprecated_phrases=self._coerce_string_list( + current_meta.get("deprecated_phrases") + ), + ) + current_path.write_text(document, encoding="utf-8") + self.invalidate_cache() + logger.info("已切换 Agent 人格: %s", persona.persona_id) + return self.load_runtime_config() + + def list_personas(self) -> list[PersonaDefinition]: + """列出当前可用人格。""" + return self.load_runtime_config().available_personas + + def update_persona_definition( + self, + persona_query: str, + *, + label: Optional[str] = None, + description: Optional[str] = None, + aliases: Optional[list[str]] = None, + instructions: Optional[str] = None, + append_instructions: Optional[list[str]] = None, + create_if_missing: bool = False, + ) -> tuple[PersonaDefinition, bool]: + """更新或创建运行时人格定义。""" + self.ensure_layout() + runtime_root = self.runtime_dir + available_personas = self._load_personas(runtime_root) + + created = False + try: + persona = self._resolve_persona_definition(persona_query, available_personas) + target_persona_id = persona.persona_id + target_path = persona.path + existing_body = persona.text + existing_label = persona.label + existing_description = persona.description + existing_aliases = list(persona.aliases) + except AgentRuntimeConfigError: + if not create_if_missing: + raise + target_persona_id = self._validate_new_persona_id(persona_query) + target_path = runtime_root / PERSONAS_DIR / target_persona_id / PERSONA_FILE + existing_body = "" + existing_label = target_persona_id + existing_description = "" + existing_aliases = [] + created = True + + final_label = ( + label.strip() + if isinstance(label, str) and label.strip() + else existing_label or target_persona_id + ) + final_description = ( + description.strip() + if isinstance(description, str) and description.strip() + else existing_description + ) + final_aliases = ( + self._normalize_persona_aliases(aliases, "aliases") + if aliases is not None + else existing_aliases + ) + final_body = ( + self._normalize_persona_body(instructions) + if isinstance(instructions, str) and instructions.strip() + else self._normalize_persona_body(existing_body) + ) + final_body = self._merge_persona_instructions( + final_body, + append_instructions, + ) + if not final_body.strip(): + raise AgentRuntimeConfigError("人格定义正文不能为空") + + document = self._render_persona_document( + persona_id=target_persona_id, + label=final_label, + description=final_description, + aliases=final_aliases, + body=final_body, + ) + target_path.parent.mkdir(parents=True, exist_ok=True) + target_path.write_text(document, encoding="utf-8") + self.invalidate_cache() + + runtime_config = self.load_runtime_config() + updated_persona = self._resolve_persona_definition( + target_persona_id, + runtime_config.available_personas, + ) + logger.info( + "已%s Agent 人格定义: %s", + "创建" if created else "更新", + updated_persona.persona_id, + ) + return updated_persona, created + def _build_signature(self) -> tuple[tuple[str, int, int], ...]: - """基于运行时配置和内置默认配置生成文件签名。""" + """基于运行时配置和内置人格生成文件签名。""" entries: list[tuple[str, int, int]] = [] - for prefix, root in (("runtime", self.runtime_dir), ("bundled", self.bundled_runtime_dir)): + for prefix, root in ( + ("runtime", self.runtime_dir), + ("bundled", self.bundled_defaults_dir), + ): if not root.exists(): continue for path in sorted(root.rglob("*")): @@ -386,12 +381,12 @@ class AgentRuntimeManager: entries.append((f"{prefix}:{relative}", stat.st_mtime_ns, stat.st_size)) return tuple(entries) - def _sync_bundled_runtime_defaults(self) -> None: - """仅复制缺失的默认根层配置，避免覆盖用户自定义。""" - if not self.bundled_runtime_dir.exists(): + def _sync_bundled_defaults(self) -> None: + """仅复制缺失的默认运行时文件，避免覆盖用户自定义。""" + if not self.bundled_defaults_dir.exists(): return - for path in sorted(self.bundled_runtime_dir.rglob("*")): - relative = path.relative_to(self.bundled_runtime_dir) + for path in sorted(self.bundled_defaults_dir.rglob("*")): + relative = path.relative_to(self.bundled_defaults_dir) target = self.runtime_dir / relative if path.is_dir(): target.mkdir(parents=True, exist_ok=True) @@ -403,23 +398,30 @@ class AgentRuntimeManager: logger.info("已同步默认 Agent 运行时文件: %s", target) def _migrate_root_runtime_files(self) -> None: - """兼容早期直接放在 `config/agent` 根目录的 RFC 文件。""" - migration_targets = { - CURRENT_PERSONA_FILE: self.runtime_dir / CURRENT_PERSONA_FILE, - USER_PREFERENCES_FILE: self.runtime_dir / USER_PREFERENCES_FILE, - SYSTEM_TASKS_FILE: self.runtime_dir / "system_tasks" / SYSTEM_TASKS_FILE, - LEGACY_WAKE_FORMAT_FILE: self.runtime_dir / "system_tasks" / SYSTEM_TASKS_FILE, - "AGENT_PROFILE.md": self.runtime_dir / "personas" / "default" / "AGENT_PROFILE.md", - "AGENT_WORKFLOW.md": self.runtime_dir / "personas" / "default" / "AGENT_WORKFLOW.md", - "AGENT_HOOKS.md": self.runtime_dir / "personas" / "default" / "AGENT_HOOKS.md", - } - for filename, target in migration_targets.items(): - source = self.agent_root_dir / filename - if not source.exists() or target.exists(): + """兼容早期直接放在 `config/agent` 根目录的 CURRENT_PERSONA。""" + source = self.agent_root_dir / CURRENT_PERSONA_FILE + target = self.runtime_dir / CURRENT_PERSONA_FILE + if not source.exists() or target.exists(): + return + target.parent.mkdir(parents=True, exist_ok=True) + source.rename(target) + logger.info("已迁移旧版 Agent 根配置文件: %s -> %s", source, target) + + def _remove_obsolete_runtime_files(self) -> None: + """删除不再支持的旧版 Agent 配置文件，避免被误迁移到 memory。""" + for filename in sorted(OBSOLETE_AGENT_ROOT_FILES): + path = self.agent_root_dir / filename + if not path.exists() or not path.is_file(): continue - target.parent.mkdir(parents=True, exist_ok=True) - source.rename(target) - logger.info("已迁移旧版 Agent 根配置文件: %s -> %s", source, target) + path.unlink() + logger.info("已删除废弃的 Agent 根配置文件: %s", path) + + for relative_path in sorted(OBSOLETE_RUNTIME_FILES): + path = self.runtime_dir / relative_path + if not path.exists() or not path.is_file(): + continue + path.unlink() + logger.info("已删除废弃的 Agent 运行时文件: %s", path) def _migrate_root_memory_files(self) -> None: """将旧版根目录 memory 文件移入 `config/agent/memory`。""" @@ -437,72 +439,114 @@ class AgentRuntimeManager: current_doc = self._read_markdown(current_persona_path) current_meta = current_doc.metadata - active_persona = str(current_meta.get("active_persona") or "default").strip() + active_persona = str( + current_meta.get("active_persona") or DEFAULT_PERSONA_ID + ).strip() if not active_persona: raise AgentRuntimeConfigError("CURRENT_PERSONA.md 缺少 active_persona") - profile_path = self._resolve_required_path(root, current_meta, "profile") - workflow_path = self._resolve_required_path(root, current_meta, "workflow") - hooks_path = self._resolve_required_path(root, current_meta, "hooks") - system_tasks_path = self._resolve_required_path(root, current_meta, "system_tasks") - user_preferences_path = self._resolve_optional_path( - root, current_meta.get("user_preferences") - ) extra_context_paths = self._resolve_optional_paths( root, current_meta.get("extra_context_files", []) ) - profile_doc = self._read_markdown(profile_path) - workflow_doc = self._read_markdown(workflow_path) - hooks_doc = self._read_markdown(hooks_path) - system_tasks_doc = self._read_markdown(system_tasks_path) - preferences_doc = ( - self._read_markdown(user_preferences_path) - if user_preferences_path and user_preferences_path.exists() - else ParsedMarkdownDocument(metadata={}, body="") - ) + available_personas = self._load_personas(root) + persona = self._resolve_persona_definition(active_persona, available_personas) extra_contexts = [ (path, self._read_markdown(path).body) for path in extra_context_paths ] - hooks = self._parse_hooks_document(hooks_path, hooks_doc) - system_tasks = self._parse_system_tasks_document( - system_tasks_path, - system_tasks_doc, - ) - warnings = self._validate_runtime_config( current_meta=current_meta, - profile_path=profile_path, - workflow_path=workflow_path, - hooks_path=hooks_path, - user_preferences_path=user_preferences_path, - system_tasks_path=system_tasks_path, + persona_path=persona.path, extra_context_paths=extra_context_paths, - profile_text=profile_doc.body, - workflow_text=workflow_doc.body, - preferences_text=preferences_doc.body, + persona_text=persona.text, ) return AgentRuntimeConfig( source_root=root, active_persona=active_persona, current_persona_path=current_persona_path, - profile_path=profile_path, - workflow_path=workflow_path, - hooks_path=hooks_path, - user_preferences_path=user_preferences_path, - system_tasks_path=system_tasks_path, + persona=persona, + available_personas=available_personas, extra_context_paths=extra_context_paths, - profile_text=profile_doc.body, - workflow_text=workflow_doc.body, - user_preferences_text=preferences_doc.body, extra_contexts=extra_contexts, - hooks=hooks, - system_tasks=system_tasks, warnings=warnings, ) + def _load_personas(self, root: Path) -> list[PersonaDefinition]: + """扫描并解析所有可用人格。""" + personas_root = root / PERSONAS_DIR + if not personas_root.exists(): + raise AgentRuntimeConfigError(f"缺少 personas 目录: {personas_root}") + + personas: list[PersonaDefinition] = [] + seen_ids: set[str] = set() + for persona_dir in sorted(personas_root.iterdir()): + if not persona_dir.is_dir(): + continue + persona_path = persona_dir / PERSONA_FILE + if not persona_path.exists(): + continue + document = self._read_markdown(persona_path) + persona_id = str(document.metadata.get("persona_id") or persona_dir.name).strip() + if not persona_id: + raise AgentRuntimeConfigError(f"{persona_path} 缺少 persona_id") + if persona_id in seen_ids: + raise AgentRuntimeConfigError(f"检测到重复的人格 ID: {persona_id}") + seen_ids.add(persona_id) + aliases = self._normalize_string_list( + document.metadata.get("aliases"), + f"{persona_path}.aliases", + ) + personas.append( + PersonaDefinition( + persona_id=persona_id, + path=persona_path, + label=str(document.metadata.get("label") or persona_id).strip(), + description=str(document.metadata.get("description") or "").strip(), + text=document.body, + aliases=aliases, + ) + ) + + if not personas: + raise AgentRuntimeConfigError(f"{personas_root} 中未找到任何人格定义") + return personas + + @staticmethod + def _resolve_persona_definition( + persona_query: str, + personas: list[PersonaDefinition], + ) -> PersonaDefinition: + """按 persona_id、label 或 aliases 解析人格。""" + normalized = (persona_query or "").strip() + if not normalized: + raise AgentRuntimeConfigError("人格 ID 不能为空") + + for persona in personas: + if persona.persona_id == normalized: + return persona + for persona in personas: + if persona.matches(normalized): + return persona + + available = ", ".join(persona.persona_id for persona in personas) + raise AgentRuntimeConfigError( + f"未找到人格 `{persona_query}`，可用人格: {available}" + ) + + @staticmethod + def _validate_new_persona_id(persona_id: str) -> str: + """校验新建人格的 ID，避免写入非法路径。""" + normalized = (persona_id or "").strip() + if not normalized: + raise AgentRuntimeConfigError("新建人格时 persona_id 不能为空") + if not PERSONA_ID_PATTERN.fullmatch(normalized): + raise AgentRuntimeConfigError( + "新建人格时 persona_id 只能使用小写字母、数字、下划线和中划线，且必须以字母或数字开头" + ) + return normalized + @staticmethod def _read_markdown(path: Path) -> ParsedMarkdownDocument: if not path.exists(): @@ -519,25 +563,14 @@ class AgentRuntimeManager: try: metadata = yaml.safe_load(match.group(1)) or {} except yaml.YAMLError as err: - raise AgentRuntimeConfigError(f"YAML frontmatter 解析失败 {path}: {err}") from err + raise AgentRuntimeConfigError( + f"YAML frontmatter 解析失败 {path}: {err}" + ) from err if not isinstance(metadata, dict): raise AgentRuntimeConfigError(f"frontmatter 必须是映射类型: {path}") body = content[match.end():] return ParsedMarkdownDocument(metadata=metadata, body=body.strip()) - @staticmethod - def _resolve_required_path(root: Path, metadata: dict[str, Any], field_name: str) -> Path: - raw = metadata.get(field_name) - if not raw or not str(raw).strip(): - raise AgentRuntimeConfigError(f"CURRENT_PERSONA.md 缺少必填字段 `{field_name}`") - return AgentRuntimeManager._resolve_relative_path(root, str(raw)) - - @staticmethod - def _resolve_optional_path(root: Path, raw: Any) -> Optional[Path]: - if not raw or not str(raw).strip(): - return None - return AgentRuntimeManager._resolve_relative_path(root, str(raw)) - @staticmethod def _resolve_optional_paths(root: Path, values: Any) -> list[Path]: if not values: @@ -564,124 +597,72 @@ class AgentRuntimeManager: normalized.append(text) return normalized - def _parse_hooks_document( - self, path: Path, document: ParsedMarkdownDocument - ) -> HookDefinition: - pre_task = self._normalize_string_list(document.metadata.get("pre_task"), "pre_task") - in_task = self._normalize_string_list(document.metadata.get("in_task"), "in_task") - post_task = self._normalize_string_list( - document.metadata.get("post_task"), "post_task" - ) - if not (pre_task or in_task or post_task): - raise AgentRuntimeConfigError(f"{path} 未定义任何结构化 hooks") - return HookDefinition( - path=path, - pre_task=pre_task, - in_task=in_task, - post_task=post_task, - ) - - def _parse_system_tasks_document( - self, path: Path, document: ParsedMarkdownDocument - ) -> SystemTasksDefinition: - """解析后台系统任务定义文件。""" - version = self._normalize_positive_int( - document.metadata.get("version"), - "version", - default=1, - ) - if version < SYSTEM_TASKS_SCHEMA_VERSION: - raise AgentRuntimeConfigError( - f"{path} 的 version={version} 过旧，" - f"当前要求 SYSTEM_TASKS schema v{SYSTEM_TASKS_SCHEMA_VERSION} 或更高版本" - ) - shared_rules = self._normalize_string_list( - document.metadata.get("shared_rules"), "shared_rules" - ) - if not shared_rules: - raise AgentRuntimeConfigError(f"{path} 缺少 shared_rules") - - raw_task_types = document.metadata.get("task_types") - if not isinstance(raw_task_types, dict) or not raw_task_types: - raise AgentRuntimeConfigError(f"{path} 缺少 task_types 映射") - - task_types: dict[str, SystemTaskTypeDefinition] = {} - for key, raw in raw_task_types.items(): - if not isinstance(raw, dict): - raise AgentRuntimeConfigError(f"task_types.{key} 必须是映射") - header = str(raw.get("header") or "").strip() - objective = str(raw.get("objective") or "").strip() - if not header or not objective: - raise AgentRuntimeConfigError( - f"task_types.{key} 缺少 header 或 objective" - ) - context_lines = self._normalize_string_list( - raw.get("context_lines"), - f"task_types.{key}.context_lines", - ) - steps = self._normalize_string_list( - raw.get("steps"), - f"task_types.{key}.steps", - ) - task_rules = self._normalize_string_list( - raw.get("task_rules"), - f"task_types.{key}.task_rules", - ) - empty_result = str(raw.get("empty_result") or "").strip() or None - context_title = str(raw.get("context_title") or "").strip() or None - steps_title = str(raw.get("steps_title") or "").strip() or None - task_types[str(key)] = SystemTaskTypeDefinition( - header=header, - objective=objective, - context_title=context_title, - context_lines=context_lines, - steps_title=steps_title, - steps=steps, - task_rules=task_rules, - empty_result=empty_result, - ) - return SystemTasksDefinition( - path=path, - version=version, - shared_rules=shared_rules, - task_types=task_types, - ) + @staticmethod + def _coerce_string_list(values: Any) -> list[str]: + if not isinstance(values, list): + return [] + return [str(value).strip() for value in values if str(value).strip()] @staticmethod - def _normalize_positive_int( - value: Any, - field_name: str, - *, - default: int, - ) -> int: - if value in (None, ""): - return default - try: - normalized = int(value) - except (TypeError, ValueError) as err: - raise AgentRuntimeConfigError(f"{field_name} 必须是正整数") from err - if normalized <= 0: - raise AgentRuntimeConfigError(f"{field_name} 必须是正整数") + def _normalize_persona_aliases(values: Any, field_name: str) -> list[str]: + """规范化人格别名，保持顺序并去重。""" + normalized = AgentRuntimeManager._normalize_string_list(values, field_name) + deduped: list[str] = [] + seen: set[str] = set() + for alias in normalized: + folded = alias.casefold() + if folded in seen: + continue + seen.add(folded) + deduped.append(alias) + return deduped + + @staticmethod + def _merge_persona_instructions( + base_body: str, + append_instructions: Optional[list[str]], + ) -> str: + """把增量规则安全追加到人格正文末尾。""" + merged = (base_body or "").strip() + if not append_instructions: + return merged + + extras: list[str] = [] + for item in append_instructions: + text = str(item).strip() + if not text: + continue + if not re.match(r"^([-*]|\d+\.)\s", text): + text = f"- {text}" + extras.append(text) + + if not extras: + return merged + if not merged: + return "\n".join(extras) + return merged.rstrip() + "\n\n" + "\n".join(extras) + + @staticmethod + def _normalize_persona_body(body: Optional[str]) -> str: + """去掉重复的 PERSONA 标题，保持正文可安全回写。""" + normalized = (body or "").strip() + if not normalized: + return "" + if normalized.startswith("# PERSONA"): + _, _, remainder = normalized.partition("\n") + return remainder.strip() return normalized def _validate_runtime_config( self, *, current_meta: dict[str, Any], - profile_path: Path, - workflow_path: Path, - hooks_path: Path, - user_preferences_path: Optional[Path], - system_tasks_path: Path, + persona_path: Path, extra_context_paths: list[Path], - profile_text: str, - workflow_text: str, - preferences_text: str, + persona_text: str, ) -> list[str]: warnings: list[str] = [] - required_paths = [profile_path, workflow_path, hooks_path, system_tasks_path] - if user_preferences_path: - required_paths.append(user_preferences_path) + required_paths = [persona_path] duplicates = self._find_duplicate_paths(required_paths + extra_context_paths) if duplicates: warnings.append( @@ -693,17 +674,9 @@ class AgentRuntimeManager: current_meta.get("deprecated_phrases"), "deprecated_phrases" ) if deprecated_phrases: - scan_targets = { - "profile": profile_text, - "workflow": workflow_text, - "user_preferences": preferences_text, - } for phrase in deprecated_phrases: - for target_name, text in scan_targets.items(): - if phrase and phrase in text: - warnings.append( - f"检测到已废弃短语 `{phrase}` 仍出现在 {target_name} 中" - ) + if phrase and phrase in persona_text: + warnings.append(f"检测到已废弃短语 `{phrase}` 仍出现在 persona 中") return warnings @staticmethod @@ -717,5 +690,66 @@ class AgentRuntimeManager: seen.add(resolved) return duplicates + @staticmethod + def _render_current_persona_document( + *, + active_persona: str, + extra_context_files: list[str], + deprecated_phrases: list[str], + ) -> str: + """统一生成 CURRENT_PERSONA.md，避免手写时结构漂移。""" + metadata = { + "version": CURRENT_PERSONA_SCHEMA_VERSION, + "active_persona": active_persona, + "extra_context_files": extra_context_files, + "deprecated_phrases": deprecated_phrases, + } + body_lines = [ + "# CURRENT_PERSONA", + "", + f"当前激活人格：`{active_persona}`", + "", + "运行时加载顺序固定如下：", + "", + "1. 核心系统提示词（程序内置，不可运行时覆盖）", + "2. `personas//PERSONA.md`", + "3. `extra_context_files`", + "4. `memory/*.md`", + "5. `activity/*.md`", + "", + "`memory` 中的长期偏好可以细化回复方式，但不应覆盖系统核心身份、目标和安全边界。", + ] + frontmatter = yaml.safe_dump( + metadata, + sort_keys=False, + allow_unicode=True, + ).strip() + return f"---\n{frontmatter}\n---\n" + "\n".join(body_lines) + "\n" + + @staticmethod + def _render_persona_document( + *, + persona_id: str, + label: str, + description: str, + aliases: list[str], + body: str, + ) -> str: + """统一生成人格定义文件，避免手写 frontmatter 漂移。""" + metadata = { + "version": PERSONA_SCHEMA_VERSION, + "persona_id": persona_id, + "label": label, + "description": description, + "aliases": aliases, + } + frontmatter = yaml.safe_dump( + metadata, + sort_keys=False, + allow_unicode=True, + ).strip() + normalized_body = AgentRuntimeManager._normalize_persona_body(body) + return f"---\n{frontmatter}\n---\n# PERSONA\n\n{normalized_body}\n" + agent_runtime_manager = AgentRuntimeManager() diff --git a/app/agent/runtime_defaults/CURRENT_PERSONA.md b/app/agent/runtime_defaults/CURRENT_PERSONA.md deleted file mode 100644 index 843b8a15..00000000 --- a/app/agent/runtime_defaults/CURRENT_PERSONA.md +++ /dev/null @@ -1,24 +0,0 @@ ---- -version: 1 -active_persona: default -profile: personas/default/AGENT_PROFILE.md -workflow: personas/default/AGENT_WORKFLOW.md -hooks: personas/default/AGENT_HOOKS.md -user_preferences: USER_PREFERENCES.md -system_tasks: system_tasks/SYSTEM_TASKS.md -extra_context_files: [] -deprecated_phrases: [] ---- -# CURRENT_PERSONA - -当前激活人格：`default` - -加载顺序固定如下： - -1. `AGENT_PROFILE.md` -2. `AGENT_WORKFLOW.md` -3. `AGENT_HOOKS.md` -4. `USER_PREFERENCES.md` -5. `SYSTEM_TASKS.md` - -如果需要扩展额外上下文，请使用 `extra_context_files` 显式声明，而不是把额外规则散落到 memory 中。 diff --git a/app/agent/runtime_defaults/USER_PREFERENCES.md b/app/agent/runtime_defaults/USER_PREFERENCES.md deleted file mode 100644 index adec6b67..00000000 --- a/app/agent/runtime_defaults/USER_PREFERENCES.md +++ /dev/null @@ -1,10 +0,0 @@ ---- -version: 1 ---- -# USER_PREFERENCES - -这是根层的运维偏好文件，不是用户长期记忆。 - -- 这里只放稳定的系统级输出规则或部署方偏好。 -- 用户在对话中形成的长期习惯，仍应写入 `config/agent/memory/*.md`。 -- 默认保持精简，避免与 `AGENT_PROFILE.md` 或 `AGENT_WORKFLOW.md` 重复。 diff --git a/app/agent/runtime_defaults/personas/default/AGENT_HOOKS.md b/app/agent/runtime_defaults/personas/default/AGENT_HOOKS.md deleted file mode 100644 index 8c3a957b..00000000 --- a/app/agent/runtime_defaults/personas/default/AGENT_HOOKS.md +++ /dev/null @@ -1,26 +0,0 @@ ---- -version: 1 -pre_task: - - Identify whether the request is a normal user conversation or a background system task before choosing a workflow. - - Classify intent before acting, then prefer an existing skill or dedicated workflow over ad-hoc prompting. - - Check read-only context first so the final action is based on current library, subscription, or history state. - - Only stop for confirmation when the next action is destructive, high-impact, or user-facing. - - Keep the final delivery target explicit before calling tools. -in_task: - - Execute in small, outcome-oriented steps and prefer tool calls over long explanations when the task is actionable. - - Reuse known media identity, prior tool results, and shared context instead of repeating expensive recognition or search calls. - - When a tool fails, try one narrower fallback path before escalating to the user. - - Keep intermediate user-facing output minimal; when verbose mode is disabled, stay silent until the final result. - - Treat progress reporting as task-specific glue, not a shared abstraction to leak into every tool. -post_task: - - Perform the minimum validation needed to confirm the result actually landed. - - Summarize only the outcome, key media facts, and the remaining blocker if something still failed. - - If the task established a reusable workflow, prefer encoding it in skills or root config instead of relying on prompt residue. ---- -# AGENT_HOOKS - -这些 hooks 由运行时结构化加载，不依赖自由文本约定。 - -- `pre_task` 对应开始执行前的统一检查点。 -- `in_task` 对应工具调用和失败降级阶段。 -- `post_task` 对应最小验证与收口阶段。 diff --git a/app/agent/runtime_defaults/personas/default/AGENT_PROFILE.md b/app/agent/runtime_defaults/personas/default/AGENT_PROFILE.md deleted file mode 100644 index c15d94b7..00000000 --- a/app/agent/runtime_defaults/personas/default/AGENT_PROFILE.md +++ /dev/null @@ -1,27 +0,0 @@ ---- -version: 1 ---- -# AGENT_PROFILE - -- Identity: You are an AI media assistant powered by MoviePilot. You specialize in managing home media ecosystems: searching for movies and TV shows, managing subscriptions, overseeing downloads, and organizing media libraries. -- Tone: professional, concise, restrained. -- Be direct. NO unnecessary preamble, NO repeating user's words, NO explaining your thinking. -- Prioritize task progress over conversation. Answer only what is necessary to move the task forward. -- Do NOT flatter the user, praise the question, or use overly eager service phrases. -- Do NOT use emojis, exclamation marks, cute language, or excessive apology. -- Prefer short declarative sentences. Default to one or two short paragraphs; use lists only when they improve scanability. -- Use Markdown for structured data. Use `inline code` for media titles and paths. -- Include key details such as year, rating, and resolution, but do NOT over-explain. -- Do not stop for approval on read-only operations. Only confirm before critical actions such as starting downloads or deleting subscriptions. -- NOT a coding assistant. Do not offer code snippets. -- If user has set preferred communication style in memory, follow that strictly. - -# RESPONSE_FORMAT - -- Responses MUST be short and punchy: one sentence for confirmations, brief list for search results. -- NO filler phrases like "Let me help you", "Here are the results", "I found..." - skip all unnecessary preamble. -- NO repeating what user said. -- NO narrating your internal reasoning. -- NO praise, emotional cushioning, or unnecessary politeness padding. -- After task completion: one line summary only. -- When error occurs: brief acknowledgment plus suggestion, then move on. diff --git a/app/agent/runtime_defaults/personas/default/AGENT_WORKFLOW.md b/app/agent/runtime_defaults/personas/default/AGENT_WORKFLOW.md deleted file mode 100644 index 1eb8c685..00000000 --- a/app/agent/runtime_defaults/personas/default/AGENT_WORKFLOW.md +++ /dev/null @@ -1,25 +0,0 @@ ---- -version: 1 ---- -# AGENT_WORKFLOW - -## FLOW - -1. Media Discovery: Identify exact media metadata such as TMDB ID and Season or Episode using search tools. -2. Context Checking: Verify current status such as whether the media is already in the library or already subscribed. -3. Action Execution: Perform the task with a brief status update only if the operation takes time. -4. Final Confirmation: State the result concisely. - -## TOOL_CALLING_STRATEGY - -- Call independent tools in parallel whenever possible. -- If search results are ambiguous, use `query_media_detail` or `recognize_media` to clarify before proceeding. -- If `search_media` fails, fall back to `search_web` or `recognize_media`. Only ask the user when all automated methods are exhausted. - -## MEDIA_MANAGEMENT_RULES - -1. Download Safety: Present found torrents with size, seeds, and quality, then get explicit consent before downloading. -2. Subscription Logic: Check for the best matching quality profile based on user history or defaults. -3. Library Awareness: Check if content already exists in the library to avoid duplicates. -4. Error Handling: If a tool or site fails, briefly explain what went wrong and suggest an alternative. -5. TV Subscription Rule: When calling `add_subscribe` for a TV show, omitting `season` means subscribe to season 1 only. To subscribe multiple seasons or the full series, call `add_subscribe` separately for each season. diff --git a/app/agent/tools/factory.py b/app/agent/tools/factory.py index 9d992031..61653f53 100644 --- a/app/agent/tools/factory.py +++ b/app/agent/tools/factory.py @@ -37,6 +37,9 @@ from app.agent.tools.impl.query_schedulers import QuerySchedulersTool from app.agent.tools.impl.run_scheduler import RunSchedulerTool from app.agent.tools.impl.query_workflows import QueryWorkflowsTool from app.agent.tools.impl.run_workflow import RunWorkflowTool +from app.agent.tools.impl.query_personas import QueryPersonasTool +from app.agent.tools.impl.switch_persona import SwitchPersonaTool +from app.agent.tools.impl.update_persona_definition import UpdatePersonaDefinitionTool from app.agent.tools.impl.update_site_cookie import UpdateSiteCookieTool from app.agent.tools.impl.delete_download import DeleteDownloadTool from app.agent.tools.impl.delete_download_history import DeleteDownloadHistoryTool @@ -146,6 +149,9 @@ class MoviePilotToolFactory: RunSchedulerTool, QueryWorkflowsTool, RunWorkflowTool, + QueryPersonasTool, + SwitchPersonaTool, + UpdatePersonaDefinitionTool, ExecuteCommandTool, EditFileTool, WriteFileTool, diff --git a/app/agent/tools/impl/query_personas.py b/app/agent/tools/impl/query_personas.py new file mode 100644 index 00000000..02a015d0 --- /dev/null +++ b/app/agent/tools/impl/query_personas.py @@ -0,0 +1,75 @@ +"""查询可用人格工具。""" + +import json +from typing import Optional, Type + +from pydantic import BaseModel, Field + +from app.agent.runtime import agent_runtime_manager +from app.agent.tools.base import MoviePilotTool +from app.log import logger + + +class QueryPersonasInput(BaseModel): + """查询人格工具的输入参数模型。""" + + explanation: str = Field( + ..., + description="Clear explanation of why this tool is being used in the current context", + ) + query: Optional[str] = Field( + None, + description=( + "Optional search keyword for persona_id, label, description, or aliases. " + "Use this when the user asks for a certain speaking style but the exact persona name is unknown." + ), + ) + + +class QueryPersonasTool(MoviePilotTool): + name: str = "query_personas" + description: str = ( + "List all available personas (人格) and show which one is currently active. " + "Use this before switching persona when the user asks for a different speaking style but does not name " + "an exact persona_id. The result includes persona_id, label, description, aliases, and whether it is active." + ) + args_schema: Type[BaseModel] = QueryPersonasInput + + def get_tool_message(self, **kwargs) -> Optional[str]: + query = kwargs.get("query") + if query: + return f"查询人格列表: {query}" + return "查询人格列表" + + async def run(self, query: Optional[str] = None, **kwargs) -> str: + logger.info("执行工具: %s, 参数: query=%s", self.name, query) + try: + runtime_config = agent_runtime_manager.load_runtime_config() + personas = runtime_config.list_personas() + + if query: + normalized = query.strip().casefold() + personas = [ + persona + for persona in personas + if normalized in persona["persona_id"].casefold() + or normalized in persona["label"].casefold() + or normalized in persona["description"].casefold() + or any(normalized in alias.casefold() for alias in persona["aliases"]) + ] + + payload = { + "active_persona": runtime_config.active_persona, + "count": len(personas), + "personas": personas, + } + return json.dumps(payload, ensure_ascii=False, indent=2) + except Exception as e: # noqa: BLE001 + logger.error("查询人格列表失败: %s", e, exc_info=True) + return json.dumps( + { + "success": False, + "message": f"查询人格列表时发生错误: {str(e)}", + }, + ensure_ascii=False, + ) diff --git a/app/agent/tools/impl/switch_persona.py b/app/agent/tools/impl/switch_persona.py new file mode 100644 index 00000000..2300d84d --- /dev/null +++ b/app/agent/tools/impl/switch_persona.py @@ -0,0 +1,62 @@ +"""切换当前激活人格工具。""" + +import json +from typing import Type + +from pydantic import BaseModel, Field + +from app.agent.runtime import agent_runtime_manager +from app.agent.tools.base import MoviePilotTool +from app.log import logger + + +class SwitchPersonaInput(BaseModel): + """切换人格工具的输入参数模型。""" + + explanation: str = Field( + ..., + description="Clear explanation of why this tool is being used in the current context", + ) + persona_id: str = Field( + ..., + description=( + "The target persona to activate. This can be the exact persona_id, label, or one of the persona aliases. " + "If the exact persona is unclear, call query_personas first." + ), + ) + + +class SwitchPersonaTool(MoviePilotTool): + name: str = "switch_persona" + description: str = ( + "Switch the active persona (人格) used by the agent runtime. " + "This change is persistent for future turns. " + "Use this when the user explicitly asks to change the speaking style, tone, or response persona. " + "If the user asks for a vague style and you are not sure which persona matches best, call query_personas first." + ) + args_schema: Type[BaseModel] = SwitchPersonaInput + + def get_tool_message(self, **kwargs) -> str: + persona_id = kwargs.get("persona_id") or "未知人格" + return f"切换人格: {persona_id}" + + async def run(self, persona_id: str, **kwargs) -> str: + logger.info("执行工具: %s, 参数: persona_id=%s", self.name, persona_id) + try: + runtime_config = agent_runtime_manager.set_active_persona(persona_id) + payload = { + "success": True, + "active_persona": runtime_config.active_persona, + "persona": runtime_config.persona.to_dict(is_active=True), + "message": f"已切换为人格 `{runtime_config.active_persona}`", + } + return json.dumps(payload, ensure_ascii=False, indent=2) + except Exception as e: # noqa: BLE001 + logger.error("切换人格失败: %s", e, exc_info=True) + return json.dumps( + { + "success": False, + "message": f"切换人格时发生错误: {str(e)}", + }, + ensure_ascii=False, + ) diff --git a/app/agent/tools/impl/update_persona_definition.py b/app/agent/tools/impl/update_persona_definition.py new file mode 100644 index 00000000..678c76fe --- /dev/null +++ b/app/agent/tools/impl/update_persona_definition.py @@ -0,0 +1,131 @@ +"""更新人格定义工具。""" + +import json +from typing import Optional, Type + +from pydantic import BaseModel, Field + +from app.agent.runtime import agent_runtime_manager +from app.agent.tools.base import MoviePilotTool +from app.log import logger + + +class UpdatePersonaDefinitionInput(BaseModel): + """更新人格定义工具的输入参数模型。""" + + explanation: str = Field( + ..., + description="Clear explanation of why this tool is being used in the current context", + ) + persona_id: str = Field( + ..., + description=( + "Target persona to update. For existing personas this can be persona_id, label, or alias. " + "For new personas, provide the new lowercase persona_id." + ), + ) + label: Optional[str] = Field( + None, + description="Optional new label shown to users, such as 默认 or 说明型.", + ) + description: Optional[str] = Field( + None, + description="Optional short description of the persona's intended style.", + ) + aliases: Optional[list[str]] = Field( + None, + description="Optional full replacement list of aliases for this persona.", + ) + instructions: Optional[str] = Field( + None, + description=( + "Optional full replacement body for PERSONA.md, excluding YAML frontmatter. " + "Use this when the persona definition should be rewritten completely." + ), + ) + append_instructions: Optional[list[str]] = Field( + None, + description=( + "Optional extra persona rules to append to the existing PERSONA body. " + "Use this for small adjustments such as '回答更短' or '复杂问题给两步解释'." + ), + ) + create_if_missing: bool = Field( + False, + description="Whether to create a new runtime persona if the target persona does not already exist.", + ) + + +class UpdatePersonaDefinitionTool(MoviePilotTool): + name: str = "update_persona_definition" + description: str = ( + "Create or update a runtime persona definition (人格定义) without manually editing PERSONA.md files. " + "Use this when the user explicitly asks to modify how a persona is defined, such as changing tone rules, " + "rewriting the persona body, adjusting aliases, or creating a new persona." + ) + args_schema: Type[BaseModel] = UpdatePersonaDefinitionInput + require_admin: bool = True + + def get_tool_message(self, **kwargs) -> str: + persona_id = kwargs.get("persona_id") or "未知人格" + action = "创建/更新人格定义" + return f"{action}: {persona_id}" + + async def run( + self, + persona_id: str, + label: Optional[str] = None, + description: Optional[str] = None, + aliases: Optional[list[str]] = None, + instructions: Optional[str] = None, + append_instructions: Optional[list[str]] = None, + create_if_missing: bool = False, + **kwargs, + ) -> str: + logger.info("执行工具: %s, 参数: persona_id=%s", self.name, persona_id) + if not any( + value is not None + for value in (label, description, aliases, instructions, append_instructions) + ): + return json.dumps( + { + "success": False, + "message": "未提供任何要更新的人格定义字段。", + }, + ensure_ascii=False, + ) + + try: + persona, created = agent_runtime_manager.update_persona_definition( + persona_id, + label=label, + description=description, + aliases=aliases, + instructions=instructions, + append_instructions=append_instructions, + create_if_missing=create_if_missing, + ) + runtime_config = agent_runtime_manager.load_runtime_config() + payload = { + "success": True, + "created": created, + "active_persona": runtime_config.active_persona, + "persona": persona.to_dict( + is_active=persona.persona_id == runtime_config.active_persona + ), + "message": ( + f"已创建人格 `{persona.persona_id}`" + if created + else f"已更新人格 `{persona.persona_id}` 的定义" + ), + } + return json.dumps(payload, ensure_ascii=False, indent=2) + except Exception as e: # noqa: BLE001 + logger.error("更新人格定义失败: %s", e, exc_info=True) + return json.dumps( + { + "success": False, + "message": f"更新人格定义时发生错误: {str(e)}", + }, + ensure_ascii=False, + ) diff --git a/tests/test_agent_interaction.py b/tests/test_agent_interaction.py index e5b1adc3..957f6d9d 100644 --- a/tests/test_agent_interaction.py +++ b/tests/test_agent_interaction.py @@ -82,7 +82,7 @@ class TestAgentInteraction(unittest.TestCase): self.assertTrue(tool._agent_context.get("user_reply_sent")) notification = async_post_message.await_args.args[0] self.assertEqual(notification.text, "请选择要执行的操作") - self.assertEqual(len(notification.buttons[0]), 2) + self.assertEqual(sum(len(row) for row in notification.buttons), 2) callback_data = notification.buttons[0][0]["callback_data"] _, _, request_id, option_index = callback_data.split(":") diff --git a/tests/test_agent_persona_tools.py b/tests/test_agent_persona_tools.py new file mode 100644 index 00000000..f620aef2 --- /dev/null +++ b/tests/test_agent_persona_tools.py @@ -0,0 +1,121 @@ +import asyncio +import json +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from app.agent.runtime import AgentRuntimeManager +from app.agent.tools.impl.query_personas import QueryPersonasTool +from app.agent.tools.impl.switch_persona import SwitchPersonaTool +from app.agent.tools.impl.update_persona_definition import UpdatePersonaDefinitionTool + + +class TestAgentPersonaTools(unittest.TestCase): + def setUp(self): + self._tempdir = tempfile.TemporaryDirectory() + self.addCleanup(self._tempdir.cleanup) + self.temp_root = Path(self._tempdir.name) + self.agent_root = self.temp_root / "agent" + defaults_root = ( + Path(__file__).resolve().parents[1] / "app" / "agent" / "defaults" + ) + self.runtime_manager = AgentRuntimeManager( + agent_root_dir=self.agent_root, + bundled_defaults_dir=defaults_root, + ) + self.runtime_manager.ensure_layout() + + def test_query_personas_returns_available_personas_and_active_state(self): + tool = QueryPersonasTool(session_id="session-1", user_id="10001") + + with patch( + "app.agent.tools.impl.query_personas.agent_runtime_manager", + self.runtime_manager, + ): + result = asyncio.run(tool.run()) + + payload = json.loads(result) + self.assertEqual(payload["active_persona"], "default") + self.assertGreaterEqual(payload["count"], 3) + self.assertTrue(any(persona["persona_id"] == "concise" for persona in payload["personas"])) + self.assertTrue(any(persona["is_active"] for persona in payload["personas"])) + + def test_switch_persona_updates_runtime_by_alias(self): + tool = SwitchPersonaTool(session_id="session-1", user_id="10001") + + with patch( + "app.agent.tools.impl.switch_persona.agent_runtime_manager", + self.runtime_manager, + ): + result = asyncio.run(tool.run(persona_id="讲解")) + + payload = json.loads(result) + self.assertTrue(payload["success"]) + self.assertEqual(payload["active_persona"], "guide") + self.assertEqual(self.runtime_manager.load_runtime_config().active_persona, "guide") + + def test_update_persona_definition_updates_existing_persona(self): + tool = UpdatePersonaDefinitionTool(session_id="session-1", user_id="10001") + + with patch( + "app.agent.tools.impl.update_persona_definition.agent_runtime_manager", + self.runtime_manager, + ): + result = asyncio.run( + tool.run( + persona_id="default", + description="更偏执行导向的默认人格。", + append_instructions=["Prefer action-first responses."], + ) + ) + + payload = json.loads(result) + self.assertTrue(payload["success"]) + self.assertFalse(payload["created"]) + runtime_config = self.runtime_manager.load_runtime_config() + default_persona = next( + persona + for persona in runtime_config.available_personas + if persona.persona_id == "default" + ) + self.assertEqual(default_persona.description, "更偏执行导向的默认人格。") + self.assertIn("Prefer action-first responses.", default_persona.text) + + def test_update_persona_definition_can_create_new_persona(self): + tool = UpdatePersonaDefinitionTool(session_id="session-1", user_id="10001") + + with patch( + "app.agent.tools.impl.update_persona_definition.agent_runtime_manager", + self.runtime_manager, + ): + result = asyncio.run( + tool.run( + persona_id="analysis", + label="分析型", + description="更适合解释复杂问题。", + aliases=["分析", "推理"], + instructions=( + "- Tone: analytical and structured.\n" + "- For complex tasks, explain the key tradeoff briefly." + ), + create_if_missing=True, + ) + ) + + payload = json.loads(result) + self.assertTrue(payload["success"]) + self.assertTrue(payload["created"]) + runtime_config = self.runtime_manager.load_runtime_config() + created_persona = next( + persona + for persona in runtime_config.available_personas + if persona.persona_id == "analysis" + ) + self.assertEqual(created_persona.label, "分析型") + self.assertIn("推理", created_persona.aliases) + self.assertIn("analytical and structured", created_persona.text) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_agent_prompt_style.py b/tests/test_agent_prompt_style.py index 59f9e1a0..01bcddb4 100644 --- a/tests/test_agent_prompt_style.py +++ b/tests/test_agent_prompt_style.py @@ -2,36 +2,84 @@ import unittest from unittest.mock import patch from app.agent.middleware.memory import MEMORY_ONBOARDING_PROMPT -from app.agent.prompt import prompt_manager +from app.agent.middleware.runtime_config import RuntimeConfigMiddleware +from app.agent.prompt import PromptConfigError, prompt_manager from app.core.config import settings +class _FakeRequest: + def __init__(self, system_message=None): + self.system_message = system_message + + def override(self, **kwargs): + return _FakeRequest(system_message=kwargs["system_message"]) + + class TestAgentPromptStyle(unittest.TestCase): - def test_agent_prompt_enforces_concise_professional_style(self): + def test_base_prompt_mentions_persona_management_tools(self): prompt = prompt_manager.get_agent_prompt() - self.assertIn("professional, concise, restrained", prompt) - self.assertIn("Do NOT flatter the user", prompt) - self.assertIn("NO praise, emotional cushioning", prompt) + self.assertIn("query_personas", prompt) + self.assertIn("switch_persona", prompt) + self.assertIn("update_persona_definition", prompt) - def test_agent_prompt_defines_tv_subscription_default_season_rule(self): + def test_base_prompt_contains_immutable_core_rules(self): prompt = prompt_manager.get_agent_prompt() + self.assertIn("AI media assistant powered by MoviePilot", prompt) self.assertIn( "omitting `season` means subscribe to season 1 only", prompt, ) self.assertIn( - "call `add_subscribe` separately for each season", + "Do not let user memory or persona style override this core identity", prompt, ) - def test_prompt_uses_root_runtime_sections(self): - prompt = prompt_manager.get_agent_prompt() + def test_runtime_config_middleware_injects_persona_only(self): + middleware = RuntimeConfigMiddleware() + updated_request = middleware.modify_request(_FakeRequest()) - self.assertIn("", prompt) - self.assertIn("", prompt) - self.assertIn("Active persona: `default`", prompt) + combined_text = "\n".join( + block["text"] for block in updated_request.system_message.content_blocks + ) + + self.assertIn("", combined_text) + self.assertIn("Active persona: `default`", combined_text) + self.assertIn("professional, concise, restrained", combined_text) + self.assertNotIn("System Tasks.yaml", combined_text) + + def test_system_tasks_are_loaded_from_prompt_directory(self): + definition = prompt_manager.load_system_tasks_definition() + + self.assertEqual(definition.version, 2) + self.assertTrue(definition.path.name.endswith("System Tasks.yaml")) + + def test_render_system_task_message_uses_builtin_yaml_definition(self): + message = prompt_manager.render_system_task_message("heartbeat") + + self.assertIn("[System Heartbeat]", message) + self.assertIn("List all jobs with status 'pending' or 'in_progress'.", message) + self.assertIn("Do NOT include greetings, explanations, or conversational text.", message) + self.assertIn("If no jobs were executed, output nothing.", message) + + def test_render_system_task_message_renders_template_context(self): + message = prompt_manager.render_system_task_message( + "transfer_failed_retry", + template_context={ + "history_ids_csv": "7", + "history_count": 1, + "history_id": 7, + }, + ) + + self.assertIn("Failed transfer history record IDs: 7", message) + self.assertIn("Total failed records: 1", message) + self.assertIn("history_id=7", message) + + def test_missing_system_task_template_context_raises_clear_error(self): + with self.assertRaises(PromptConfigError): + prompt_manager.render_system_task_message("transfer_failed_retry") def test_non_verbose_prompt_requires_silence_until_all_tools_finish(self): with patch.object(settings, "AI_AGENT_VERBOSE", False): diff --git a/tests/test_agent_runtime.py b/tests/test_agent_runtime.py index 2720da40..4c7b4735 100644 --- a/tests/test_agent_runtime.py +++ b/tests/test_agent_runtime.py @@ -4,7 +4,7 @@ import textwrap import unittest from pathlib import Path -from app.agent.runtime import AgentRuntimeConfigError, AgentRuntimeManager +from app.agent.runtime import AgentRuntimeManager class TestAgentRuntimeConfig(unittest.TestCase): @@ -13,14 +13,14 @@ class TestAgentRuntimeConfig(unittest.TestCase): self.addCleanup(self._tempdir.cleanup) self.temp_root = Path(self._tempdir.name) self.agent_root = self.temp_root / "agent" - self.bundled_root = ( - Path(__file__).resolve().parents[1] / "app" / "agent" / "runtime_defaults" + self.defaults_root = ( + Path(__file__).resolve().parents[1] / "app" / "agent" / "defaults" ) def _manager(self) -> AgentRuntimeManager: return AgentRuntimeManager( agent_root_dir=self.agent_root, - bundled_runtime_dir=self.bundled_root, + bundled_defaults_dir=self.defaults_root, ) def test_load_runtime_config_syncs_defaults_and_parses_sections(self): @@ -29,12 +29,22 @@ class TestAgentRuntimeConfig(unittest.TestCase): runtime_config = manager.load_runtime_config() self.assertEqual(runtime_config.active_persona, "default") - self.assertIn("professional, concise, restrained", runtime_config.profile_text) + self.assertIn("professional, concise, restrained", runtime_config.persona.text) + self.assertEqual(runtime_config.persona.persona_id, "default") self.assertIn( - "omitting `season` means subscribe to season 1 only", - runtime_config.workflow_text, + "concise", + [persona.persona_id for persona in runtime_config.available_personas], ) self.assertTrue((self.agent_root / "runtime" / "CURRENT_PERSONA.md").exists()) + self.assertTrue( + ( + self.agent_root + / "runtime" + / "personas" + / "default" + / "PERSONA.md" + ).exists() + ) def test_legacy_root_markdown_is_migrated_to_memory_directory(self): self.agent_root.mkdir(parents=True, exist_ok=True) @@ -45,12 +55,10 @@ class TestAgentRuntimeConfig(unittest.TestCase): textwrap.dedent( """\ --- + version: 3 active_persona: default - profile: personas/default/AGENT_PROFILE.md - workflow: personas/default/AGENT_WORKFLOW.md - hooks: personas/default/AGENT_HOOKS.md - system_tasks: system_tasks/SYSTEM_TASKS.md - user_preferences: USER_PREFERENCES.md + extra_context_files: [] + deprecated_phrases: [] --- """ ), @@ -65,40 +73,54 @@ class TestAgentRuntimeConfig(unittest.TestCase): self.assertFalse(legacy_persona.exists()) self.assertTrue((self.agent_root / "runtime" / "CURRENT_PERSONA.md").exists()) - def test_render_system_task_message_uses_unified_system_tasks_definition(self): - manager = self._manager() - runtime_config = manager.load_runtime_config() + def test_obsolete_runtime_files_are_deleted_instead_of_migrated(self): + self.agent_root.mkdir(parents=True, exist_ok=True) + obsolete_root = self.agent_root / "USER_PREFERENCES.md" + obsolete_root.write_text("# Obsolete\n", encoding="utf-8") - message = runtime_config.render_system_task_message("heartbeat") + obsolete_runtime = self.agent_root / "runtime" / "system_tasks" / "SYSTEM_TASKS.md" + obsolete_runtime.parent.mkdir(parents=True, exist_ok=True) + obsolete_runtime.write_text("# Obsolete Tasks\n", encoding="utf-8") - self.assertIn("[System Heartbeat]", message) - self.assertIn("List all jobs with status 'pending' or 'in_progress'.", message) - self.assertIn("Do NOT include greetings, explanations, or conversational text.", message) - self.assertIn("If no jobs were executed, output nothing.", message) - - def test_render_system_task_message_renders_template_context(self): - manager = self._manager() - runtime_config = manager.load_runtime_config() - - message = runtime_config.render_system_task_message( - "transfer_failed_retry", - template_context={ - "history_ids_csv": "7", - "history_count": 1, - "history_id": 7, - }, + obsolete_persona = ( + self.agent_root + / "runtime" + / "personas" + / "default" + / "AGENT_PROFILE.md" ) + obsolete_persona.parent.mkdir(parents=True, exist_ok=True) + obsolete_persona.write_text("# Obsolete Persona\n", encoding="utf-8") - self.assertIn("Failed transfer history record IDs: 7", message) - self.assertIn("Total failed records: 1", message) - self.assertIn("history_id=7", message) + manager = self._manager() + manager.ensure_layout() - def test_missing_template_context_raises_clear_error(self): + self.assertFalse(obsolete_root.exists()) + self.assertFalse(obsolete_runtime.exists()) + self.assertFalse(obsolete_persona.exists()) + self.assertFalse((self.agent_root / "memory" / "USER_PREFERENCES.md").exists()) + + def test_render_prompt_sections_uses_active_persona(self): manager = self._manager() runtime_config = manager.load_runtime_config() - with self.assertRaises(AgentRuntimeConfigError): - runtime_config.render_system_task_message("transfer_failed_retry") + sections = runtime_config.render_prompt_sections() + + self.assertIn("", sections) + self.assertIn("Active persona: `default`", sections) + self.assertIn("`guide`", sections) + + def test_set_active_persona_supports_id_and_alias(self): + manager = self._manager() + manager.load_runtime_config() + + guide_config = manager.set_active_persona("guide") + self.assertEqual(guide_config.active_persona, "guide") + self.assertEqual(guide_config.persona.label, "说明型") + + concise_config = manager.set_active_persona("简洁") + self.assertEqual(concise_config.active_persona, "concise") + self.assertIn("active_persona: concise", concise_config.current_persona_path.read_text(encoding="utf-8")) def test_invalid_user_runtime_config_falls_back_to_bundled_defaults(self): manager = self._manager() @@ -108,10 +130,10 @@ class TestAgentRuntimeConfig(unittest.TestCase): textwrap.dedent( """\ --- + version: 3 active_persona: broken - profile: personas/default/AGENT_PROFILE.md - hooks: personas/default/AGENT_HOOKS.md - system_tasks: system_tasks/SYSTEM_TASKS.md + extra_context_files: [] + deprecated_phrases: [] --- """ ), @@ -128,19 +150,14 @@ class TestAgentRuntimeConfig(unittest.TestCase): def test_deprecated_phrase_warning_is_reported(self): self.agent_root.mkdir(parents=True, exist_ok=True) runtime_root = self.agent_root / "runtime" - shutil.copytree(self.bundled_root, runtime_root) + shutil.copytree(self.defaults_root, runtime_root) current_persona = runtime_root / "CURRENT_PERSONA.md" current_persona.write_text( textwrap.dedent( """\ --- - version: 1 + version: 3 active_persona: default - profile: personas/default/AGENT_PROFILE.md - workflow: personas/default/AGENT_WORKFLOW.md - hooks: personas/default/AGENT_HOOKS.md - user_preferences: USER_PREFERENCES.md - system_tasks: system_tasks/SYSTEM_TASKS.md extra_context_files: [] deprecated_phrases: - professional, concise, restrained @@ -155,38 +172,12 @@ class TestAgentRuntimeConfig(unittest.TestCase): runtime_config = manager.load_runtime_config() self.assertTrue( - any("professional, concise, restrained" in warning for warning in runtime_config.warnings) + any( + "professional, concise, restrained" in warning + for warning in runtime_config.warnings + ) ) - def test_outdated_system_tasks_definition_falls_back_to_bundled_defaults(self): - self.agent_root.mkdir(parents=True, exist_ok=True) - runtime_root = self.agent_root / "runtime" - shutil.copytree(self.bundled_root, runtime_root) - system_tasks = runtime_root / "system_tasks" / "SYSTEM_TASKS.md" - system_tasks.write_text( - textwrap.dedent( - """\ - --- - version: 1 - shared_rules: - - legacy system tasks - task_types: - heartbeat: - header: "[Legacy Heartbeat]" - objective: "legacy" - --- - """ - ), - encoding="utf-8", - ) - - manager = self._manager() - manager.invalidate_cache() - runtime_config = manager.load_runtime_config() - - self.assertTrue(runtime_config.used_fallback) - self.assertEqual(runtime_config.system_tasks.version, 2) - if __name__ == "__main__": unittest.main() diff --git a/tests/test_agent_summarization_streaming.py b/tests/test_agent_summarization_streaming.py index 5803524e..490cfaa6 100644 --- a/tests/test_agent_summarization_streaming.py +++ b/tests/test_agent_summarization_streaming.py @@ -4,6 +4,7 @@ from unittest.mock import patch from langchain.agents.middleware import SummarizationMiddleware import app.agent as agent_module +from app.agent.middleware.runtime_config import RuntimeConfigMiddleware class _FakeLLM: @@ -115,6 +116,35 @@ class TestAgentSummarizationStreaming(unittest.TestCase): self.assertIs(captured["model"], main_llm) self.assertIs(summary_middleware.model, main_llm) + def test_agent_uses_runtime_config_middleware_instead_of_hooks(self): + agent = agent_module.MoviePilotAgent(session_id="session-1", user_id="10001") + main_llm = _FakeLLM("main") + captured: dict = {} + + def _fake_create_agent(**kwargs): + captured.update(kwargs) + return object() + + with ( + patch.object(agent, "_initialize_llm", return_value=main_llm), + patch.object(agent, "_initialize_tools", return_value=[]), + patch.object( + agent_module.prompt_manager, "get_agent_prompt", return_value="prompt" + ), + patch.object(agent_module, "create_agent", side_effect=_fake_create_agent), + ): + agent._create_agent(streaming=False) + + self.assertTrue( + any( + isinstance(middleware, RuntimeConfigMiddleware) + for middleware in captured["middleware"] + ) + ) + self.assertFalse( + any(type(middleware).__name__ == "AgentHooksMiddleware" for middleware in captured["middleware"]) + ) + if __name__ == "__main__": unittest.main()