Build a Schema-First AI Call QA Pipeline in Python

Free-text call summaries are useful for people, but awkward for software. A dashboard cannot reliably group a paragraph by sentiment, count missing disclosures, or identify why callers escalated without another parsing step.

This tutorial builds a schema-first AI call QA pipeline in Python. Each completed conversation becomes a set of typed records for sentiment, compliance, and escalation. Those records can move directly into a scorecard because the expected fields, enums, and required values are declared before analysis begins.

The complete sample is available in the Telnyx Code Examples repository.

What the pipeline does

The example has four layers:

JSON-schema insight templates
        |
        v
Insight group: qa-scorecard-v1
        |
        v
AI Assistant insight_settings
        |
        v
Conversation insights -> validation -> QA scorecard

It creates three versioned insight templates:

• sentiment-v1 returns positivity, frustration, overall sentiment, and key emotions.
• compliance-v1 returns an identity-verification result and a constrained list of compliance flags.
• escalation-v1 returns a controlled escalation reason and whether the agent handled it well.

The templates are assigned to one insight group. That group is attached to an AI Assistant once, so future assistant conversations can be analyzed against all three contracts.

Why start with JSON Schema?

Without a contract, an analysis model might describe the same issue as "caller asked for a manager," "requested escalation," or "wanted a supervisor." Those phrases may be equivalent to a reader, but they create inconsistent analytics.

The escalation template constrains the result instead:

ESCALATION_SCHEMA = {
    "type": "object",
    "properties": {
        "escalation_reason": {
            "type": "string",
            "enum": [
                "caller_request",
                "compliance_violation",
                "poor_call_quality",
                "system_error",
                "other",
            ],
        },
        "agent_performed_well": {"type": "boolean"},
    },
    "required": ["escalation_reason", "agent_performed_well"],
    "additionalProperties": False,
}

The sample sends json_schema as a JSON string, which is the canonical wire format used by create_templates.py:

payload = {
    "name": "escalation-v1",
    "instructions": "Determine whether the caller escalated and why.",
    "json_schema": json.dumps(ESCALATION_SCHEMA),
}

requests.post(
    f"{API_BASE}/ai/conversations/insights",
    headers=headers,
    json=payload,
    timeout=30,
)

The API also accepts the schema as an object. See the structured insights documentation for the concept and request shape.

Create and verify the insight group

create_group.py finds or creates qa-scorecard-v1, assigns the three templates, and then reads the group back. The script exits if any expected template is missing from the returned membership.

This read-after-write check is important. A successful request only proves that the API accepted it; reading the resource back proves that the configuration the assistant will use is present.

for name in TEMPLATE_NAMES:
    _assign(headers, group_id, templates[name])

membership = _read_membership(headers, group_id)
assigned_names = {item.get("name") for item in membership}

missing = [name for name in TEMPLATE_NAMES if name not in assigned_names]
if missing:
    sys.exit(f"Assignment verification failed - missing {missing}")

Telnyx insight groups let you apply a reusable bundle of analysis templates to an assistant instead of managing each template separately.

Attach the group to an AI Assistant

The assistant is connected to the group through insight_settings:

payload = {
    "insight_settings": {
        "insight_group_id": group_id,
    }
}

requests.post(
    f"{API_BASE}/ai/assistants/{assistant_id}",
    headers=headers,
    json=payload,
    timeout=30,
)

wire_assistant.py follows that update with a GET request and verifies that the returned insight_group_id matches the expected group. If ASSISTANT_ID is not set, the script first creates an assistant using the platform-default model and voice.

Receive completed insights without depending on one delivery path

The sample supports two ways to discover completed analysis:

1. A call.conversation.insights_generated webhook provides a completion signal for real-time workflows.
2. GET /ai/conversations/{conversation_id}/conversations-insights remains the polling source of truth and works even when the webhook receiver is not running.

The Flask receiver verifies Telnyx's Ed25519 signature before accepting an event. TELNYX_PUBLIC_KEY is required, and there is no unverified fallback path. Missing or invalid signature data returns 400.

Polling also makes the lifecycle explicit:

That distinction prevents a slow record from being mislabeled as bad data.

Validate before aggregating

scorecard.py maps each returned template ID back to its template name, parses the result, and validates required fields and enum values. Only valid completed records contribute to the scorecard.

The output aggregates:

• sentiment distribution;
• compliance-flag counts; and
• escalation reasons.

A completed result that violates its schema is printed with its conversation ID and excluded from the totals. You can test this behavior locally with --strict, which injects a synthetic sentiment record missing positivity_score:

python scorecard.py --strict

This gives the pipeline two separate failure categories: records that are not ready yet and records that are complete but structurally invalid.

Run the sample

Clone the repository and install the Python dependencies:

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/insights-pipeline

python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt

cp .env.example .env

Configure .env with your Telnyx API key and placeholders for the assistant and destination numbers:

TELNYX_API_KEY=YOUR_TELNYX_API_KEY
ASSISTANT_PHONE_NUMBER=<E164_ASSISTANT_NUMBER>
DESTINATION_NUMBER=<E164_TEST_NUMBER>
ASSISTANT_ID=<OPTIONAL_ASSISTANT_ID>
PUBLIC_WEBHOOK_BASE_URL=<OPTIONAL_HTTPS_BASE_URL>
TELNYX_PUBLIC_KEY=<REQUIRED_ONLY_FOR_WEBHOOK_RECEIVER>
DEMO_MODE=true

Keep API keys and private credentials in environment variables or a secret manager. Do not place them in source control, prompts, or screenshots.

Run the scripts in order:

python create_templates.py
python create_group.py
python wire_assistant.py
python place_call.py
python scorecard.py --date YYYY-MM-DD

DEMO_MODE=true is the default. In that mode, place_call.py prints the scheduled-event payload but does not dial a real number or write a fake conversation ID. Set DEMO_MODE=false only when you are ready to place the configured test call.

The webhook receiver is optional because the polling path can complete the workflow without it:

python webhook_receiver.py

For a manual scorecard run, you can bypass the local conversation store:

python scorecard.py --conversation-id <CONVERSATION_ID>

Where to take it next

The schema-first pattern extends beyond support QA. Add templates for appointment outcomes, sales qualification, cancellation reasons, accessibility needs, or any other result that downstream software must count and filter reliably.

For production, replace the local conversations.json ledger with durable storage, send invalid records to an operational review queue, and persist the template version beside each result. The central contract remains the same: define the record first, analyze the conversation second, and aggregate only data that passes validation.

Resources

• Source code
• Structured data insights
• Insight groups
• Create an insight template
• Get insights for a conversation
• call.conversation.insights_generated webhook