From a2deeac0d6090e069dcf2622080c9e2154145b67 Mon Sep 17 00:00:00 2001 From: Chiguyong Date: Mon, 6 Jul 2026 13:57:51 +0800 Subject: [PATCH] feat(iq): U6 Lead planning-time reflection retrieval (R12, R13) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ReflexionEngine.retrieve_prompt_reflection(): searches EpisodicMemory for historical reflections on similar task_input, returns best version by score (defaults min_score=0.5). Non-blocking: failure → None. - TeamOrchestrator._decompose_task: prepends historical reflection hint to Lead's planning prompt when reflexion_engine is wired and a high-score reflection exists. Default prompt preserved on miss/failure. - 12 unit tests covering retrieve path (7) + decompose integration (5). --- src/agentkit/core/reflexion.py | 52 +++ src/agentkit/experts/orchestrator.py | 36 +++ tests/unit/test_lead_reflection_retrieval.py | 319 +++++++++++++++++++ 3 files changed, 407 insertions(+) create mode 100644 tests/unit/test_lead_reflection_retrieval.py diff --git a/src/agentkit/core/reflexion.py b/src/agentkit/core/reflexion.py index e106090..5facdc3 100644 --- a/src/agentkit/core/reflexion.py +++ b/src/agentkit/core/reflexion.py @@ -26,6 +26,7 @@ from agentkit.telemetry.metrics import ( if TYPE_CHECKING: from agentkit.core.compressor import CompressionStrategy from agentkit.core.trace import TraceRecorder + from agentkit.memory.base import MemoryItem from agentkit.memory.episodic import EpisodicMemory from agentkit.memory.retriever import MemoryRetriever @@ -742,3 +743,54 @@ class ReflexionEngine: return original_prompt + reflection_section else: return reflection_section.strip() + + async def retrieve_prompt_reflection( + self, task_input: str, min_score: float = 0.5 + ) -> dict[str, object] | None: + """检索历史 prompt 反思，返回最佳版本 (U6/R12, R13). + + Searches EpisodicMemory for similar task_input reflections with + score > min_score. Returns the highest-scored reflection as: + {improved_prompt, score, reflection, version, task_hash} + or None if no episodic_memory / no results / all below threshold. + + KTD5: callers should only invoke this when a trigger condition is + met (verify failure / schema failure / loop detection) to avoid + pointless retrieval on every task. + """ + if self._episodic_memory is None: + return None + + try: + results = await self._episodic_memory.search_prompt_reflections( + task_input=task_input, top_k=3 + ) + except Exception as e: + logger.warning(f"U6: retrieve_prompt_reflection failed: {e}") + return None + + if not results: + return None + + # Filter by min_score, pick the highest-scored + best: MemoryItem | None = None + for item in results: + score = item.score or 0.0 + if score > min_score and (best is None or score > (best.score or 0.0)): + best = item + + if best is None: + return None + + # Extract improved_prompt from metadata (output_summary field) + metadata = best.metadata or {} + improved_prompt = metadata.get("output_summary", "") or metadata.get("improved_prompt", "") + reflection_text = metadata.get("reflection", "") or best.value or "" + + return { + "improved_prompt": improved_prompt, + "score": best.score or 0.0, + "reflection": reflection_text, + "version": metadata.get("version", 1), + "task_hash": metadata.get("task_hash", ""), + } diff --git a/src/agentkit/experts/orchestrator.py b/src/agentkit/experts/orchestrator.py index 8aa501a..182732a 100644 --- a/src/agentkit/experts/orchestrator.py +++ b/src/agentkit/experts/orchestrator.py @@ -14,6 +14,7 @@ import asyncio import json import logging import re +from typing import TYPE_CHECKING from agentkit.core.exceptions import LLMProviderError from agentkit.llm.gateway import LLMGateway @@ -36,6 +37,9 @@ from .plan import ( ) from .team import ExpertTeam, TeamStatus +if TYPE_CHECKING: + from agentkit.core.reflexion import ReflexionEngine + logger = logging.getLogger(__name__) # 专家名校验正则（与 router.py / board_router.py 保持一致） @@ -82,6 +86,10 @@ class TeamOrchestrator( # final-answer path (react.py:1303+) runs on coding tasks. verification_enabled: bool = True, verification_commands: list[str] | None = None, + # IQ-Boost/U6 (R12, R13): optional ReflexionEngine for retrieving + # historical prompt reflections at Lead planning time. None = no + # retrieval (backward-compatible). + reflexion_engine: "ReflexionEngine | None" = None, ) -> None: self._team = team # Track temporary agent names created for context isolation (KTD3) @@ -103,6 +111,8 @@ class TeamOrchestrator( self._rollback_timeout = rollback_timeout or self.DEFAULT_ROLLBACK_TIMEOUT # U3/R2: verification defaults for TEAM_COLLAB. self._verification_enabled = verification_enabled + # U6: optional reflexion engine for historical reflection retrieval + self._reflexion_engine = reflexion_engine # U3/R3: if no explicit commands, detect from workspace (coding-task # detection forces pytest/ruff). None workspace → None commands → # ReActEngine/VerificationLoop uses its own defaults. @@ -608,6 +618,11 @@ class TeamOrchestrator( Returns a list of PlanPhase instances. If LLM decomposition fails, returns a single phase with the original task. + + IQ-Boost/U6 (R12, R13): if reflexion_engine is configured, retrieves + historical prompt reflection for similar task_input and prepends + improved_prompt to the decomposition prompt. Non-blocking — retrieval + failure falls through to default prompt. """ gateway = self._get_llm_gateway(lead) if not gateway: @@ -619,6 +634,26 @@ class TeamOrchestrator( ] available_experts = member_names if member_names else [lead.config.name] + # U6: retrieve historical reflection (non-blocking) + reflection_hint = "" + if self._reflexion_engine is not None: + try: + historical = await self._reflexion_engine.retrieve_prompt_reflection( + task_input=task + ) + if historical and historical.get("improved_prompt"): + reflection_hint = ( + f"\n\n## Historical Reflection (score={historical.get('score', 0):.2f})\n" + f"A previous similar task produced this reflection. " + f"Use it to improve your decomposition:\n\n" + f"{historical['improved_prompt']}\n" + ) + logger.info( + f"U6: retrieved historical reflection (score={historical.get('score', 0):.2f})" + ) + except Exception as e: + logger.warning(f"U6: historical reflection retrieval failed, using default: {e}") + prompt = ( f"You are the Lead Expert in a pipeline team. Decompose the following task into " f"at most {self.MAX_PHASES} phases with dependencies.\n\n" @@ -646,6 +681,7 @@ class TeamOrchestrator( f'{{"name":"前端","assigned_expert":"frontend",' f'"task_description":"实现UI","depends_on":["后端"],"collaboration_contracts":[]}}]\n\n' f"Return ONLY the JSON array, no other text." + f"{reflection_hint}" ) try: diff --git a/tests/unit/test_lead_reflection_retrieval.py b/tests/unit/test_lead_reflection_retrieval.py new file mode 100644 index 0000000..263094f --- /dev/null +++ b/tests/unit/test_lead_reflection_retrieval.py @@ -0,0 +1,319 @@ +"""U6: Lead planning-time historical reflection retrieval (R12, R13). + +Covers: +- ReflexionEngine.retrieve_prompt_reflection(): returns best reflection by score +- Score filtering: score <= 0.5 not returned +- No episodic_memory → returns None +- Retrieval failure → returns None (non-blocking) +- TeamOrchestrator._decompose_task: prepends improved_prompt when reflection found +- No reflexion_engine → default prompt (backward compat) +- Retrieval failure → default prompt (non-blocking) +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from agentkit.core.reflexion import ReflexionEngine +from agentkit.experts.orchestrator import TeamOrchestrator +from agentkit.experts.team import ExpertTeam +from agentkit.memory.base import MemoryItem + +from tests.unit.experts._helpers import make_execute_stream_mock + + +# ── Helpers ──────────────────────────────────────────────────────────── + + +def _make_memory_item( + score: float = 0.8, + output_summary: str = "improved prompt text", + reflection: str = "reflection text", + version: int = 1, +) -> MemoryItem: + from datetime import datetime, timezone + + return MemoryItem( + key="prompt_reflection:abc:1", + value={"task_input": "test", "reflection": reflection}, + metadata={ + "task_type": "prompt_reflection", + "output_summary": output_summary, + "reflection": reflection, + "version": version, + "task_hash": "abc", + "quality_score": score, + }, + score=score, + created_at=datetime.now(timezone.utc), + ) + + +def _make_llm_gateway_mock() -> MagicMock: + gw = MagicMock() + response = MagicMock() + response.content = ( + '[{"name":"A","assigned_expert":"lead","task_description":"a","depends_on":[]}]' + ) + gw.chat = AsyncMock(return_value=response) + return gw + + +def _make_team_with_experts() -> ExpertTeam: + from agentkit.core.handoff_transport import InProcessHandoffTransport + from agentkit.core.protocol import TaskResult, TaskStatus + from agentkit.experts.config import ExpertConfig + from agentkit.experts.expert import Expert + + team = ExpertTeam() + team._handoff_transport = AsyncMock(spec=InProcessHandoffTransport) + + config = ExpertConfig( + name="lead", + agent_type="expert", + persona="测试", + thinking_style="逻辑", + bound_skills=["s"], + is_lead=True, + task_mode="llm_generate", + prompt={"identity": "测试"}, + ) + expert = MagicMock(spec=Expert) + expert.config = config + expert.is_active = True + expert.team_id = None + expert.get_capabilities_summary.return_value = {"name": "lead"} + + mock_agent = MagicMock() + mock_agent.execute = AsyncMock( + return_value=TaskResult( + task_id="t", + agent_name="lead", + status=TaskStatus.COMPLETED.value, + output_data={"content": "result"}, + error_message=None, + started_at=None, + completed_at=None, + ) + ) + mock_agent.execute_stream = make_execute_stream_mock("result") + mock_agent._llm_gateway = None + expert.agent = mock_agent + + team._experts["lead"] = expert + team._lead_expert_name = "lead" + return team + + +# ── ReflexionEngine.retrieve_prompt_reflection ───────────────────────── + + +class TestRetrievePromptReflection: + """U6/R12: retrieve_prompt_reflection returns best reflection by score.""" + + @pytest.mark.asyncio + async def test_returns_none_when_no_episodic_memory(self): + gw = MagicMock() + engine = ReflexionEngine(llm_gateway=gw, episodic_memory=None) + result = await engine.retrieve_prompt_reflection(task_input="test") + assert result is None + + @pytest.mark.asyncio + async def test_returns_best_reflection_by_score(self): + episodic = MagicMock() + # Two results: score 0.6 and 0.9 — should return 0.9 + items = [ + _make_memory_item(score=0.6, output_summary="prompt v1"), + _make_memory_item(score=0.9, output_summary="prompt v2"), + ] + episodic.search_prompt_reflections = AsyncMock(return_value=items) + gw = MagicMock() + engine = ReflexionEngine(llm_gateway=gw, episodic_memory=episodic) + + result = await engine.retrieve_prompt_reflection(task_input="test") + + assert result is not None + assert result["score"] == 0.9 + assert result["improved_prompt"] == "prompt v2" + + @pytest.mark.asyncio + async def test_filters_low_score_reflections(self): + """score <= 0.5 should not be returned.""" + episodic = MagicMock() + items = [_make_memory_item(score=0.3, output_summary="low score")] + episodic.search_prompt_reflections = AsyncMock(return_value=items) + gw = MagicMock() + engine = ReflexionEngine(llm_gateway=gw, episodic_memory=episodic) + + result = await engine.retrieve_prompt_reflection(task_input="test", min_score=0.5) + assert result is None + + @pytest.mark.asyncio + async def test_returns_none_when_no_results(self): + episodic = MagicMock() + episodic.search_prompt_reflections = AsyncMock(return_value=[]) + gw = MagicMock() + engine = ReflexionEngine(llm_gateway=gw, episodic_memory=episodic) + + result = await engine.retrieve_prompt_reflection(task_input="test") + assert result is None + + @pytest.mark.asyncio + async def test_returns_none_on_search_failure(self): + episodic = MagicMock() + episodic.search_prompt_reflections = AsyncMock(side_effect=RuntimeError("DB down")) + gw = MagicMock() + engine = ReflexionEngine(llm_gateway=gw, episodic_memory=episodic) + + result = await engine.retrieve_prompt_reflection(task_input="test") + assert result is None + + @pytest.mark.asyncio + async def test_returns_reflection_fields_complete(self): + episodic = MagicMock() + items = [ + _make_memory_item( + score=0.85, + output_summary="improved prompt", + reflection="what went wrong", + version=3, + ) + ] + episodic.search_prompt_reflections = AsyncMock(return_value=items) + gw = MagicMock() + engine = ReflexionEngine(llm_gateway=gw, episodic_memory=episodic) + + result = await engine.retrieve_prompt_reflection(task_input="test") + + assert result is not None + assert result["improved_prompt"] == "improved prompt" + assert result["reflection"] == "what went wrong" + assert result["version"] == 3 + assert result["score"] == 0.85 + assert "task_hash" in result + + @pytest.mark.asyncio + async def test_custom_min_score_threshold(self): + """min_score=0.7 filters out score=0.6.""" + episodic = MagicMock() + items = [_make_memory_item(score=0.6, output_summary="medium")] + episodic.search_prompt_reflections = AsyncMock(return_value=items) + gw = MagicMock() + engine = ReflexionEngine(llm_gateway=gw, episodic_memory=episodic) + + result = await engine.retrieve_prompt_reflection(task_input="test", min_score=0.7) + assert result is None + + +# ── TeamOrchestrator._decompose_task with reflection ────────────────── + + +class TestDecomposeWithReflection: + """U6/R13: _decompose_task prepends improved_prompt when reflection found.""" + + @pytest.mark.asyncio + async def test_decompose_prepends_reflection_when_found(self): + """When reflexion_engine returns a reflection, the decomposition + prompt includes the improved_prompt.""" + team = _make_team_with_experts() + gw = _make_llm_gateway_mock() + team._experts["lead"].agent._llm_gateway = gw + + # Mock reflexion_engine + reflexion = MagicMock(spec=ReflexionEngine) + reflexion.retrieve_prompt_reflection = AsyncMock( + return_value={ + "improved_prompt": "USE THIS IMPROVED APPROACH", + "score": 0.85, + "reflection": "past mistake", + "version": 2, + "task_hash": "abc", + } + ) + + orchestrator = TeamOrchestrator(team, reflexion_engine=reflexion) + await orchestrator._decompose_task(team.lead_expert, "test task") + + # Verify retrieve was called + reflexion.retrieve_prompt_reflection.assert_awaited_once() + # Verify the prompt sent to LLM includes the improved_prompt + call_kwargs = gw.chat.await_args.kwargs + messages = call_kwargs.get("messages") or gw.chat.await_args.args[0] + prompt_content = messages[0]["content"] if isinstance(messages, list) else str(messages) + assert "USE THIS IMPROVED APPROACH" in prompt_content + assert "Historical Reflection" in prompt_content + + @pytest.mark.asyncio + async def test_decompose_uses_default_when_no_reflexion_engine(self): + """No reflexion_engine → default prompt (backward compat).""" + team = _make_team_with_experts() + gw = _make_llm_gateway_mock() + team._experts["lead"].agent._llm_gateway = gw + + orchestrator = TeamOrchestrator(team, reflexion_engine=None) + await orchestrator._decompose_task(team.lead_expert, "test task") + + # Verify default prompt (no Historical Reflection section) + call_kwargs = gw.chat.await_args.kwargs + messages = call_kwargs.get("messages") or gw.chat.await_args.args[0] + prompt_content = messages[0]["content"] if isinstance(messages, list) else str(messages) + assert "Historical Reflection" not in prompt_content + + @pytest.mark.asyncio + async def test_decompose_uses_default_when_no_reflection_found(self): + """reflexion_engine returns None → default prompt.""" + team = _make_team_with_experts() + gw = _make_llm_gateway_mock() + team._experts["lead"].agent._llm_gateway = gw + + reflexion = MagicMock(spec=ReflexionEngine) + reflexion.retrieve_prompt_reflection = AsyncMock(return_value=None) + + orchestrator = TeamOrchestrator(team, reflexion_engine=reflexion) + await orchestrator._decompose_task(team.lead_expert, "test task") + + call_kwargs = gw.chat.await_args.kwargs + messages = call_kwargs.get("messages") or gw.chat.await_args.args[0] + prompt_content = messages[0]["content"] if isinstance(messages, list) else str(messages) + assert "Historical Reflection" not in prompt_content + + @pytest.mark.asyncio + async def test_decompose_uses_default_when_retrieval_fails(self): + """reflexion_engine.retrieve raises → default prompt (non-blocking).""" + team = _make_team_with_experts() + gw = _make_llm_gateway_mock() + team._experts["lead"].agent._llm_gateway = gw + + reflexion = MagicMock(spec=ReflexionEngine) + reflexion.retrieve_prompt_reflection = AsyncMock(side_effect=RuntimeError("search failed")) + + orchestrator = TeamOrchestrator(team, reflexion_engine=reflexion) + await orchestrator._decompose_task(team.lead_expert, "test task") + + # Default prompt used despite retrieval failure + call_kwargs = gw.chat.await_args.kwargs + messages = call_kwargs.get("messages") or gw.chat.await_args.args[0] + prompt_content = messages[0]["content"] if isinstance(messages, list) else str(messages) + assert "Historical Reflection" not in prompt_content + + @pytest.mark.asyncio + async def test_decompose_skips_reflection_when_no_improved_prompt(self): + """reflexion_engine returns dict without improved_prompt → no hint.""" + team = _make_team_with_experts() + gw = _make_llm_gateway_mock() + team._experts["lead"].agent._llm_gateway = gw + + reflexion = MagicMock(spec=ReflexionEngine) + reflexion.retrieve_prompt_reflection = AsyncMock( + return_value={"improved_prompt": "", "score": 0.8} # empty improved_prompt + ) + + orchestrator = TeamOrchestrator(team, reflexion_engine=reflexion) + await orchestrator._decompose_task(team.lead_expert, "test task") + + call_kwargs = gw.chat.await_args.kwargs + messages = call_kwargs.get("messages") or gw.chat.await_args.args[0] + prompt_content = messages[0]["content"] if isinstance(messages, list) else str(messages) + assert "Historical Reflection" not in prompt_content