Spaces:
Sleeping
Sleeping
File size: 6,520 Bytes
116524e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 | """RR-specific tool registrars and dependency container.
Generic tools (execute_code, recurse) are provided by
:mod:`ace.core.recursive_agent`. This module adds RR-specific
tools and the RR dependency container.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, Optional, Union
from pydantic_ai import ModelRetry, RunContext
from ace.core.context import SkillbookView
from ace.core.recursive_agent import AgenticDeps
from ace.core.skillbook import Skillbook
if TYPE_CHECKING:
from pydantic_ai import Agent as PydanticAgent
from .config import RecursiveConfig
# ------------------------------------------------------------------
# Dependency container
# ------------------------------------------------------------------
@dataclass
class RRDeps(AgenticDeps):
"""Dependencies injected into RR tool calls via ``RunContext``.
Extends :class:`AgenticDeps` with RR-specific trace and skillbook fields.
``sandbox`` is inherited from :class:`AgenticDeps``.
``skillbook`` (optional) is the real :class:`Skillbook` — provided so the
read-only ``search_skillbook`` and ``read_skill`` tools can inspect
strategies without the agent having to scan serialized text.
"""
trace_data: dict[str, Any] = field(default_factory=dict)
skillbook_text: str = ""
skillbook: Optional[Union[Skillbook, SkillbookView]] = None
thoughts: list[dict[str, Any]] = field(default_factory=list)
# ------------------------------------------------------------------
# RR-specific tool registrars
# ------------------------------------------------------------------
def register_output_validator(agent: "PydanticAgent[RRDeps, Any]") -> None:
"""Register the standard output validator on any RR agent."""
@agent.output_validator
def validate_output(ctx: RunContext[RRDeps], output: Any) -> Any:
"""Ensure the agent explored data before concluding."""
if ctx.deps.iteration < 1:
raise ModelRetry(
"You haven't explored the data enough. "
"Use execute_code first, then provide your final answer."
)
return output
def register_read_skill(agent: "PydanticAgent[RRDeps, Any]") -> None:
"""Register the ``read_skill`` read-only tool.
Returns the full skill payload (including counters) for a given ID,
or a ``not found`` message. No sandbox, no mutation.
"""
@agent.tool
def read_skill(ctx: RunContext[RRDeps], skill_id: str) -> dict[str, Any]:
"""Look up a skill by ID."""
sb = ctx.deps.skillbook
if sb is None:
return {"error": "skillbook unavailable"}
skill = sb.get_skill(skill_id)
if skill is None:
return {"error": f"skill not found: {skill_id}"}
return {
"id": skill.id,
"section": skill.section,
"keywords": list(skill.keywords),
"issue": skill.issue,
"insight": skill.insight,
"active": skill.active,
"used_count": skill.used_count,
"helpful_count": skill.helpful_count,
"harmful_count": skill.harmful_count,
"neutral_count": skill.neutral_count,
"occurrences": [source.to_dict() for source in skill.occurrences],
}
def register_think(agent: "PydanticAgent[RRDeps, Any]") -> None:
"""Register the ``think`` narration channel.
``think`` is the home for the model's running narration during a
tool-use turn — what it just confirmed, what it is checking next, brief
observations. This keeps prose out of ``execute_code`` stdout, where
Python should only print compact structured evidence. The final
conclusion still belongs in ``ReflectorOutput`` (the only sink that
propagates to the SkillManager); ``think`` notes are surfaced in
``output.raw["thoughts"]`` for inspection only.
"""
@agent.tool
def think(
ctx: RunContext[RRDeps],
thought: str,
evidence_refs: list[str] | None = None,
) -> dict[str, Any]:
"""Narrate your working state during the run.
Use this for mid-run prose: "checking the constraint window next",
"the mismatch is confirmed", "the decisive message is at index 12".
Use it freely — it is the right home for everything you would
naturally say while working. The final conclusion still goes in
``ReflectorOutput``; reusable data still lives in sandbox variables
via ``execute_code``.
"""
normalized = thought.strip()
if not normalized:
raise ModelRetry("Thought must be non-empty.")
refs = [ref.strip() for ref in (evidence_refs or []) if ref.strip()]
entry = {
"thought": normalized,
"evidence_refs": refs,
}
ctx.deps.thoughts.append(entry)
return {
"ok": True,
"thought_count": len(ctx.deps.thoughts),
}
def register_search_skillbook(agent: "PydanticAgent[RRDeps, Any]") -> None:
"""Register the ``search_skillbook`` read-only tool.
Returns the top-k skills most relevant to the query via embedding
similarity. Falls back to the first k active skills if embeddings are
unavailable.
"""
@agent.tool
def search_skillbook(
ctx: RunContext[RRDeps], query: str, top_k: int = 5
) -> list[dict[str, Any]]:
"""Search for skills matching a natural-language query."""
sb = ctx.deps.skillbook
if sb is None:
return [{"error": "skillbook unavailable"}]
from ace.implementations.skill_rendering import retrieve_top_k
actual_sb = sb._sb if isinstance(sb, SkillbookView) else sb
results = retrieve_top_k(actual_sb, query, top_k=top_k)
return [
{
"id": s.id,
"section": s.section,
"keywords": list(s.keywords),
"issue": s.issue,
"insight": s.insight,
"active": s.active,
"used_count": s.used_count,
"helpful_count": s.helpful_count,
"harmful_count": s.harmful_count,
"neutral_count": s.neutral_count,
}
for s in results
]
|