How do I set up a chat completion API for extraction workflows with streaming responses?

To set up a chat completion API for extraction workflows with streaming responses, the usual pattern is:

1. Send structured extraction instructions
2. Use a schema or strict output format
3. Stream tokens/events as the model generates them
4. Incrementally assemble and validate the partial output
5. Finalize into a clean extracted object

1) Define the extraction target

Be explicit about what you want extracted. For example:

• entities: names, dates, amounts, organizations
• fields: invoice number, total, due date
• relations: sender → recipient, product → price

A good prompt says:

• what to extract
• what to ignore
• the required output shape
• how to handle missing values

Example instruction:

Extract the following fields from the document:
- customer_name
- invoice_number
- invoice_date
- total_amount
- currency

Return only valid JSON matching the schema.
If a field is missing, use null.

2) Use a chat completion request

Typical structure:

• system: defines role and output constraints
• user: provides the text/document to extract from
• optional response_format or schema enforcement if supported
• stream: true to receive incremental output

Example request pattern

{
  "model": "your-model",
  "stream": true,
  "messages": [
    {
      "role": "system",
      "content": "You are an information extraction engine. Return only valid JSON."
    },
    {
      "role": "user",
      "content": "Invoice 1042 ... total due $245.80 ..."
    }
  ]
}

3) Prefer a schema-driven output

If your API supports it, use a JSON schema or structured response mode. That makes extraction workflows much more reliable than free-form text.

Example schema:

{
  "type": "object",
  "properties": {
    "customer_name": { "type": ["string", "null"] },
    "invoice_number": { "type": ["string", "null"] },
    "invoice_date": { "type": ["string", "null"] },
    "total_amount": { "type": ["number", "null"] },
    "currency": { "type": ["string", "null"] }
  },
  "required": [
    "customer_name",
    "invoice_number",
    "invoice_date",
    "total_amount",
    "currency"
  ],
  "additionalProperties": false
}

4) Enable streaming

When streaming is enabled, the server sends partial events/chunks rather than one final response.

Why streaming helps in extraction workflows

• faster perceived latency
• early progress display
• easier handling of long documents
• can stop once enough content is received in some cases

5) Handle streamed chunks

You typically:

• append each chunk to a buffer
• detect when the structured payload is complete
• parse/validate JSON
• recover from partial chunks if needed

Example client logic

Pseudo-code:

buffer = ""

for event in stream:
    chunk = event.delta_content
    if chunk:
        buffer += chunk

# after stream ends
data = json.loads(buffer)

If your API emits SSE-style events, you’ll parse each event and extract the content delta.

6) Validate and normalize results

After streaming finishes:

• parse JSON
• validate against schema
• normalize dates, numbers, and currencies
• flag low-confidence or missing fields

Example normalization:

• "245.80" → 245.8
• "07/19/2026" → "2026-07-19"
• "USD" stays "USD"

7) Recommended extraction workflow architecture

A robust setup often looks like this:

Pipeline

1. Input document ingestion
2. Chunking / OCR / text cleanup
3. Chat completion extraction
4. Streaming assembly
5. Schema validation
6. Post-processing
7. Persistence to DB or downstream system

For large documents

• split into sections
• extract per section
• merge results
• deduplicate entities
• reconcile conflicting values

8) Example with streaming in practice

Python-style pseudo-code

import json

def extract_with_streaming(client, document_text):
    stream = client.chat.completions.create(
        model="your-model",
        stream=True,
        messages=[
            {"role": "system", "content": "Return only valid JSON."},
            {"role": "user", "content": f"Extract fields from:\n{document_text}"}
        ]
    )

    buffer = ""
    for event in stream:
        if event.choices[0].delta.get("content"):
            buffer += event.choices[0].delta["content"]

    return json.loads(buffer)

9) Best practices

• Keep prompts short and precise
• Use schemas whenever possible
• Require null for missing fields
• Avoid asking for explanation text
• Make outputs machine-parseable
• Add retries for malformed JSON
• Use confidence scores if your workflow needs triage

10) Common pitfalls

• Partial JSON during streaming: don’t parse until complete unless you have incremental parsing logic
• Hallucinated fields: constrain with schema and explicit instructions
• Mixed prose + JSON: instruct “return only JSON”
• Inconsistent formatting: add examples and normalization rules
• Long-document overload: chunk and merge

If you want, I can give you:

1. a concrete OpenAI API example for streaming extraction,
2. a Node.js example, or
3. a FastAPI backend template for an extraction service.