feat: refine job handling by filtering active jobs and updating date context in prompts

2026-05-12 05:29:40 +08:00 · 2026-05-11 13:15:32 +08:00
parent 1b2433f7c2
commit b7fc5b0203
8 changed files with 302 additions and 36 deletions
--- a/app/agent/init.py
+++ b/app/agent/init.py
@@ -22,7 +22,11 @@ from app.agent.callback import StreamingHandler
 from app.agent.llm import LLMHelper
 from app.agent.memory import memory_manager
 from app.agent.middleware.activity_log import ActivityLogMiddleware
-from app.agent.middleware.jobs import JobsMiddleware
+from app.agent.middleware.jobs import (
+    JobsMiddleware,
+    filter_active_jobs,
+    load_jobs_metadata,
+)
 from app.agent.middleware.memory import MemoryMiddleware
 from app.agent.middleware.patch_tool_calls import PatchToolCallsMiddleware
 from app.agent.middleware.runtime_config import RuntimeConfigMiddleware
@@ -160,6 +164,9 @@ class ReplyMode(str, Enum):
    CAPTURE_ONLY = "capture_only"


+HEARTBEAT_SESSION_PREFIX = "__agent_heartbeat_"
+
+
 class MoviePilotAgent:
    """
    MoviePilot AI智能体（基于 LangChain v1 + LangGraph）
@@ -288,6 +295,16 @@ class MoviePilotAgent:
        """
        return self.reply_mode == ReplyMode.DISPATCH

+    @property
+    def is_heartbeat_session(self) -> bool:
+        """
+        是否为后台心跳会话。
+
+        心跳场景只负责检查并执行待处理 job，不需要携带近期活动日志，
+        否则会让这类高频后台调用持续带入无关动态上下文，影响缓存命中率。
+        """
+        return self.session_id.startswith(HEARTBEAT_SESSION_PREFIX)
+
    def _should_stream(self) -> bool:
        """
        判断是否应启用流式输出：
@@ -430,10 +447,6 @@ class MoviePilotAgent:
                RuntimeConfigMiddleware(),
                # 记忆管理
                MemoryMiddleware(memory_dir=str(agent_runtime_manager.memory_dir)),
-                # 活动日志
-                ActivityLogMiddleware(
-                    activity_dir=str(agent_runtime_manager.activity_dir),
-                ),
                # 上下文压缩
                SummarizationMiddleware(
                    model=non_streaming_model, trigger=("fraction", 0.85)
@@ -444,6 +457,14 @@ class MoviePilotAgent:
                UsageMiddleware(on_usage=self._record_usage),
            ]

+            if not self.is_heartbeat_session:
+                middlewares.insert(
+                    4,
+                    ActivityLogMiddleware(
+                        activity_dir=str(agent_runtime_manager.activity_dir),
+                    ),
+                )
+
            # 工具选择
            if max_tools > 0:
                middlewares.append(
@@ -1085,8 +1106,17 @@ class AgentManager:
        由定时调度器周期性调用，每次使用独立的会话避免上下文干扰。
        """
        try:
+            active_jobs = filter_active_jobs(
+                await load_jobs_metadata([str(agent_runtime_manager.jobs_dir)])
+            )
+            # 先在本地判断是否存在活跃任务。没有任务时直接短路，避免一次完整
+            # 的后台 Agent/LLM 空调用。
+            if not active_jobs:
+                logger.info("智能体心跳唤醒：没有活跃任务，跳过模型调用")
+                return
+
            # 每次使用唯一的 session_id，避免共享上下文
-            session_id = f"__agent_heartbeat_{uuid.uuid4().hex[:12]}__"
+            session_id = f"{HEARTBEAT_SESSION_PREFIX}{uuid.uuid4().hex[:12]}__"
            user_id = SYSTEM_INTERNAL_USER_ID

            logger.info("智能体心跳唤醒：开始检查待处理任务...")
--- a/app/agent/middleware/jobs.py
+++ b/app/agent/middleware/jobs.py
@@ -21,6 +21,7 @@ from app.log import logger

 # JOB.md 文件最大限制为 1MB
 MAX_JOB_FILE_SIZE = 1 * 1024 * 1024
+ACTIVE_JOB_STATUSES = ("pending", "in_progress")


 class JobMetadata(TypedDict):
