Module livekit.plugins.posthog.tools
Classes
class PostHogDualChannelTools (gateway_url: str = 'http://localhost:8080',
room_name: str = 'prism-demo',
livekit_room: Any | None = None)-
Expand source code
class PostHogDualChannelTools: """ Dual-Channel Tool suite connecting LiveKit Voice Agents with PostHog's analytical engine. Splits analytical query responses into: 1. High-throughput WebRTC data-channel visualization packet. 2. Concise, speech-ready delta (< 30 tokens) for conversational LLM context. """ def __init__( self, gateway_url: str = DEFAULT_GATEWAY_URL, room_name: str = "prism-demo", livekit_room: Optional[Any] = None, ): self.gateway_url = gateway_url.rstrip("/") self.room_name = room_name self.livekit_room = livekit_room self.http_client = httpx.Client(timeout=3.0) # Wrap tools with LiveKit function_tool self.query_funnel_tool = function_tool(self._query_funnel_impl) self.query_errors_tool = function_tool(self._query_errors_impl) self.query_retention_tool = function_tool(self._query_retention_impl) self.query_hogql_tool = function_tool(self._query_hogql_impl) def _query_funnel_impl(self, funnel_id: str = "checkout_v2") -> str: """ Query PostHog user conversion funnel analytics by funnel ID (e.g., checkout_v2 or signup). Visual chart data is pushed over the WebRTC Data Channel directly to the user's browser, and an ultra-compact speech-ready string (< 30 tokens) is returned to the agent context. """ payload = { "room_name": self.room_name, "session_id": f"session_{int(time.time())}", "tool": "query_funnel", "args": {"funnel_id": funnel_id}, } try: resp = self.http_client.post(f"{self.gateway_url}/api/tools/funnel", json=payload) if resp.status_code == 200: data = resp.json() return data.get( "speech_ready_delta", "The checkout funnel has a 22.5 percent conversion rate. I've projected the conversion funnel onto your screen.", ) except Exception as e: logger.warning("[Prism Tools] Gateway unreachable (%s). Using local fallback.", e) # Local fallback if Go Gateway is offline mock_card = { "type": "funnel_viz", "title": "Checkout Conversion Dropoff", "funnel_id": funnel_id, "steps": [ {"name": "Cart Viewed", "visitors": 14200, "conversion_rate": 100.0}, {"name": "Billing Entered", "visitors": 9100, "conversion_rate": 64.1}, {"name": "Payment Confirmed", "visitors": 3200, "conversion_rate": 22.5}, ], "primary_drop_step": "Payment Confirmed", "timestamp": int(time.time()), } self._publish_webrtc_card(mock_card) return "The checkout funnel has a 22.5 percent conversion rate. The main drop-off is at the payment confirmation step. I've projected the conversion funnel onto your screen." def _query_errors_impl(self, timeframe: str = "24h") -> str: """ Query PostHog API error waterfall and exception logs over a timeframe (e.g., 24h, 7d). Renders an interactive Error Waterfall card over WebRTC and returns a conversational speech delta. """ payload = { "room_name": self.room_name, "session_id": f"session_{int(time.time())}", "tool": "query_errors", "args": {"timeframe": timeframe}, } try: resp = self.http_client.post(f"{self.gateway_url}/api/tools/execute", json=payload) if resp.status_code == 200: return resp.json().get( "speech_ready_delta", "We detected 42 server errors in the last 24 hours. The highest failure rate is Payment Gateway Timeouts at 52 percent.", ) except Exception as e: logger.warning("[Prism Tools] Gateway unreachable (%s). Using local fallback.", e) mock_card = { "type": "error_waterfall", "title": f"API Error Waterfall ({timeframe})", "timeframe": timeframe, "total_errors": 42, "primary_root_cause": "PaymentGatewayTimeout (504)", "errors": [ {"error_type": "PaymentGatewayTimeout (504)", "count": 22, "percentage": 52.4, "status_code": 504}, {"error_type": "StripeCardDeclined (400)", "count": 13, "percentage": 30.9, "status_code": 400}, {"error_type": "DatabaseDeadlock (500)", "count": 7, "percentage": 16.7, "status_code": 500}, ], "timestamp": int(time.time()), } self._publish_webrtc_card(mock_card) return "We detected 42 server errors in the last 24 hours. The highest failure rate is Payment Gateway Timeouts at 52 percent. The error waterfall is on your screen." def _query_retention_impl(self, cohort: str = "weekly") -> str: """ Query PostHog cohort retention trends over an 8-week horizon. Renders a Retention Trend line chart over WebRTC and returns a conversational speech delta. """ payload = { "room_name": self.room_name, "session_id": f"session_{int(time.time())}", "tool": "query_retention", "args": {"cohort": cohort}, } try: resp = self.http_client.post(f"{self.gateway_url}/api/tools/execute", json=payload) if resp.status_code == 200: return resp.json().get( "speech_ready_delta", "Your eight-week user retention stands at 19.5 percent, beating the industry average of 15 percent.", ) except Exception as e: logger.warning("[Prism Tools] Gateway unreachable (%s). Using local fallback.", e) mock_card = { "type": "retention_trend", "title": "8-Week Cohort Retention Trend", "cohort_name": "Weekly Active Users", "terminal_retention": 19.5, "intervals": [ {"interval": "Week 1", "retention_rate": 100.0, "users": 8400}, {"interval": "Week 2", "retention_rate": 44.2, "users": 3712}, {"interval": "Week 3", "retention_rate": 34.0, "users": 2856}, {"interval": "Week 4", "retention_rate": 28.1, "users": 2360}, {"interval": "Week 6", "retention_rate": 23.4, "users": 1965}, {"interval": "Week 8", "retention_rate": 19.5, "users": 1638}, ], "timestamp": int(time.time()), } self._publish_webrtc_card(mock_card) return "Your eight-week user retention stands at 19.5 percent, beating the industry average of 15 percent. The retention curve is displayed on your screen." def _query_hogql_impl(self, query: str) -> str: """ Execute an analytical HogQL query against PostHog and ClickHouse. Renders the analytical results card over WebRTC and returns a concise verbal summary. """ payload = { "room_name": self.room_name, "session_id": f"session_{int(time.time())}", "tool": "query_hogql", "args": {"query": query}, } try: resp = self.http_client.post(f"{self.gateway_url}/api/tools/hogql", json=payload) if resp.status_code == 200: return resp.json().get( "speech_ready_delta", "Your HogQL query completed successfully. I've projected the analytical results onto your screen.", ) except Exception as e: logger.warning("[Prism Tools] Gateway unreachable (%s). Using local fallback.", e) mock_card = { "type": "hogql_viz", "title": "HogQL Analytical Query Result", "query": query, "columns": ["event", "count()"], "rows": [ ["$pageview", 48200], ["user_signed_up", 1240], ["checkout_step_completed", 890], ], "timestamp": int(time.time()), } self._publish_webrtc_card(mock_card) return "Your HogQL query completed successfully across 48,000 pageviews and 1,240 signups. I've projected the analytical results onto your screen." def _publish_webrtc_card(self, card_dict: dict): """Broadcasts data card locally if LiveKit room participant is available.""" if self.livekit_room and hasattr(self.livekit_room, "local_participant"): try: card_bytes = json.dumps(card_dict).encode("utf-8") import asyncio loop = asyncio.get_event_loop() if loop.is_running(): asyncio.create_task( self.livekit_room.local_participant.publish_data( card_bytes, reliable=True, topic="posthog-viz-card", ) ) except Exception as ex: logger.error("[Prism Tools] Failed to broadcast local data packet: %s", ex) def get_tools(self) -> List[Any]: """Returns all LiveKit FunctionTools provided by this suite.""" return [ self.query_funnel_tool, self.query_errors_tool, self.query_retention_tool, self.query_hogql_tool, ] @property def query_funnel(self): return self.query_funnel_tool @property def query_errors(self): return self.query_errors_tool @property def query_retention(self): return self.query_retention_tool @property def query_hogql(self): return self.query_hogql_toolDual-Channel Tool suite connecting LiveKit Voice Agents with PostHog's analytical engine. Splits analytical query responses into: 1. High-throughput WebRTC data-channel visualization packet. 2. Concise, speech-ready delta (< 30 tokens) for conversational LLM context.
Instance variables
prop query_errors-
Expand source code
@property def query_errors(self): return self.query_errors_tool prop query_funnel-
Expand source code
@property def query_funnel(self): return self.query_funnel_tool prop query_hogql-
Expand source code
@property def query_hogql(self): return self.query_hogql_tool prop query_retention-
Expand source code
@property def query_retention(self): return self.query_retention_tool
Methods
def get_tools(self) ‑> List[Any]-
Expand source code
def get_tools(self) -> List[Any]: """Returns all LiveKit FunctionTools provided by this suite.""" return [ self.query_funnel_tool, self.query_errors_tool, self.query_retention_tool, self.query_hogql_tool, ]Returns all LiveKit FunctionTools provided by this suite.
class Tools (gateway_url: str = 'http://localhost:8080',
room_name: str = 'prism-demo',
livekit_room: Any | None = None)-
Expand source code
class PostHogDualChannelTools: """ Dual-Channel Tool suite connecting LiveKit Voice Agents with PostHog's analytical engine. Splits analytical query responses into: 1. High-throughput WebRTC data-channel visualization packet. 2. Concise, speech-ready delta (< 30 tokens) for conversational LLM context. """ def __init__( self, gateway_url: str = DEFAULT_GATEWAY_URL, room_name: str = "prism-demo", livekit_room: Optional[Any] = None, ): self.gateway_url = gateway_url.rstrip("/") self.room_name = room_name self.livekit_room = livekit_room self.http_client = httpx.Client(timeout=3.0) # Wrap tools with LiveKit function_tool self.query_funnel_tool = function_tool(self._query_funnel_impl) self.query_errors_tool = function_tool(self._query_errors_impl) self.query_retention_tool = function_tool(self._query_retention_impl) self.query_hogql_tool = function_tool(self._query_hogql_impl) def _query_funnel_impl(self, funnel_id: str = "checkout_v2") -> str: """ Query PostHog user conversion funnel analytics by funnel ID (e.g., checkout_v2 or signup). Visual chart data is pushed over the WebRTC Data Channel directly to the user's browser, and an ultra-compact speech-ready string (< 30 tokens) is returned to the agent context. """ payload = { "room_name": self.room_name, "session_id": f"session_{int(time.time())}", "tool": "query_funnel", "args": {"funnel_id": funnel_id}, } try: resp = self.http_client.post(f"{self.gateway_url}/api/tools/funnel", json=payload) if resp.status_code == 200: data = resp.json() return data.get( "speech_ready_delta", "The checkout funnel has a 22.5 percent conversion rate. I've projected the conversion funnel onto your screen.", ) except Exception as e: logger.warning("[Prism Tools] Gateway unreachable (%s). Using local fallback.", e) # Local fallback if Go Gateway is offline mock_card = { "type": "funnel_viz", "title": "Checkout Conversion Dropoff", "funnel_id": funnel_id, "steps": [ {"name": "Cart Viewed", "visitors": 14200, "conversion_rate": 100.0}, {"name": "Billing Entered", "visitors": 9100, "conversion_rate": 64.1}, {"name": "Payment Confirmed", "visitors": 3200, "conversion_rate": 22.5}, ], "primary_drop_step": "Payment Confirmed", "timestamp": int(time.time()), } self._publish_webrtc_card(mock_card) return "The checkout funnel has a 22.5 percent conversion rate. The main drop-off is at the payment confirmation step. I've projected the conversion funnel onto your screen." def _query_errors_impl(self, timeframe: str = "24h") -> str: """ Query PostHog API error waterfall and exception logs over a timeframe (e.g., 24h, 7d). Renders an interactive Error Waterfall card over WebRTC and returns a conversational speech delta. """ payload = { "room_name": self.room_name, "session_id": f"session_{int(time.time())}", "tool": "query_errors", "args": {"timeframe": timeframe}, } try: resp = self.http_client.post(f"{self.gateway_url}/api/tools/execute", json=payload) if resp.status_code == 200: return resp.json().get( "speech_ready_delta", "We detected 42 server errors in the last 24 hours. The highest failure rate is Payment Gateway Timeouts at 52 percent.", ) except Exception as e: logger.warning("[Prism Tools] Gateway unreachable (%s). Using local fallback.", e) mock_card = { "type": "error_waterfall", "title": f"API Error Waterfall ({timeframe})", "timeframe": timeframe, "total_errors": 42, "primary_root_cause": "PaymentGatewayTimeout (504)", "errors": [ {"error_type": "PaymentGatewayTimeout (504)", "count": 22, "percentage": 52.4, "status_code": 504}, {"error_type": "StripeCardDeclined (400)", "count": 13, "percentage": 30.9, "status_code": 400}, {"error_type": "DatabaseDeadlock (500)", "count": 7, "percentage": 16.7, "status_code": 500}, ], "timestamp": int(time.time()), } self._publish_webrtc_card(mock_card) return "We detected 42 server errors in the last 24 hours. The highest failure rate is Payment Gateway Timeouts at 52 percent. The error waterfall is on your screen." def _query_retention_impl(self, cohort: str = "weekly") -> str: """ Query PostHog cohort retention trends over an 8-week horizon. Renders a Retention Trend line chart over WebRTC and returns a conversational speech delta. """ payload = { "room_name": self.room_name, "session_id": f"session_{int(time.time())}", "tool": "query_retention", "args": {"cohort": cohort}, } try: resp = self.http_client.post(f"{self.gateway_url}/api/tools/execute", json=payload) if resp.status_code == 200: return resp.json().get( "speech_ready_delta", "Your eight-week user retention stands at 19.5 percent, beating the industry average of 15 percent.", ) except Exception as e: logger.warning("[Prism Tools] Gateway unreachable (%s). Using local fallback.", e) mock_card = { "type": "retention_trend", "title": "8-Week Cohort Retention Trend", "cohort_name": "Weekly Active Users", "terminal_retention": 19.5, "intervals": [ {"interval": "Week 1", "retention_rate": 100.0, "users": 8400}, {"interval": "Week 2", "retention_rate": 44.2, "users": 3712}, {"interval": "Week 3", "retention_rate": 34.0, "users": 2856}, {"interval": "Week 4", "retention_rate": 28.1, "users": 2360}, {"interval": "Week 6", "retention_rate": 23.4, "users": 1965}, {"interval": "Week 8", "retention_rate": 19.5, "users": 1638}, ], "timestamp": int(time.time()), } self._publish_webrtc_card(mock_card) return "Your eight-week user retention stands at 19.5 percent, beating the industry average of 15 percent. The retention curve is displayed on your screen." def _query_hogql_impl(self, query: str) -> str: """ Execute an analytical HogQL query against PostHog and ClickHouse. Renders the analytical results card over WebRTC and returns a concise verbal summary. """ payload = { "room_name": self.room_name, "session_id": f"session_{int(time.time())}", "tool": "query_hogql", "args": {"query": query}, } try: resp = self.http_client.post(f"{self.gateway_url}/api/tools/hogql", json=payload) if resp.status_code == 200: return resp.json().get( "speech_ready_delta", "Your HogQL query completed successfully. I've projected the analytical results onto your screen.", ) except Exception as e: logger.warning("[Prism Tools] Gateway unreachable (%s). Using local fallback.", e) mock_card = { "type": "hogql_viz", "title": "HogQL Analytical Query Result", "query": query, "columns": ["event", "count()"], "rows": [ ["$pageview", 48200], ["user_signed_up", 1240], ["checkout_step_completed", 890], ], "timestamp": int(time.time()), } self._publish_webrtc_card(mock_card) return "Your HogQL query completed successfully across 48,000 pageviews and 1,240 signups. I've projected the analytical results onto your screen." def _publish_webrtc_card(self, card_dict: dict): """Broadcasts data card locally if LiveKit room participant is available.""" if self.livekit_room and hasattr(self.livekit_room, "local_participant"): try: card_bytes = json.dumps(card_dict).encode("utf-8") import asyncio loop = asyncio.get_event_loop() if loop.is_running(): asyncio.create_task( self.livekit_room.local_participant.publish_data( card_bytes, reliable=True, topic="posthog-viz-card", ) ) except Exception as ex: logger.error("[Prism Tools] Failed to broadcast local data packet: %s", ex) def get_tools(self) -> List[Any]: """Returns all LiveKit FunctionTools provided by this suite.""" return [ self.query_funnel_tool, self.query_errors_tool, self.query_retention_tool, self.query_hogql_tool, ] @property def query_funnel(self): return self.query_funnel_tool @property def query_errors(self): return self.query_errors_tool @property def query_retention(self): return self.query_retention_tool @property def query_hogql(self): return self.query_hogql_toolDual-Channel Tool suite connecting LiveKit Voice Agents with PostHog's analytical engine. Splits analytical query responses into: 1. High-throughput WebRTC data-channel visualization packet. 2. Concise, speech-ready delta (< 30 tokens) for conversational LLM context.
Instance variables
prop query_errors-
Expand source code
@property def query_errors(self): return self.query_errors_tool prop query_funnel-
Expand source code
@property def query_funnel(self): return self.query_funnel_tool prop query_hogql-
Expand source code
@property def query_hogql(self): return self.query_hogql_tool prop query_retention-
Expand source code
@property def query_retention(self): return self.query_retention_tool
Methods
def get_tools(self) ‑> List[Any]-
Expand source code
def get_tools(self) -> List[Any]: """Returns all LiveKit FunctionTools provided by this suite.""" return [ self.query_funnel_tool, self.query_errors_tool, self.query_retention_tool, self.query_hogql_tool, ]Returns all LiveKit FunctionTools provided by this suite.
class DualChannelTools (gateway_url: str = 'http://localhost:8080',
room_name: str = 'prism-demo',
livekit_room: Any | None = None)-
Expand source code
class PostHogDualChannelTools: """ Dual-Channel Tool suite connecting LiveKit Voice Agents with PostHog's analytical engine. Splits analytical query responses into: 1. High-throughput WebRTC data-channel visualization packet. 2. Concise, speech-ready delta (< 30 tokens) for conversational LLM context. """ def __init__( self, gateway_url: str = DEFAULT_GATEWAY_URL, room_name: str = "prism-demo", livekit_room: Optional[Any] = None, ): self.gateway_url = gateway_url.rstrip("/") self.room_name = room_name self.livekit_room = livekit_room self.http_client = httpx.Client(timeout=3.0) # Wrap tools with LiveKit function_tool self.query_funnel_tool = function_tool(self._query_funnel_impl) self.query_errors_tool = function_tool(self._query_errors_impl) self.query_retention_tool = function_tool(self._query_retention_impl) self.query_hogql_tool = function_tool(self._query_hogql_impl) def _query_funnel_impl(self, funnel_id: str = "checkout_v2") -> str: """ Query PostHog user conversion funnel analytics by funnel ID (e.g., checkout_v2 or signup). Visual chart data is pushed over the WebRTC Data Channel directly to the user's browser, and an ultra-compact speech-ready string (< 30 tokens) is returned to the agent context. """ payload = { "room_name": self.room_name, "session_id": f"session_{int(time.time())}", "tool": "query_funnel", "args": {"funnel_id": funnel_id}, } try: resp = self.http_client.post(f"{self.gateway_url}/api/tools/funnel", json=payload) if resp.status_code == 200: data = resp.json() return data.get( "speech_ready_delta", "The checkout funnel has a 22.5 percent conversion rate. I've projected the conversion funnel onto your screen.", ) except Exception as e: logger.warning("[Prism Tools] Gateway unreachable (%s). Using local fallback.", e) # Local fallback if Go Gateway is offline mock_card = { "type": "funnel_viz", "title": "Checkout Conversion Dropoff", "funnel_id": funnel_id, "steps": [ {"name": "Cart Viewed", "visitors": 14200, "conversion_rate": 100.0}, {"name": "Billing Entered", "visitors": 9100, "conversion_rate": 64.1}, {"name": "Payment Confirmed", "visitors": 3200, "conversion_rate": 22.5}, ], "primary_drop_step": "Payment Confirmed", "timestamp": int(time.time()), } self._publish_webrtc_card(mock_card) return "The checkout funnel has a 22.5 percent conversion rate. The main drop-off is at the payment confirmation step. I've projected the conversion funnel onto your screen." def _query_errors_impl(self, timeframe: str = "24h") -> str: """ Query PostHog API error waterfall and exception logs over a timeframe (e.g., 24h, 7d). Renders an interactive Error Waterfall card over WebRTC and returns a conversational speech delta. """ payload = { "room_name": self.room_name, "session_id": f"session_{int(time.time())}", "tool": "query_errors", "args": {"timeframe": timeframe}, } try: resp = self.http_client.post(f"{self.gateway_url}/api/tools/execute", json=payload) if resp.status_code == 200: return resp.json().get( "speech_ready_delta", "We detected 42 server errors in the last 24 hours. The highest failure rate is Payment Gateway Timeouts at 52 percent.", ) except Exception as e: logger.warning("[Prism Tools] Gateway unreachable (%s). Using local fallback.", e) mock_card = { "type": "error_waterfall", "title": f"API Error Waterfall ({timeframe})", "timeframe": timeframe, "total_errors": 42, "primary_root_cause": "PaymentGatewayTimeout (504)", "errors": [ {"error_type": "PaymentGatewayTimeout (504)", "count": 22, "percentage": 52.4, "status_code": 504}, {"error_type": "StripeCardDeclined (400)", "count": 13, "percentage": 30.9, "status_code": 400}, {"error_type": "DatabaseDeadlock (500)", "count": 7, "percentage": 16.7, "status_code": 500}, ], "timestamp": int(time.time()), } self._publish_webrtc_card(mock_card) return "We detected 42 server errors in the last 24 hours. The highest failure rate is Payment Gateway Timeouts at 52 percent. The error waterfall is on your screen." def _query_retention_impl(self, cohort: str = "weekly") -> str: """ Query PostHog cohort retention trends over an 8-week horizon. Renders a Retention Trend line chart over WebRTC and returns a conversational speech delta. """ payload = { "room_name": self.room_name, "session_id": f"session_{int(time.time())}", "tool": "query_retention", "args": {"cohort": cohort}, } try: resp = self.http_client.post(f"{self.gateway_url}/api/tools/execute", json=payload) if resp.status_code == 200: return resp.json().get( "speech_ready_delta", "Your eight-week user retention stands at 19.5 percent, beating the industry average of 15 percent.", ) except Exception as e: logger.warning("[Prism Tools] Gateway unreachable (%s). Using local fallback.", e) mock_card = { "type": "retention_trend", "title": "8-Week Cohort Retention Trend", "cohort_name": "Weekly Active Users", "terminal_retention": 19.5, "intervals": [ {"interval": "Week 1", "retention_rate": 100.0, "users": 8400}, {"interval": "Week 2", "retention_rate": 44.2, "users": 3712}, {"interval": "Week 3", "retention_rate": 34.0, "users": 2856}, {"interval": "Week 4", "retention_rate": 28.1, "users": 2360}, {"interval": "Week 6", "retention_rate": 23.4, "users": 1965}, {"interval": "Week 8", "retention_rate": 19.5, "users": 1638}, ], "timestamp": int(time.time()), } self._publish_webrtc_card(mock_card) return "Your eight-week user retention stands at 19.5 percent, beating the industry average of 15 percent. The retention curve is displayed on your screen." def _query_hogql_impl(self, query: str) -> str: """ Execute an analytical HogQL query against PostHog and ClickHouse. Renders the analytical results card over WebRTC and returns a concise verbal summary. """ payload = { "room_name": self.room_name, "session_id": f"session_{int(time.time())}", "tool": "query_hogql", "args": {"query": query}, } try: resp = self.http_client.post(f"{self.gateway_url}/api/tools/hogql", json=payload) if resp.status_code == 200: return resp.json().get( "speech_ready_delta", "Your HogQL query completed successfully. I've projected the analytical results onto your screen.", ) except Exception as e: logger.warning("[Prism Tools] Gateway unreachable (%s). Using local fallback.", e) mock_card = { "type": "hogql_viz", "title": "HogQL Analytical Query Result", "query": query, "columns": ["event", "count()"], "rows": [ ["$pageview", 48200], ["user_signed_up", 1240], ["checkout_step_completed", 890], ], "timestamp": int(time.time()), } self._publish_webrtc_card(mock_card) return "Your HogQL query completed successfully across 48,000 pageviews and 1,240 signups. I've projected the analytical results onto your screen." def _publish_webrtc_card(self, card_dict: dict): """Broadcasts data card locally if LiveKit room participant is available.""" if self.livekit_room and hasattr(self.livekit_room, "local_participant"): try: card_bytes = json.dumps(card_dict).encode("utf-8") import asyncio loop = asyncio.get_event_loop() if loop.is_running(): asyncio.create_task( self.livekit_room.local_participant.publish_data( card_bytes, reliable=True, topic="posthog-viz-card", ) ) except Exception as ex: logger.error("[Prism Tools] Failed to broadcast local data packet: %s", ex) def get_tools(self) -> List[Any]: """Returns all LiveKit FunctionTools provided by this suite.""" return [ self.query_funnel_tool, self.query_errors_tool, self.query_retention_tool, self.query_hogql_tool, ] @property def query_funnel(self): return self.query_funnel_tool @property def query_errors(self): return self.query_errors_tool @property def query_retention(self): return self.query_retention_tool @property def query_hogql(self): return self.query_hogql_toolDual-Channel Tool suite connecting LiveKit Voice Agents with PostHog's analytical engine. Splits analytical query responses into: 1. High-throughput WebRTC data-channel visualization packet. 2. Concise, speech-ready delta (< 30 tokens) for conversational LLM context.
Instance variables
prop query_errors-
Expand source code
@property def query_errors(self): return self.query_errors_tool prop query_funnel-
Expand source code
@property def query_funnel(self): return self.query_funnel_tool prop query_hogql-
Expand source code
@property def query_hogql(self): return self.query_hogql_tool prop query_retention-
Expand source code
@property def query_retention(self): return self.query_retention_tool
Methods
def get_tools(self) ‑> List[Any]-
Expand source code
def get_tools(self) -> List[Any]: """Returns all LiveKit FunctionTools provided by this suite.""" return [ self.query_funnel_tool, self.query_errors_tool, self.query_retention_tool, self.query_hogql_tool, ]Returns all LiveKit FunctionTools provided by this suite.