File size: 3,364 Bytes
bea8b2d | 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 | """Backend protocol — what the agent loop needs from any LLM provider.
The loop is provider-agnostic. Each backend owns its own conversation state
and per-API translation; the loop only sees the unified `AgentTurn` shape.
Two backends ship:
* `ClaudeBackend` — anthropic.AsyncAnthropic, system as a top-level field,
tool_result blocks batched in one user message.
* `QwenHFBackend` — huggingface_hub.AsyncInferenceClient.chat_completion,
system as the first message, tool_result as one role="tool" message per
call. Routes to whichever provider serves the chosen Qwen model.
A future LiveQwenBackend talking to a self-hosted vLLM-on-MI300X endpoint
slots in identically — it just speaks the OpenAI-compatible shape.
"""
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any
@dataclass
class ToolCall:
"""One tool call requested by the model in a turn."""
id: str
"""Provider-assigned identifier. Used to correlate the eventual tool_result."""
name: str
input: dict[str, Any] = field(default_factory=dict)
@dataclass
class AgentTurn:
"""One turn's response, normalized across providers."""
text_blocks: list[str] = field(default_factory=list)
"""Free-text the model produced this turn (rendered as `thought` SSE events)."""
tool_calls: list[ToolCall] = field(default_factory=list)
stop_reason: str = "end_turn"
"""One of: 'end_turn', 'tool_use', 'max_tokens', 'other'.
The loop breaks on 'end_turn'. Other values keep iterating up to MAX_STEPS.
"""
class Backend(ABC):
"""Pluggable LLM driver for the agent loop.
Lifecycle:
backend = SomeBackend(system_prompt=...)
backend.add_user_message("Audit this workload: ...")
for step in range(MAX_STEPS):
turn = await backend.next_turn(tool_schemas)
... yield events ...
for tc in turn.tool_calls:
result = call_tool(tc)
backend.add_tool_result(tc.id, tc.name, result.content, is_error=...)
if turn.stop_reason == "end_turn":
break
"""
name: str = "base"
"""Short label used in /healthz and logs (e.g. 'claude', 'qwen-hf')."""
@abstractmethod
def add_user_message(self, content: str) -> None:
"""Append a user message to the internal conversation."""
@abstractmethod
async def next_turn(self, tool_schemas: list[dict[str, Any]]) -> AgentTurn:
"""Run one turn against the provider. Updates internal state so the
next call to `next_turn` already has the assistant's last response in
context. Tool schemas are in the loop's neutral shape:
{"name": str, "description": str, "input_schema": json-schema dict}
Each backend translates to the provider's own tool format.
"""
@abstractmethod
def add_tool_result(
self,
tool_call_id: str,
name: str,
content: str,
is_error: bool,
) -> None:
"""Record a tool result for the model to consume on the next turn.
Different providers want different shapes (Claude batches into one
user message; OpenAI/Qwen wants one role='tool' message per call).
Implementations buffer or append as appropriate.
"""
|