109 lines
3.3 KiB
Python
109 lines
3.3 KiB
Python
"""Long-lived Cell worker for RabbitMQ ``call.execute`` messages."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import signal
|
|
import time
|
|
from pathlib import Path
|
|
from threading import Event
|
|
from typing import Any
|
|
|
|
from .ai_runtime import ConversationEngine, load_prompt
|
|
from .bailian import BailianASR, BailianLLM, BailianTTS, build_bailian_config
|
|
from .core import PikaBroker
|
|
from .real_cell import (
|
|
CellCallConfig,
|
|
CellExecutionLedger,
|
|
RealCellCall,
|
|
RealCellWorker,
|
|
load_cell_routes,
|
|
)
|
|
|
|
|
|
def _profile(path: str) -> dict[str, Any]:
|
|
candidate = Path(path)
|
|
try:
|
|
value = json.loads(candidate.read_text(encoding="utf-8"))
|
|
except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
|
|
raise RuntimeError("Cell AI profile is unavailable") from exc
|
|
if not isinstance(value, dict):
|
|
raise TypeError("Cell AI profile must be an object")
|
|
return value
|
|
|
|
|
|
def _integer(value: Any, default: int, field: str, minimum: int = 1) -> int:
|
|
try:
|
|
parsed = int(value)
|
|
except (TypeError, ValueError, OverflowError) as exc:
|
|
raise RuntimeError(f"{field} must be an integer") from exc
|
|
if parsed < minimum:
|
|
raise RuntimeError(f"{field} must be at least {minimum}")
|
|
return parsed
|
|
|
|
|
|
def build_worker() -> RealCellWorker:
|
|
profile = _profile(
|
|
os.environ.get(
|
|
"AGENT_CALL_AI_PROFILE", "/app/configs/ai-test.bailian.example.yaml"
|
|
)
|
|
)
|
|
prompt_path = Path(
|
|
os.environ.get("AGENT_CALL_PROMPT_FILE", "/app/prompts/test-call.txt")
|
|
)
|
|
prompt = load_prompt(
|
|
prompt_path,
|
|
_integer(profile.get("max_prompt_bytes", 32768), 32768, "max_prompt_bytes"),
|
|
)
|
|
agent_version_id = str(profile.get("agent_version_id", ""))
|
|
if not agent_version_id:
|
|
raise RuntimeError("agent_version_id is required in the Cell AI profile")
|
|
config = build_bailian_config(
|
|
agent_version_id,
|
|
prompt,
|
|
str(profile.get("model", "qwen-plus")),
|
|
profile.get("tts_model"),
|
|
profile.get("voice"),
|
|
str(profile.get("language", "zh-CN")),
|
|
profile.get("asr_model"),
|
|
str(profile.get("opening", "")),
|
|
_integer(profile.get("max_turns", 20), 20, "max_turns"),
|
|
_integer(profile.get("max_duration_ms", 120000), 120000, "max_duration_ms"),
|
|
)
|
|
asr = BailianASR.from_env()
|
|
engine = ConversationEngine(
|
|
config,
|
|
llm=BailianLLM.from_env(),
|
|
tts=BailianTTS.from_env(),
|
|
asr=asr,
|
|
)
|
|
broker = PikaBroker(os.environ.get("RABBITMQ_URL", ""))
|
|
tenant_key = os.environ.get("AGENT_CALL_TENANT_KEY", "")
|
|
ledger = CellExecutionLedger(
|
|
os.environ.get("AGENT_CALL_EXECUTION_LEDGER", "/data/cell-executions.sqlite3")
|
|
)
|
|
routes = load_cell_routes(os.environ.get("CELL_ROUTE_MAP_JSON"))
|
|
return RealCellWorker(
|
|
broker,
|
|
tenant_key,
|
|
ledger,
|
|
RealCellCall(CellCallConfig.from_env(), engine, asr),
|
|
routes=routes,
|
|
)
|
|
|
|
|
|
def main() -> int:
|
|
stopped = Event()
|
|
signal.signal(signal.SIGTERM, lambda _signum, _frame: stopped.set())
|
|
signal.signal(signal.SIGINT, lambda _signum, _frame: stopped.set())
|
|
worker = build_worker()
|
|
while not stopped.is_set():
|
|
if worker.process_once() is None:
|
|
time.sleep(0.2)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|