110 lines
3.1 KiB
Python
110 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI, WebSocket
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
from .config import EngineConfig, load_config
|
|
from .pipeline import run_product_voice_pipeline, run_voice_pipeline
|
|
|
|
WEBPAGE_DIR = Path(__file__).resolve().parent.parent / "examples" / "webpage"
|
|
|
|
|
|
@lru_cache(maxsize=8)
|
|
def get_config(path: str = "config.json") -> EngineConfig:
|
|
return load_config(path)
|
|
|
|
|
|
def _normalize_mount_path(path: str) -> str:
|
|
normalized = path.strip() or "/voice-demo"
|
|
if not normalized.startswith("/"):
|
|
normalized = f"/{normalized}"
|
|
return normalized.rstrip("/") or "/"
|
|
|
|
|
|
def create_app(config_path: str = "config.json") -> FastAPI:
|
|
config = get_config(config_path)
|
|
app = FastAPI(title="AI VideoAssistant Engine v5 Pipecat Minimal", version="0.1.0")
|
|
app.state.config = config
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=config.server.cors_origins,
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
webpage_mount = (
|
|
_normalize_mount_path(config.server.webpage_mount)
|
|
if config.server.serve_webpage
|
|
else None
|
|
)
|
|
|
|
@app.get("/health")
|
|
async def health() -> dict[str, object]:
|
|
return {
|
|
"status": "healthy",
|
|
"protocols": {
|
|
"/ws": "pipecat.websocket.protobuf",
|
|
"/ws-product": "va.ws.v1.json_base64",
|
|
},
|
|
"features": {
|
|
"product_text_input": True,
|
|
"product_text_interrupt": True,
|
|
"product_image_input": True,
|
|
},
|
|
"demo": webpage_mount,
|
|
"llm_backend": (
|
|
"fastgpt" if config.services.llm.is_fastgpt else "openai"
|
|
),
|
|
"llm_provider": config.services.llm.provider,
|
|
"stt_provider": config.services.stt.provider,
|
|
"tts_provider": config.services.tts.provider,
|
|
}
|
|
|
|
@app.websocket("/ws")
|
|
async def websocket_endpoint(websocket: WebSocket) -> None:
|
|
await websocket.accept()
|
|
await run_voice_pipeline(websocket, config)
|
|
|
|
@app.websocket("/ws-product")
|
|
async def product_websocket_endpoint(websocket: WebSocket) -> None:
|
|
await websocket.accept()
|
|
await run_product_voice_pipeline(websocket, config)
|
|
|
|
if config.server.serve_webpage and WEBPAGE_DIR.is_dir() and webpage_mount:
|
|
app.mount(
|
|
webpage_mount,
|
|
StaticFiles(directory=str(WEBPAGE_DIR), html=True),
|
|
name="webpage",
|
|
)
|
|
|
|
return app
|
|
|
|
|
|
app = create_app()
|
|
|
|
|
|
def main() -> None:
|
|
import uvicorn
|
|
|
|
parser = argparse.ArgumentParser(description="Run the minimal Pipecat voice engine.")
|
|
parser.add_argument("--config", default="config.json")
|
|
args = parser.parse_args()
|
|
|
|
config = load_config(args.config)
|
|
uvicorn.run(
|
|
create_app(args.config),
|
|
host=config.server.host,
|
|
port=config.server.port,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|