Skip to content

stream.drivers.base

stream.drivers.base

Shared base for direct-API agent drivers.

ApiDriver implements the provider-agnostic tool-use loop: prompt the model, execute any tool calls in-process against the zettelkasten store, feed the results back, and repeat until the model stops calling tools. Concrete drivers (ClaudeDriver, GptDriver) implement three small provider hooks for the wire-format differences.

Nothing here imports a provider SDK; subclasses import theirs lazily.

ToolCall dataclass

A single tool invocation requested by the model.

Source code in stream/drivers/base.py 
20
21
22
23
24
25
26
@dataclass
class ToolCall:
    """A single tool invocation requested by the model."""

    id: str
    name: str
    input: dict[str, Any] = field(default_factory=dict)

ProviderTurn dataclass

One assistant turn parsed into a provider-neutral shape.

Source code in stream/drivers/base.py 
29
30
31
32
33
34
35
@dataclass
class ProviderTurn:
    """One assistant turn parsed into a provider-neutral shape."""

    text: str = ""
    tool_calls: list[ToolCall] = field(default_factory=list)
    raw: Any = None

ApiDriver

Provider-agnostic tool-use loop.

Subclasses set name/default_model and implement _format_tools, _complete, _record_assistant, _record_tool_results, and _init_messages.

Source code in stream/drivers/base.py 
 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
class ApiDriver:
    """Provider-agnostic tool-use loop.

    Subclasses set ``name``/``default_model`` and implement ``_format_tools``,
    ``_complete``, ``_record_assistant``, ``_record_tool_results``, and
    ``_init_messages``.
    """

    name: str = "api"
    supports_logprobs: bool = False
    default_model: str = ""
    max_tool_iterations: int = 24

    # -- public protocol ----------------------------------------------------

    def run_agent(
        self,
        *,
        role: str,
        persona: str,
        prompt: str,
        tools: list[Tool],
        model: str = "",
        ctx: dict[str, Any] | None = None,
    ) -> AgentResult:
        tool_map = {t.name: t for t in tools}
        tool_specs = self._format_tools(tools)
        chosen_model = model or self.default_model
        messages = self._init_messages(prompt)
        calls = 0
        final_text = ""

        for _ in range(self.max_tool_iterations):
            turn = self._complete(persona, messages, tool_specs, chosen_model)
            self._record_assistant(messages, turn)
            if not turn.tool_calls:
                final_text = turn.text
                break
            results: list[tuple[ToolCall, Any]] = []
            for tc in turn.tool_calls:
                calls += 1
                tool = tool_map.get(tc.name)
                if tool is None:
                    out: Any = {"error": f"unknown tool: {tc.name}"}
                else:
                    try:
                        out = tool.execute(**(tc.input or {}))
                    except Exception as exc:  # surfaced back to the model
                        out = {"error": f"{type(exc).__name__}: {exc}"}
                results.append((tc, out))
            self._record_tool_results(messages, results)
        else:
            # Loop budget exhausted without a tool-free turn.
            final_text = final_text or (
                "RESULT: FAIL\nNOTES: tool-use loop exhausted without completion"
            )

        return AgentResult(text=final_text, ok=True, tool_calls=calls)

    def score(
        self,
        prompt: str,
        *,
        choices: list[str] | None = None,
        model: str = "",
    ) -> Score:
        raise NotImplementedError(
            f"{self.name} driver does not support score(); "
            "use a logprob-capable driver (e.g. GptDriver against vLLM)."
        )

    # -- provider hooks (override) -----------------------------------------

    def _init_messages(self, prompt: str) -> list[Any]:
        raise NotImplementedError

    def _format_tools(self, tools: list[Tool]) -> Any:
        raise NotImplementedError

    def _complete(
        self, system: str, messages: list[Any], tool_specs: Any, model: str
    ) -> ProviderTurn:
        raise NotImplementedError

    def _record_assistant(self, messages: list[Any], turn: ProviderTurn) -> None:
        raise NotImplementedError

    def _record_tool_results(
        self, messages: list[Any], results: list[tuple[ToolCall, Any]]
    ) -> None:
        raise NotImplementedError