@@ -143,6 +144,9 @@ async def _alist_jobs(source_path: AsyncPath) -> list[JobMetadata]:
    if not job_dirs:
        return []

+    # 显式按目录名排序，避免文件系统返回顺序不稳定时破坏提示词缓存命中。
+    job_dirs.sort(key=lambda p: p.name.casefold())
+
    # 解析 JOB.md
    for job_path in job_dirs:
        job_md_path = job_path / "JOB.md"
@@ -161,6 +165,31 @@ async def _alist_jobs(source_path: AsyncPath) -> list[JobMetadata]:
    return jobs


+def filter_active_jobs(jobs_metadata: list[JobMetadata]) -> list[JobMetadata]:
+    """筛选需要参与心跳检查的活跃任务。
+
+    这里严格以任务状态为准，只保留 `pending` / `in_progress`。
+    `recurring` 任务执行完成后按约定应回写为 `pending`，因此无需再额外放宽
+    到 `completed`，避免已结束任务被重复注入后台心跳。
+    """
+    return [
+        job for job in jobs_metadata if job.get("status") in ACTIVE_JOB_STATUSES
+    ]
+
+
+async def load_jobs_metadata(source_paths: list[str]) -> list[JobMetadata]:
+    """按顺序加载多个 jobs 目录下的任务元数据。"""
+    all_jobs: list[JobMetadata] = []
+    for source_path_str in source_paths:
+        source_path = AsyncPath(source_path_str)
+        if not await source_path.exists():
+            await source_path.mkdir(parents=True, exist_ok=True)
+            continue
+        source_jobs = await _alist_jobs(source_path)
+        all_jobs.extend(source_jobs)
+    return all_jobs
+
+
 JOBS_SYSTEM_PROMPT = """
 <jobs_system>
 You have a **scheduled jobs** system that allows you to track and execute long-running or recurring tasks.
@@ -289,13 +318,8 @@ class JobsMiddleware(AgentMiddleware[JobsState, ContextT, ResponseT]):  # noqa
        """将任务文档注入模型请求的系统消息中。"""
        jobs_metadata = request.state.get("jobs_metadata", [])  # noqa

-        # 过滤：只展示活跃任务（pending / in_progress / recurring）
-        active_jobs = [
-            j
-            for j in jobs_metadata
-            if j["status"] in ("pending", "in_progress")
-            or (j["schedule"] == "recurring" and j["status"] not in ("cancelled",))
-        ]
+        # 仅注入真正活跃的任务，避免把已完成任务继续塞进心跳上下文。
+        active_jobs = filter_active_jobs(jobs_metadata)

        jobs_list = self._format_jobs_list(active_jobs)
        jobs_location = self.sources[0] if self.sources else ""
@@ -322,18 +346,9 @@ class JobsMiddleware(AgentMiddleware[JobsState, ContextT, ResponseT]):  # noqa
        if "jobs_metadata" in state:
            return None

-        all_jobs: list[JobMetadata] = []
-
-        # 遍历源加载任务
-        for source_path_str in self.sources:
-            source_path = AsyncPath(source_path_str)
-            if not await source_path.exists():
-                await source_path.mkdir(parents=True, exist_ok=True)
-                continue
-            source_jobs = await _alist_jobs(source_path)
-            all_jobs.extend(source_jobs)
-
-        return JobsStateUpdate(jobs_metadata=all_jobs)
+        return JobsStateUpdate(
+            jobs_metadata=await load_jobs_metadata(self.sources)
+        )

    async def awrap_model_call(
        self,
@@ -347,4 +362,10 @@ class JobsMiddleware(AgentMiddleware[JobsState, ContextT, ResponseT]):  # noqa
        return await handler(modified_request)


-__all__ = ["JobMetadata", "JobsMiddleware"]
+__all__ = [
+    "ACTIVE_JOB_STATUSES",
+    "JobMetadata",
+    "JobsMiddleware",
+    "filter_active_jobs",
+    "load_jobs_metadata",
+]
--- a/app/agent/middleware/skills.py
+++ b/app/agent/middleware/skills.py
@@ -227,6 +227,9 @@ async def _alist_skills(source_path: AsyncPath) -> list[SkillMetadata]:
    if not skill_dirs:
        return []

+    # 显式按目录名排序，避免文件系统返回顺序不稳定时破坏提示词缓存命中。
+    skill_dirs.sort(key=lambda p: p.name.casefold())
+
    # 解析已下载的 SKILL.md
    for skill_path in skill_dirs:
        skill_md_path = skill_path / "SKILL.md"
--- a/app/agent/prompt/init.py
+++ b/app/agent/prompt/init.py
@@ -286,8 +286,10 @@ class PromptManager:
                f"{settings.DB_POSTGRESQL_TARGET}/{settings.DB_POSTGRESQL_DATABASE})"
            )

+        # 保留日期用于提供“今天是哪天”的稳定上下文，但不再注入秒级时间，
+        # 避免每次请求都生成不同的 system prompt，影响 provider 侧 cache 命中率。
        info_lines = [
-            f"- 当前时间: {strftime('%Y-%m-%d %H:%M:%S')}",
+            f"- 当前日期: {strftime('%Y-%m-%d')}",
            f"- 运行环境: {SystemUtils.platform} {'docker' if SystemUtils.is_docker() else ''}",
            f"- 主机名: {hostname}",
            f"- IP地址: {ip_address}",
@@ -426,11 +428,11 @@ class PromptManager:
            return text

        context = cls._normalize_template_context(template_context)
-        missing_fields = sorted(field for field in required_fields if field not in context)
+        missing_fields = sorted(f for f in required_fields if f not in context)
        if missing_fields:
            raise PromptConfigError(
                f"系统任务定义 `{task_type}` 的 `{field_name}` 缺少变量: "
-                + ", ".join(f"`{field}`" for field in missing_fields)
+                + ", ".join(f"`{f}`" for f in missing_fields)
            )

        # 这里统一做字符串替换，让 YAML 成为后台任务文案的唯一行为来源。
--- a/tests/test_agent_background_output.py
+++ b/tests/test_agent_background_output.py
@@ -4,8 +4,14 @@ from unittest.mock import AsyncMock, patch

 from langchain_core.messages import AIMessage

-from app.agent import MoviePilotAgent, AgentManager, ReplyMode
+from app.agent import (
+    HEARTBEAT_SESSION_PREFIX,
+    MoviePilotAgent,
+    AgentManager,
+    ReplyMode,
+)
 from app.agent.memory import memory_manager
+from app.core.config import settings
 from app.utils.identity import SYSTEM_INTERNAL_USER_ID


@@ -38,8 +44,8 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
            stop_streaming=AsyncMock(return_value=(False, ""))
        )
        agent._should_stream = lambda: False
-        agent._create_agent = lambda streaming=False: _FakeAgent(
-            [AIMessage(content="后台结果")]
+        agent._create_agent = AsyncMock(
+            return_value=_FakeAgent([AIMessage(content="后台结果")])
        )
        agent.send_agent_message = AsyncMock()
        agent._save_agent_message_to_db = AsyncMock()
@@ -66,8 +72,8 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
            stop_streaming=AsyncMock(return_value=(False, ""))
        )
        agent._should_stream = lambda: False
-        agent._create_agent = lambda streaming=False: _FakeAgent(
-            [AIMessage(content="后台结果")]
+        agent._create_agent = AsyncMock(
+            return_value=_FakeAgent([AIMessage(content="后台结果")])
        )
        agent.send_agent_message = AsyncMock()
        agent._save_agent_message_to_db = AsyncMock()
@@ -94,8 +100,8 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
            stop_streaming=AsyncMock(return_value=(False, ""))
        )
        agent._should_stream = lambda: False
-        agent._create_agent = lambda streaming=False: _FakeAgent(
-            [AIMessage(content="后台结果")]
+        agent._create_agent = AsyncMock(
+            return_value=_FakeAgent([AIMessage(content="后台结果")])
        )
        agent.send_agent_message = AsyncMock()
        agent._save_agent_message_to_db = AsyncMock()
@@ -114,6 +120,15 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
        manager = AgentManager()

        with (
+            patch("app.agent.load_jobs_metadata", new=AsyncMock(return_value=[{
+                "id": "job-1",
+                "name": "测试任务",
+                "description": "desc",
+                "path": "/tmp/job-1/JOB.md",
+                "schedule": "once",
+                "status": "pending",
+                "last_run": None,
+            }])),
            patch.object(manager, "_build_heartbeat_prompt", return_value="HEARTBEAT"),
            patch.object(manager, "process_message", new=AsyncMock()) as process_message,
        ):
@@ -125,6 +140,80 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
            process_message.await_args.kwargs["reply_mode"],
        )

+    async def test_heartbeat_check_jobs_skips_when_no_active_jobs(self):
+        manager = AgentManager()
+
+        with (
+            patch("app.agent.load_jobs_metadata", new=AsyncMock(return_value=[])),
+            patch.object(manager, "process_message", new=AsyncMock()) as process_message,
+        ):
+            await manager.heartbeat_check_jobs()
+
+        process_message.assert_not_awaited()
+
+    async def test_create_agent_excludes_activity_log_for_heartbeat_session(self):
+        agent = MoviePilotAgent(
+            session_id=f"{HEARTBEAT_SESSION_PREFIX}test__",
+            user_id="system",
+        )
+        agent._initialize_tools = lambda: []
+
+        with (
+            patch.object(settings, "LLM_MAX_TOOLS", 0),
+            patch.object(agent, "_initialize_llm", new=AsyncMock(return_value=object())),
+            patch("app.agent.prompt_manager.get_agent_prompt", return_value="PROMPT"),
+            patch(
+                "app.agent.MoviePilotToolFactory.get_tool_selector_always_include_names",
+                return_value=[],
+            ),
+            patch("app.agent.SkillsMiddleware", side_effect=lambda *args, **kwargs: "skills"),
+            patch("app.agent.JobsMiddleware", side_effect=lambda *args, **kwargs: "jobs"),
+            patch("app.agent.RuntimeConfigMiddleware", side_effect=lambda *args, **kwargs: "runtime"),
+            patch("app.agent.MemoryMiddleware", side_effect=lambda *args, **kwargs: "memory"),
+            patch("app.agent.ActivityLogMiddleware", side_effect=lambda *args, **kwargs: "activity"),
+            patch("app.agent.SummarizationMiddleware", side_effect=lambda *args, **kwargs: "summary"),
+            patch("app.agent.PatchToolCallsMiddleware", side_effect=lambda *args, **kwargs: "patch"),
+            patch("app.agent.UsageMiddleware", side_effect=lambda *args, **kwargs: "usage"),
+            patch("app.agent.InMemorySaver", return_value="checkpointer"),
+            patch("app.agent.create_agent", side_effect=lambda **kwargs: kwargs),
+        ):
+            created = await agent._create_agent(streaming=False)
+
+        self.assertEqual(
+            ["skills", "jobs", "runtime", "memory", "summary", "patch", "usage"],
+            created["middleware"],
+        )
+
+    async def test_create_agent_keeps_activity_log_for_normal_session(self):
+        agent = MoviePilotAgent(session_id="normal-session", user_id="system")
+        agent._initialize_tools = lambda: []
+
+        with (
+            patch.object(settings, "LLM_MAX_TOOLS", 0),
+            patch.object(agent, "_initialize_llm", new=AsyncMock(return_value=object())),
+            patch("app.agent.prompt_manager.get_agent_prompt", return_value="PROMPT"),
+            patch(
+                "app.agent.MoviePilotToolFactory.get_tool_selector_always_include_names",
+                return_value=[],
+            ),
+            patch("app.agent.SkillsMiddleware", side_effect=lambda *args, **kwargs: "skills"),
+            patch("app.agent.JobsMiddleware", side_effect=lambda *args, **kwargs: "jobs"),
+            patch("app.agent.RuntimeConfigMiddleware", side_effect=lambda *args, **kwargs: "runtime"),
+            patch("app.agent.MemoryMiddleware", side_effect=lambda *args, **kwargs: "memory"),
+            patch("app.agent.ActivityLogMiddleware", side_effect=lambda *args, **kwargs: "activity"),
+            patch("app.agent.SummarizationMiddleware", side_effect=lambda *args, **kwargs: "summary"),
+            patch("app.agent.PatchToolCallsMiddleware", side_effect=lambda *args, **kwargs: "patch"),
+            patch("app.agent.UsageMiddleware", side_effect=lambda *args, **kwargs: "usage"),
+            patch("app.agent.InMemorySaver", return_value="checkpointer"),
+            patch("app.agent.create_agent", side_effect=lambda **kwargs: kwargs),
+        ):
+            created = await agent._create_agent(streaming=False)
+
+        self.assertEqual(
+            ["skills", "jobs", "runtime", "memory", "activity", "summary", "patch", "usage"],
+            created["middleware"],
+        )
+
    async def test_run_background_prompt_forces_disable_message_tools_when_capture_only(self):
        captured = {}

--- a/tests/test_agent_jobs_middleware.py
+++ b/tests/test_agent_jobs_middleware.py
@@ -0,0 +1,82 @@
+import tempfile
+import unittest
+from pathlib import Path
+
+from anyio import Path as AsyncPath
+
+from app.agent.middleware.jobs import _alist_jobs, filter_active_jobs
+
+
+class JobsMiddlewareTest(unittest.TestCase):
+    def test_filter_active_jobs_only_keeps_pending_and_in_progress(self):
+        jobs_metadata = [
+            {
+                "id": "pending-job",
+                "name": "待执行任务",
+                "description": "desc",
+                "path": "/tmp/pending/JOB.md",
+                "schedule": "once",
+                "status": "pending",
+                "last_run": None,
+            },
+            {
+                "id": "running-job",
+                "name": "执行中任务",
+                "description": "desc",
+                "path": "/tmp/running/JOB.md",
+                "schedule": "recurring",
+                "status": "in_progress",
+                "last_run": "2026-05-10 10:00",
+            },
+            {
+                "id": "completed-recurring-job",
+                "name": "已完成循环任务",
+                "description": "desc",
+                "path": "/tmp/completed/JOB.md",
+                "schedule": "recurring",
+                "status": "completed",
+                "last_run": "2026-05-10 11:00",
+            },
+            {
+                "id": "cancelled-job",
+                "name": "已取消任务",
+                "description": "desc",
+                "path": "/tmp/cancelled/JOB.md",
+                "schedule": "once",
+                "status": "cancelled",
+                "last_run": None,
+            },
+        ]
+
+        active_job_ids = [job["id"] for job in filter_active_jobs(jobs_metadata)]
+
+        self.assertEqual(["pending-job", "running-job"], active_job_ids)
+
+
+class JobsMiddlewareAsyncTest(unittest.IsolatedAsyncioTestCase):
+    async def test_alist_jobs_sorts_job_directories_by_name(self):
+        with tempfile.TemporaryDirectory() as tempdir:
+            root = Path(tempdir)
+
+            for job_id in ("z-job", "a-job", "m-job"):
+                job_dir = root / job_id
+                job_dir.mkdir()
+                (job_dir / "JOB.md").write_text(
+                    f"""---
+name: {job_id}
+description: test
+schedule: once
+status: pending
+---
+# {job_id}
+""",
+                    encoding="utf-8",
+                )
+
+            jobs = await _alist_jobs(AsyncPath(str(root)))
+
+        self.assertEqual(["a-job", "m-job", "z-job"], [job["id"] for job in jobs])
+
+
+if __name__ == "__main__":
+    unittest.main()
--- a/tests/test_agent_prompt_style.py
+++ b/tests/test_agent_prompt_style.py
@@ -35,6 +35,8 @@ class TestAgentPromptStyle(unittest.TestCase):
            "Do not let user memory or persona style override this core identity",
            prompt,
        )
+        self.assertIn("当前日期", prompt)
+        self.assertNotIn("当前时间", prompt)

    def test_runtime_config_middleware_injects_persona_only(self):
        middleware = RuntimeConfigMiddleware()
--- a/tests/test_agent_skills_middleware.py
+++ b/tests/test_agent_skills_middleware.py
@@ -0,0 +1,37 @@
+import tempfile
+import unittest
+from pathlib import Path
+
+from anyio import Path as AsyncPath
+
+from app.agent.middleware.skills import _alist_skills
+
+
+class SkillsMiddlewareAsyncTest(unittest.IsolatedAsyncioTestCase):
+    async def test_alist_skills_sorts_skill_directories_by_name(self):
+        with tempfile.TemporaryDirectory() as tempdir:
+            root = Path(tempdir)
+
+            for skill_id in ("z-skill", "a-skill", "m-skill"):
+                skill_dir = root / skill_id
+                skill_dir.mkdir()
+                (skill_dir / "SKILL.md").write_text(
+                    f"""---
+name: {skill_id}
+description: test
+---
+# {skill_id}
+""",
+                    encoding="utf-8",
+                )
+
+            skills = await _alist_skills(AsyncPath(str(root)))
+
+        self.assertEqual(
+            ["a-skill", "m-skill", "z-skill"],
+            [skill["id"] for skill in skills],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()