Paygent integration

[nihal0514]

Pull Request: Paygent Integration

This pull request integrates Paygent into the Dograh backend to enable real-time cost-tracking and automatic agent/customer registration for both standard and speech-to-speech (Realtime) pipeline runs.

---

🚀 Overview of Changes

We have added robust, non-blocking telemetry and synchronization with the Paygent API. All HTTP interactions are executed asynchronously on background threads using a ThreadPoolExecutor to guarantee that they never block the FastAPI event loop or the live audio pipeline. All external API interactions are fully exception-guarded to ensure any network or Paygent failures never degrade the voice call experience.

1. Pipecat Submodule Enhancements (pipecat/)

To support full multimodal speech-to-speech cost tracking (such as OpenAI Realtime and Gemini Live), we extended the core Pipecat metrics layer to preserve and pass-through the underlying rich usage metadata:

• src/pipecat/metrics/metrics.py: Added an optional raw_usage_metadata field to the standard LLMTokenUsage schema.
• src/pipecat/services/google/gemini_live/llm.py: Extracted the rich usage_metadata payload (containing detailed modality splits for prompt, response, tool-use, and cache) and attached it to LLMTokenUsage.
• src/pipecat/services/openai/realtime/llm.py: Captured and attached the full usage model dump (including audio, text, cached token breakdowns) to LLMTokenUsage.

2. Core Telemetry Aggregator (api/services/pipecat/paygent.py [NEW])

Created PaygentPipelineMetricsAggregator, a custom Pipecat FrameProcessor that intercepts metric frames and forwards events:

• Dual-Mode Tracking:
  • Standard Mode (STT + LLM + TTS): Tracks LLM turns via /api/v1/voice/llm, TTS characters via /api/v1/voice/tts (with intelligent fallback estimation based on completion tokens for providers like Deepgram WS that don't emit metrics), STT approximated from call duration via /api/v1/voice/stt, and finalizes with /api/v1/voice/indicator.
  • Realtime / STS Mode (Speech-to-Speech): Maps and forwards multimodal tokens (text, audio, image, video, cache) to /api/v1/voice/speech-to-speech. Includes special-case minute-based billing for providers like Grok and Ultravox.
• Resiliency: Built-in fire-and-forget networking, daemon threads, automatic host resolution (re-writing localhost -> host.docker.internal inside Docker environments), and retry logic for network timeouts or 5xx/429 status codes.

3. Asynchronous Agent Registration (api/services/pipecat/paygent_agent_sync.py [NEW])

Introduced a module to register Dograh workflows as Paygent voice agents:

• ensure_agent_async(workflow_id, workflow_name): Idempotently registers/updates a voice agent via /api/v2/agents/ensure. Safely runs as an asynchronous background task on the server's thread pool.
• Startup Backfill: Added a startup runner run_backfill_if_requested() to scan and sync existing active workflows to Paygent in a paginated, non-blocking manner at server startup if PAYGENT_BACKFILL=true is set.

4. Integration Points & Hookups

• api/app.py: Triggers the startup backfill during the lifespan startup sequence.
• api/routes/workflow.py: Hooks into the workflow creation flows (create_workflow, create_workflow_from_template) to register new workflows as voice agents in the background.
• api/services/pipecat/pipeline_builder.py: Integrated the new paygent_aggregator as an optional processor in pipeline structures.
• api/services/pipecat/run_pipeline.py: Sourced all Paygent environment configurations, derived end-customer identifiers dynamically (telephony outbound maps to called number, inbound maps to caller number, and web calls use web-call), instantiated PaygentPipelineMetricsAggregator, and injected it into the running call pipeline.
• docker-compose.override.yaml [NEW]: Added a local development override to set context and load .env configurations automatically for the api service.

---

🛠️ Configuration & Environment Variables

The integration is fully optional and nil-safe. If PAYGENT_API_KEY is not present in the environment, all modules gracefully become silent no-ops.

# Essential configuration
PAYGENT_API_KEY=pk_...                  # Setting this enables cost tracking
PAYGENT_BASE_URL=https://cp-api.withpaygent.com  # Target Paygent server URL

# Optional agent-sync billing controls
PAYGENT_AGENT_PRICING_ENABLED=true      # Enable flat rate pricing configuration
PAYGENT_AGENT_PRICE_PER_MINUTE=0.015    # Set flat price per minute 
PAYGENT_AGENT_INDICATOR_ID=per-minute   # Paygent indicator name

# Startup tools
PAYGENT_BACKFILL=true                   # Sync existing active workflows at boot

---

🧪 Verification & Testing

• Verified that when PAYGENT_API_KEY is absent, standard and realtime call sessions spin up with zero impact or log noise.
• Verified that customer IDs map dynamically to caller/called phone numbers on telephony runs, or degrade gracefully to web-call for WebRTC sessions.
• Verified that metrics translate correctly for both standard LLM/TTS flows and complex multimodal STS OpenAI/Gemini events.