asterisk_ai_voice_agent/llm.py
Rama main · réplica de lectura del repositorio original.
"""
LLM: Anthropic Claude, streamed, with tool_use.
Per-call instance: takes the persona's system prompt + tool specs, accumulates
caller turns, streams Claude's reply token-by-token, and surfaces tool_use
blocks so the orchestrator can execute them (via your webhook) and feed the
result back.
Swapping providers: keep the stream_reply() event shape ({'kind': ...}) and the
orchestrator doesn't care what's underneath.
Derived from ICTContact (https://www.ictcontact.com).
ICT Innovations (https://www.ictinnovations.com) - ICT Vision (https://ict.vision)
Author: Tahir Almas. MIT licensed.
"""
import json
import logging
from typing import AsyncIterator, List, Dict, Optional
from anthropic import AsyncAnthropic
log = logging.getLogger("ai.llm")
DEFAULT_MAX_TOKENS = 512
_PHP_TYPE_TO_JSON = {"string": "string", "integer": "integer",
"number": "number", "boolean": "boolean"}
def _to_anthropic_tools(specs: Dict[str, dict], enabled: List[str]) -> List[dict]:
"""Convert a spec dict { name: { description, parameters: {...} } } into the
Anthropic SDK tool format, for the tools named in `enabled`."""
out = []
for name in enabled:
spec = specs.get(name)
if not spec:
continue
props = {}
required = []
for pname, p in (spec.get("parameters") or {}).items():
props[pname] = {
"type": _PHP_TYPE_TO_JSON.get(p.get("type", "string"), "string"),
"description": p.get("description", ""),
}
# Treat every declared param as required for v1 - Claude is less
# likely to hallucinate args this way.
required.append(pname)
out.append({
"name": name,
"description": spec.get("description", ""),
"input_schema": {"type": "object", "properties": props, "required": required},
})
return out
class LLM:
def __init__(self, provider: str, model: str, temperature: float,
system_prompt: str,
tool_specs: Optional[Dict[str, dict]] = None,
tools_enabled: Optional[List[str]] = None,
api_key: Optional[str] = None):
self.provider = provider
self.model = model or "claude-sonnet-4-6"
self.temperature = float(temperature) if temperature is not None else 0.4
self.system_prompt = system_prompt or ""
self.tool_specs = tool_specs or {}
self.tools_enabled = tools_enabled or []
self.api_key = api_key
self.history: List[Dict] = []
self._client: Optional[AsyncAnthropic] = None
self._anth_tools = _to_anthropic_tools(self.tool_specs, self.tools_enabled)
async def start(self) -> None:
if self.provider != "anthropic":
log.warning("llm provider=%s not implemented; using anthropic", self.provider)
if not self.api_key:
raise RuntimeError("Anthropic API key not configured (providers.anthropic.api_key)")
self._client = AsyncAnthropic(api_key=self.api_key)
log.info("LLM ready: model=%s temp=%s tools=%d",
self.model, self.temperature, len(self._anth_tools))
async def add_user(self, text: str) -> None:
text = (text or "").strip()
if not text:
return
# A barge-in can leave the last history entry as role=user. Merge instead
# of appending a second consecutive user message.
if self.history and self.history[-1]["role"] == "user":
prev = self.history[-1]
if isinstance(prev["content"], str):
prev["content"] = [{"type": "text", "text": prev["content"]}]
prev["content"].append({"type": "text", "text": text})
return
self.history.append({"role": "user", "content": text})
async def add_assistant_content(self, content) -> None:
"""Append an assistant message (str or block list) to history verbatim."""
if isinstance(content, str):
content = content.strip()
if not content:
return
self.history.append({"role": "assistant", "content": content})
async def replace_last_assistant(self, text: str) -> None:
"""Make the most recent assistant message say `text`, or drop it if empty.
For barge-in. stream_reply() appends the assistant message only once the
stream completes, so an abandoned reply leaves none, while a completed
one holds the full text and any tool_use blocks. After an interrupt the
record should be the part the caller actually heard, with no tool_use,
because the tools are not going to be run.
"""
if self.history and self.history[-1]["role"] == "assistant":
self.history.pop()
text = (text or "").strip()
if text:
self.history.append({"role": "assistant", "content": text})
async def add_tool_result(self, tool_use_id: str, result_obj) -> None:
"""Feed a tool's result back to Claude so it can continue."""
result_str = json.dumps(result_obj) if not isinstance(result_obj, str) else result_obj
self.history.append({
"role": "user",
"content": [{"type": "tool_result", "tool_use_id": tool_use_id, "content": result_str}],
})
async def stream_reply(self) -> AsyncIterator[dict]:
"""
Stream the assistant turn. Yields a mix of:
{'kind': 'text', 'text': str} - partial text delta
{'kind': 'tool', 'name': str, 'args': dict, 'id': str} - run a tool
{'kind': 'end', 'stop_reason': str, 'assistant_blocks': list}
After 'end', the orchestrator runs any tool calls, calls
add_tool_result(...) per tool, and calls stream_reply() again for
Claude's next turn.
"""
if self._client is None:
await self.start()
kwargs = dict(
model=self.model,
max_tokens=DEFAULT_MAX_TOKENS,
temperature=self.temperature,
system=self.system_prompt,
messages=self.history,
)
if self._anth_tools:
kwargs["tools"] = self._anth_tools
assistant_blocks: List[Dict] = []
cur_text = ""
cur_tool_id: Optional[str] = None
cur_tool_name: Optional[str] = None
cur_tool_input_json = ""
stop_reason = "end_turn"
try:
async with self._client.messages.stream(**kwargs) as stream:
async for event in stream:
t = getattr(event, "type", None)
if t == "content_block_start":
block = event.content_block
if block.type == "tool_use":
cur_tool_id = block.id
cur_tool_name = block.name
cur_tool_input_json = ""
elif block.type == "text":
cur_text = ""
elif t == "content_block_delta":
d = event.delta
if getattr(d, "type", "") == "text_delta":
cur_text += d.text
yield {"kind": "text", "text": d.text}
elif getattr(d, "type", "") == "input_json_delta":
cur_tool_input_json += d.partial_json
elif t == "content_block_stop":
if cur_tool_id:
try:
args = json.loads(cur_tool_input_json) if cur_tool_input_json else {}
except Exception:
args = {}
assistant_blocks.append({
"type": "tool_use", "id": cur_tool_id,
"name": cur_tool_name, "input": args,
})
yield {"kind": "tool", "name": cur_tool_name, "args": args, "id": cur_tool_id}
cur_tool_id = None
cur_tool_name = None
cur_tool_input_json = ""
elif cur_text:
assistant_blocks.append({"type": "text", "text": cur_text})
cur_text = ""
elif t == "message_delta":
if getattr(event, "delta", None) and getattr(event.delta, "stop_reason", None):
stop_reason = event.delta.stop_reason
except Exception as e:
log.exception("Anthropic stream error: %s", e)
msg = "I'm sorry, I had a problem. Could you say that again?"
yield {"kind": "text", "text": msg}
assistant_blocks.append({"type": "text", "text": msg})
stop_reason = "end_turn"
# Append the full assistant message - required for the tool_result loop.
if assistant_blocks:
await self.add_assistant_content(assistant_blocks)
yield {"kind": "end", "stop_reason": stop_reason, "assistant_blocks": assistant_blocks}
async def close(self) -> None:
if self._client:
await self._client.close()