88 lines
2.2 KiB
Python
88 lines
2.2 KiB
Python
"""Compatibility wrappers around backend adapter implementations."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from app.backend_adapters import build_backend_adapter_from_settings
|
|
|
|
|
|
def _adapter():
|
|
return build_backend_adapter_from_settings()
|
|
|
|
|
|
async def fetch_assistant_config(assistant_id: str) -> Optional[Dict[str, Any]]:
|
|
"""Fetch assistant config payload from backend adapter."""
|
|
return await _adapter().fetch_assistant_config(assistant_id)
|
|
|
|
|
|
async def create_history_call_record(
|
|
*,
|
|
user_id: int,
|
|
assistant_id: Optional[str],
|
|
source: str = "debug",
|
|
) -> Optional[str]:
|
|
"""Create a call record via backend history API and return call_id."""
|
|
return await _adapter().create_call_record(
|
|
user_id=user_id,
|
|
assistant_id=assistant_id,
|
|
source=source,
|
|
)
|
|
|
|
|
|
async def add_history_transcript(
|
|
*,
|
|
call_id: str,
|
|
turn_index: int,
|
|
speaker: str,
|
|
content: str,
|
|
start_ms: int,
|
|
end_ms: int,
|
|
confidence: Optional[float] = None,
|
|
duration_ms: Optional[int] = None,
|
|
) -> bool:
|
|
"""Append a transcript segment to backend history."""
|
|
return await _adapter().add_transcript(
|
|
call_id=call_id,
|
|
turn_index=turn_index,
|
|
speaker=speaker,
|
|
content=content,
|
|
start_ms=start_ms,
|
|
end_ms=end_ms,
|
|
confidence=confidence,
|
|
duration_ms=duration_ms,
|
|
)
|
|
|
|
|
|
async def finalize_history_call_record(
|
|
*,
|
|
call_id: str,
|
|
status: str,
|
|
duration_seconds: int,
|
|
) -> bool:
|
|
"""Finalize a call record with status and duration."""
|
|
return await _adapter().finalize_call_record(
|
|
call_id=call_id,
|
|
status=status,
|
|
duration_seconds=duration_seconds,
|
|
)
|
|
|
|
|
|
async def search_knowledge_context(
|
|
*,
|
|
kb_id: str,
|
|
query: str,
|
|
n_results: int = 5,
|
|
) -> List[Dict[str, Any]]:
|
|
"""Search backend knowledge base and return retrieval results."""
|
|
return await _adapter().search_knowledge_context(
|
|
kb_id=kb_id,
|
|
query=query,
|
|
n_results=n_results,
|
|
)
|
|
|
|
|
|
async def fetch_tool_resource(tool_id: str) -> Optional[Dict[str, Any]]:
|
|
"""Fetch tool resource configuration from backend API."""
|
|
return await _adapter().fetch_tool_resource(tool_id)
|