Add engagement detection link implementation

- Introduced a new link for analyzing customer-agent engagement in conversations using OpenAI's GPT-4.1 model.
- Implemented functionality to analyze transcripts, store results as analyses and tags, and included retry logic and error handling.
- Added comprehensive tests to ensure functionality and reliability.
- Updated README.md with configuration options, features, and metrics for the new link.

Author: howethomas

server/links/detect_engagement/README.md

@@ -0,0 +1,58 @@
+# Engagement Detection Link
+
+This link analyzes conversations to determine if both the customer and agent are engaged in the dialogue. It uses OpenAI's GPT-4.1 model to analyze transcripts and determine engagement status.
+
+## Features
+
+- Analyzes each dialog in a vCon to detect engagement
+- Uses GPT-4.1 for accurate conversation analysis
+- Stores results both as analysis and tags
+- Includes retry logic and error handling
+- Provides metrics for monitoring
+
+## Configuration Options
+
+The link can be configured with the following options:
+
+```python
+default_options = {
+    "prompt": "Did both the customer and the agent speak? Respond with 'true' if yes, 'false' if not. Respond with only 'true' or 'false'.",
+    "analysis_type": "engagement_analysis",
+    "model": "gpt-4.1",
+    "sampling_rate": 1,
+    "temperature": 0.2,
+    "source": {
+        "analysis_type": "transcript",
+        "text_location": "body.paragraphs.transcript",
+    }
+}
+```
+
+### Options Description
+
+- `prompt`: The prompt used to analyze engagement
+- `analysis_type`: The type of analysis to store in the vCon
+- `model`: The OpenAI model to use (default: gpt-4.1)
+- `sampling_rate`: Rate at which to sample vCons for analysis
+- `temperature`: Model temperature for response generation
+- `source`: Configuration for where to find the transcript data
+
+## Output
+
+The link adds two types of data to the vCon:
+
+1. Analysis: Stores the engagement status as an analysis object
+2. Tags: Adds an "engagement" tag with the boolean result
+
+## Metrics
+
+The link provides the following metrics:
+
+- `conserver.link.openai.engagement_detected`: Gauge for engagement status
+- `conserver.link.openai.engagement_analysis_time`: Time taken for analysis
+- `conserver.link.openai.engagement_analysis_failures`: Count of analysis failures
+
+## Requirements
+
+- OpenAI API key must be set in the environment
+- vCon must contain transcript data 

server/links/detect_engagement/__init__.py

