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.

JSON-schema insight templates | v Insight group: qa-scorecard-v1 | v AI Assistant insight_settings | v Conversation insights -> validation -> QA scorecard Enter fullscreen mode Exit fullscreen mode It creates three versioned insight templates:

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.

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, } Enter fullscreen mode Exit fullscreen mode 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, ) Enter fullscreen mode Exit fullscreen mode The API also accepts the schema as an object. See the structured insights documentation for the concept and request shape.

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 } " ) Enter fullscreen mode Exit fullscreen mode Telnyx insight groups let you apply a reusable bundle of analysis templates to an assistant instead of managing each template separately.

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, ) Enter fullscreen mode Exit fullscreen mode 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.

The sample supports two ways to discover completed analysis:

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.

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.

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 Enter fullscreen mode Exit fullscreen mode This gives the pipeline two separate failure categories: records that are not ready yet and records that are complete but structurally invalid.

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 Enter fullscreen mode Exit fullscreen mode 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 Enter fullscreen mode Exit fullscreen mode Keep API keys and private credentials in environment variables or a secret manager. Do not place them in source control, prompts, or screenshots.

python create_templates.py python create_group.py python wire_assistant.py python place_call.py python scorecard.py --date YYYY-MM-DD Enter fullscreen mode Exit fullscreen mode 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 Enter fullscreen mode Exit fullscreen mode For a manual scorecard run, you can bypass the local conversation store:

python scorecard.py --conversation-id <CONVERSATION_ID> Enter fullscreen mode Exit fullscreen mode 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.