Add Google Dialogflow CX plugin for LiveKit Agents

[Speediing]

Summary

This PR introduces a new LiveKit Agents plugin that integrates Google Dialogflow CX as an LLM provider. Dialogflow CX is an intent-based conversational AI engine where all conversation logic is configured in the Dialogflow CX console rather than through agent instructions.

Key Changes

• New plugin package: livekit-plugins-dialogflow with full LLM integration

  • LLM class implementing the LiveKit llm.LLM interface for Dialogflow CX
  • DialogflowLLMStream class handling request/response streaming
  • Support for environment variables (GOOGLE_CLOUD_PROJECT, GOOGLE_CLOUD_LOCATION) with explicit parameter overrides
  • Configurable language code, environment ID, and session TTL

• Core functionality:

  • Detects intent and retrieves responses from Dialogflow CX agents
  • Extracts response text from multiple message types
  • Collects custom payloads (cards, suggestions, etc.) in response metadata
  • Builds proper Dialogflow session paths with optional environment IDs
  • Comprehensive error handling mapping Google API exceptions to LiveKit exception types

• Session management:

  • Supports custom session IDs via extra_kwargs for multi-turn conversations
  • Generates UUIDs for new sessions when not explicitly provided
  • Caches session IDs for conversation continuity

• Comprehensive test suite (416 lines):

  • Tests for LLM instantiation, environment variable fallbacks, and configuration validation
  • Tests for chat stream creation and message passing
  • Tests for Dialogflow API integration with mocked responses
  • Tests for error handling (rate limiting, service unavailability, invalid arguments)
  • Tests for session path construction with/without environment IDs
  • Tests for language code and custom payload handling

• Documentation and examples:

  • README with setup instructions, authentication requirements, and usage examples
  • Example voice agent (dialogflow_agent.py) demonstrating STT → Dialogflow → TTS pipeline
  • Inline code documentation explaining Dialogflow-specific behavior

Notable Implementation Details

• Non-streaming responses: Dialogflow CX returns full responses in one shot, not token-by-token. The entire response is emitted as a single ChatChunk.
• No token counts: Dialogflow does not provide token usage metrics; usage fields are set to 0.
• Google Cloud authentication: Uses Application Default Credentials (ADC) via google-cloud-dialogflow-cx SDK.
• API endpoint selection: Automatically selects regional or global Dialogflow endpoints based on location parameter.
• Metadata extraction: Captures intent detection confidence, matched intent, current page, and match type for observability.

https://claude.ai/code/session_01HTkStArHmt7uy1zX51pEvp

[CLAassistant]

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our Contributor License Agreement before we can accept your contribution.
_{You have signed the CLA already but the status is still pending? Let us recheck it.}

[devin-ai-integration]

Devin Review found 2 potential issues.

View 7 additional findings in Devin Review.

[devin-ai-integration]

🔴 aclose() is a no-op, leaking the gRPC client/channel

The LLM.aclose() method at line 150-151 does nothing (pass), but the SessionsAsyncClient created at line 113 opens a gRPC channel that should be closed when the LLM is no longer needed.

Resource Leak Details

The SessionsAsyncClient from google-cloud-dialogflow-cx establishes a gRPC transport/channel on creation. Without closing it, the channel remains open indefinitely. In a long-running AgentServer, if LLM instances are created and discarded (e.g., per-session agents as shown in the example at examples/voice_agents/dialogflow_agent.py:48), each will leak a gRPC channel.

The base class llm.LLM defines aclose() at livekit-agents/livekit/agents/llm/llm.py:144 and it is called via __aexit__ at line 155. The Dialogflow plugin should close self._client in its override.

Impact: gRPC channel and associated resources (file descriptors, memory) are never released, which can cause resource exhaustion in long-running processes.

Suggested change

	async def aclose(self) -> None:
	pass
	async def aclose(self) -> None:
	await self._client.transport.close()

---

Was this helpful? React with 👍 or 👎 to provide feedback.

[devin-ai-integration]

🟡 NotFound and PermissionDenied exceptions mapped to wrong HTTP status code 400

NotFound and PermissionDenied exceptions are caught together with InvalidArgument and all mapped to status_code=400. NotFound should be 404 and PermissionDenied should be 403.

Incorrect Status Code Mapping

At lines 223-230, the code catches (InvalidArgument, NotFound, PermissionDenied) as a group and assigns status_code=400 to all of them:

except (InvalidArgument, NotFound, PermissionDenied) as e:
    raise APIStatusError(
        f"dialogflow: {type(e).__name__}",
        status_code=400,  # wrong for NotFound (404) and PermissionDenied (403)
        ...
    )

While retryable=False is explicitly set (so retryability is unaffected), the status code is used in APIStatusError.__str__ (at livekit-agents/livekit/agents/_exceptions.py:74-81) and exposed for observability/logging. Incorrect status codes make debugging harder — a NotFound error appearing as a 400 obscures whether the agent ID is wrong vs. the request payload being invalid.

Impact: Misleading diagnostic information in logs and error messages; developers cannot distinguish between a bad request and a missing resource from the status code alone.

Suggested change

	except (InvalidArgument, NotFound, PermissionDenied) as e:
	raise APIStatusError(
	f"dialogflow: {type(e).__name__}",
	status_code=400,
	body=str(e),
	request_id=request_id,
	retryable=False,
	) from e
	except InvalidArgument as e:
	raise APIStatusError(
	f"dialogflow: {type(e).__name__}",
	status_code=400,
	body=str(e),
	request_id=request_id,
	retryable=False,
	) from e
	except NotFound as e:
	raise APIStatusError(
	f"dialogflow: {type(e).__name__}",
	status_code=404,
	body=str(e),
	request_id=request_id,
	retryable=False,
	) from e
	except PermissionDenied as e:
	raise APIStatusError(
	f"dialogflow: {type(e).__name__}",
	status_code=403,
	body=str(e),
	request_id=request_id,
	retryable=False,
	) from e

---

Was this helpful? React with 👍 or 👎 to provide feedback.