# Build a Schema-First AI Call QA Pipeline in Python > Source: > Published: 2026-10-21 20:26:29+00:00 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](https://github.com/team-telnyx/telnyx-code-examples/tree/main/insights-pipeline). 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. 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](https://developers.telnyx.com/docs/inference/ai-insights/structured-insights) 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}") ``` Telnyx [insight groups](https://developers.telnyx.com/docs/inference/ai-insights/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, ) ``` `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: `call.conversation.insights_generated` webhook provides a completion signal for real-time workflows.`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: | Status | Pipeline behavior | |---|---| | `pending` | Count the record as pending. | | `in_progress` | Keep waiting; analysis is not complete. | | `completed` | Validate the result against its declared schema. | | `failed` | Report the failed record and stop the live-call workflow with a nonzero exit. | 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. The output aggregates: 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. 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= DESTINATION_NUMBER= ASSISTANT_ID= PUBLIC_WEBHOOK_BASE_URL= TELNYX_PUBLIC_KEY= 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 ``` 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.