Index

agents

Agents primitive — multi-turn reasoning and tool use.

Classes

AgentContext dataclass

AgentContext(
    conversation: Conversation = Conversation(),
    tools: List[str] = list(),
    memory_results: List[Any] = list(),
    metadata: Dict[str, Any] = dict(),
)

Runtime context handed to an agent on each invocation.

AgentResult dataclass

AgentResult(
    content: str,
    tool_results: List[ToolResult] = list(),
    turns: int = 0,
    metadata: Dict[str, Any] = dict(),
)

Result returned after an agent completes a run.

BaseAgent

BaseAgent(
    engine: InferenceEngine,
    model: str,
    *,
    bus: Optional[EventBus] = None,
    temperature: Optional[float] = None,
    max_tokens: Optional[int] = None,
    prompt_builder: Optional[Any] = None,
)

Bases: ABC

Base class for all agent implementations.

Subclasses must be registered via @AgentRegistry.register("name") to become discoverable.

Provides concrete helper methods that eliminate boilerplate in subclasses:

• :meth:_emit_turn_start / :meth:_emit_turn_end -- event bus
• :meth:_build_messages -- conversation + system prompt assembly
• :meth:_generate -- delegates to engine with stored defaults
• :meth:_max_turns_result -- standard max-turns-exceeded result
• :meth:_strip_think_tags -- remove <think> blocks

Source code in src/diapason/agents/_stubs.py

def __init__(
    self,
    engine: InferenceEngine,
    model: str,
    *,
    bus: Optional[EventBus] = None,
    temperature: Optional[float] = None,
    max_tokens: Optional[int] = None,
    prompt_builder: Optional[Any] = None,
) -> None:
    self._engine = engine
    self._model = model
    self._bus = bus
    self._prompt_builder = prompt_builder

    # Three-tier resolution: explicit arg > config > class default > hardcoded
    if temperature is not None and max_tokens is not None:
        self._temperature = temperature
        self._max_tokens = max_tokens
    else:
        try:
            cfg = load_config()
            self._temperature = (
                temperature
                if temperature is not None
                else cfg.intelligence.temperature
            )
            self._max_tokens = (
                max_tokens
                if max_tokens is not None
                else cfg.intelligence.max_tokens
            )
        except Exception:
            self._temperature = (
                temperature
                if temperature is not None
                else getattr(self, "_default_temperature", 0.7)
            )
            self._max_tokens = (
                max_tokens
                if max_tokens is not None
                else getattr(self, "_default_max_tokens", 1024)
            )

Methods:

run abstractmethod

run(
    input: str,
    context: Optional[AgentContext] = None,
    **kwargs: Any,
) -> AgentResult

Execute the agent on input and return an AgentResult.

Source code in src/diapason/agents/_stubs.py

@abstractmethod
def run(
    self,
    input: str,
    context: Optional[AgentContext] = None,
    **kwargs: Any,
) -> AgentResult:
    """Execute the agent on *input* and return an ``AgentResult``."""

ToolUsingAgent

ToolUsingAgent(
    engine: InferenceEngine,
    model: str,
    *,
    tools: Optional[List["BaseTool"]] = None,
    bus: Optional[EventBus] = None,
    max_turns: Optional[int] = None,
    temperature: Optional[float] = None,
    max_tokens: Optional[int] = None,
    loop_guard_config: Optional[Any] = None,
    capability_policy: Any = _SECURITY_CONTROL_UNSET,
    boundary_guard: Optional[Any] = None,
    rate_limiter: Optional[Any] = None,
    agent_id: Optional[str] = None,
    interactive: bool = False,
    confirm_callback: Optional[Any] = None,
    skill_few_shot_examples: Optional[List[str]] = None,
    prompt_builder: Optional[Any] = None,
)

Bases: BaseAgent

Intermediate base for agents that accept and use tools.

Sets accepts_tools = True for CLI/SDK introspection, and initialises a :class:ToolExecutor from the provided tools.

Source code in src/diapason/agents/_stubs.py

def __init__(
    self,
    engine: InferenceEngine,
    model: str,
    *,
    tools: Optional[List["BaseTool"]] = None,  # noqa: F821
    bus: Optional[EventBus] = None,
    max_turns: Optional[int] = None,
    temperature: Optional[float] = None,
    max_tokens: Optional[int] = None,
    loop_guard_config: Optional[Any] = None,
    capability_policy: Any = _SECURITY_CONTROL_UNSET,
    boundary_guard: Optional[Any] = None,
    rate_limiter: Optional[Any] = None,
    agent_id: Optional[str] = None,
    interactive: bool = False,
    confirm_callback: Optional[Any] = None,
    skill_few_shot_examples: Optional[List[str]] = None,
    prompt_builder: Optional[Any] = None,
) -> None:
    super().__init__(
        engine,
        model,
        bus=bus,
        temperature=temperature,
        max_tokens=max_tokens,
        prompt_builder=prompt_builder,
    )
    from diapason.tools._stubs import ToolExecutor

    self._tools = tools or []
    # Plan 2B I3: store optimized few-shot examples for agents to inject
    # into their own system prompt templates as appropriate.
    self._skill_few_shot_examples = list(skill_few_shot_examples or [])
    _aid = agent_id or getattr(self, "agent_id", "")
    autoload_capability_policy = capability_policy is _SECURITY_CONTROL_UNSET
    resolved_capability_policy = (
        None if autoload_capability_policy else capability_policy
    )
    self._executor = ToolExecutor(
        self._tools,
        bus=bus,
        capability_policy=resolved_capability_policy,
        boundary_guard=boundary_guard,
        rate_limiter=rate_limiter,
        agent_id=_aid,
        interactive=interactive,
        confirm_callback=confirm_callback,
        # Omission means "load the configured policy". An explicit None
        # from a higher-level builder means security was deliberately
        # resolved as disabled and must remain fail-closed for tools that
        # carry capabilities.
        autoload_capability_policy=autoload_capability_policy,
    )
    # Resolve max_turns: explicit arg > config > class default > 10
    if max_turns is not None:
        self._max_turns = max_turns
    else:
        try:
            cfg = load_config()
            self._max_turns = cfg.agent.max_turns
        except Exception:
            self._max_turns = getattr(self, "_default_max_turns", 10)

    # Loop guard
    self._loop_guard = None
    try:
        from diapason.agents.loop_guard import LoopGuard, LoopGuardConfig

        if loop_guard_config is None:
            loop_guard_config = LoopGuardConfig()
        elif isinstance(loop_guard_config, dict):
            loop_guard_config = LoopGuardConfig(**loop_guard_config)
        if loop_guard_config.enabled:
            self._loop_guard = LoopGuard(loop_guard_config, bus=bus)
    except ImportError:
        pass