feat(iq): U6 Lead planning-time reflection retrieval (R12, R13)
- 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).
This commit is contained in:
Chiguyong
parent
9653b1d5f7
commit
a2deeac0d6
|
|
@ -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", ""),
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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
|
||||
Loading…
Reference in New Issue