@@ -0,0 +1,189 @@
+from lib.vcon_redis import VconRedis
+from lib.logging_utils import init_logger
+import logging
+from openai import OpenAI
+from tenacity import (
+    retry,
+    stop_after_attempt,
+    wait_exponential,
+    before_sleep_log,
+)
+from lib.metrics import init_metrics, stats_gauge, stats_count
+import time
+from lib.links.filters import is_included, randomly_execute_with_sampling
+import os
+
+init_metrics()
+logger = init_logger(__name__)
+
+default_options = {
+    "prompt": "Did both the customer and the agent speak? Respond with 'true' if yes, 'false' if not. Respond with only 'true' or 'false'.",
+    "analysis_type": "engagement_analysis",
+    "model": "gpt-4.1",
+    "sampling_rate": 1,
+    "temperature": 0.2,
+    "source": {
+        "analysis_type": "transcript",
+        "text_location": "body.paragraphs.transcript",
+    },
+    "OPENAI_API_KEY": os.getenv("OPENAI_API_KEY", "")  # Make it optional with empty default
+}
+
+def get_analysis_for_type(vcon, index, analysis_type):
+    for a in vcon.analysis:
+        if a["dialog"] == index and a["type"] == analysis_type:
+            return a
+    return None
+
+@retry(
+    wait=wait_exponential(multiplier=2, min=1, max=65),
+    stop=stop_after_attempt(6),
+    before_sleep=before_sleep_log(logger, logging.INFO),
+)
+def check_engagement(transcript, prompt, model, temperature, client) -> bool:
+    # The new responses API expects a single string or a list of message dicts as 'input'
+    input_text = f"{prompt}\n\nTranscript: {transcript}"
+
+    response = client.responses.create(
+        model=model,
+        input=input_text,
+        temperature=temperature
+    )
+    
+    # The new API returns the result in response.output_text
+    answer = response.output_text.strip().lower()
+    return answer == "true"
+
+def run(
+    vcon_uuid,
+    link_name,
+    opts=default_options,
+):
+    module_name = __name__.split(".")[-1]
+    logger.info(f"Starting {module_name}: {link_name} plugin for: {vcon_uuid}")
+    merged_opts = default_options.copy()
+    merged_opts.update(opts)
+    opts = merged_opts
+
+    # Check for OPENAI_API_KEY in opts or environment
+    openai_key = opts.get("OPENAI_API_KEY") or os.getenv("OPENAI_API_KEY")
+    if not openai_key:
+        logger.warning("OPENAI_API_KEY not defined, skipping analysis for vCon: %s", vcon_uuid)
+        return vcon_uuid
+    opts["OPENAI_API_KEY"] = openai_key
+
+    vcon_redis = VconRedis()
+    vCon = vcon_redis.get_vcon(vcon_uuid)
+
+    if not is_included(opts, vCon):
+        logger.info(f"Skipping {link_name} vCon {vcon_uuid} due to filters")
+        return vcon_uuid
+
+    if not randomly_execute_with_sampling(opts):
+        logger.info(f"Skipping {link_name} vCon {vcon_uuid} due to sampling")
+        return vcon_uuid
+
+    client = OpenAI(api_key=opts["OPENAI_API_KEY"], timeout=120.0, max_retries=0)
+    source_type = opts["source"]["analysis_type"]
+    text_location = opts["source"]["text_location"]
+
+    for index, dialog in enumerate(vCon.dialog):
+        source = get_analysis_for_type(vCon, index, source_type)
+        if not source:
+            logger.warning("No %s found for vCon: %s", source_type, vCon.uuid)
+            continue
+
+        source_text = navigate_dict(source, text_location)
+        if not source_text:
+            logger.warning("No source_text found at %s for vCon: %s", text_location, vCon.uuid)
+            continue
+
+        analysis = get_analysis_for_type(vCon, index, opts["analysis_type"])
+        if analysis:
+            logger.info(
+                "Dialog %s already has a %s in vCon: %s",
+                index,
+                opts["analysis_type"],
+                vCon.uuid,
+            )
+            continue
+
+        logger.info(
+            "Analyzing engagement for dialog %s with options: %s",
+            index,
+            {k: v for k, v in opts.items() if k != "OPENAI_API_KEY"},
+        )
+        start = time.time()
+        try:
+            is_engaged = check_engagement(
+                transcript=source_text,
+                prompt=opts["prompt"],
+                model=opts["model"],
+                temperature=opts["temperature"],
+                client=client
+            )
+
+            # Always use string 'true'/'false' for tag and body
+            is_engaged_str = "true" if is_engaged else "false"
+
+            vendor_schema = {
+                "model": opts["model"],
+                "prompt": opts["prompt"],
+                "is_engaged": is_engaged_str
+            }
+
+            vCon.add_analysis(
+                type=opts["analysis_type"],
+                dialog=index,
+                vendor="openai",
+                body=is_engaged_str,
+                encoding="none",
+                extra={
+                    "vendor_schema": vendor_schema,
+                },
+            )
+
+            vCon.add_tag(tag_name="engagement", tag_value=is_engaged_str)
+            logger.info(f"Applied engagement tag: {is_engaged_str}")
+
+            stats_gauge(
+                "conserver.link.openai.engagement_detected",
+                1 if is_engaged else 0,
+                tags=[f"analysis_type:{opts['analysis_type']}"],
+            )
+
+        except Exception as e:
+            import traceback
+            logger.error(
+                "Failed to generate engagement analysis for vCon %s after multiple retries: %s\nException type: %s\nTraceback:\n%s",
+                vcon_uuid,
+                e,
+                type(e).__name__,
+                traceback.format_exc()
+            )
+            stats_count(
+                "conserver.link.openai.engagement_analysis_failures",
+                tags=[f"analysis_type:{opts['analysis_type']}"],
+            )
+            raise e
+
+        stats_gauge(
+            "conserver.link.openai.engagement_analysis_time",
+            time.time() - start,
+            tags=[f"analysis_type:{opts['analysis_type']}"],
+        )
+
+    vcon_redis.store_vcon(vCon)
+    logger.info(f"Finished detect_engagement - {module_name}:{link_name} plugin for: {vcon_uuid}")
+
+    return vcon_uuid
+
+def navigate_dict(dictionary, path):
+    keys = path.split(".")
+    current = dictionary
+    for key in keys:
+        if key in current:
+            current = current[key]
+        else:
+            return None
+    return current