Skip to main content

​
Batch Processing

The Python SDK exposes synchronous completion helpers under client.completions. Async jobs that run in the background are managed under client.batches.
Method groupUse for
client.completions.classify/extract/summarize/enrich/runSynchronous completion work that returns results in the current request
client.batches.submit/retrieve/output/cancelAsync jobs that run in the background
Use client.completions.classify(...) for inline classification. For async batch classification, submit operation="label" through client.batches.submit(...).

​
Sync Completions

​
Classify

from agentflow import AgentFlow

items = [
    {"content": "The product is amazing, but delivery was slow."},
    {"content": "Terrible support experience. I waited three hours."},
]

with AgentFlow.from_profile("prod") as client:
    result = client.completions.classify(
        items=items,
        categories=["positive", "negative", "mixed"],
    )

    for row in result.results:
        print(row["result"]["label"], row["result"]["confidence"])
classify is the classification helper. The backend prompts the model to return one of the provided categories. The lower-level batch operation name is still label.

​
Extract

result = client.completions.extract(
    items=[{"content": "Sarah Chen, VP Engineering at Acme Corp, wants 500 seats by Q3."}],
    extraction_instructions="Extract contact, title, company, seat_count, and timeline.",
)

​
Summarize

result = client.completions.summarize(
    items=[{"content": transcript}],
    summary_instructions="Summarize decisions, action items, and risks in under 200 words.",
)

​
Enrich

result = client.completions.enrich(
    items=[{"content": "Enterprise SaaS company, 500 employees, Series C"}],
    fields=[
        {
            "name": "company_segment",
            "type": "categorical",
            "valid_values": ["enterprise", "mid-market", "smb"],
            "description": "Market segment",
        }
    ],
    include_context=True,
)

​
Custom Run

result = client.completions.run(
    operation="feedback_analysis",
    items=[{"content": "Customer feedback text..."}],
    system_prompt="You analyze customer feedback for product teams.",
    user_prompt_template="Analyze this feedback:\n\n{content}",
    response_format={"type": "json_object"},
)

​
Async Batches

Submit larger jobs with client.batches.submit(...). The request accepts operation, items, optional model, and the same operation-specific fields used by the backend CompletionRequest. 
from agentflow import AgentFlow

with AgentFlow.from_profile("prod") as client:
    batch = client.batches.submit(
        operation="label",
        items=[
            {"content": "The product is amazing, but delivery was slow."},
            {"content": "Terrible support experience. I waited three hours."},
        ],
        categories=["positive", "negative", "mixed"],
    )

    print(batch.id, batch.status, batch.total_items)
Use submit_and_wait when you want the SDK to poll until the job reaches a terminal status: 
batch = client.batches.submit_and_wait(
    operation="extract",
    items=[{"content": "Sarah Chen, VP Engineering at Acme Corp, wants 500 seats by Q3."}],
    extraction_instructions="Extract contact, title, company, seat_count, and timeline.",
    timeout=300,
)

​
Retrieve Status

batch = client.batches.retrieve("batch_abc123")
print(batch.status, batch.progress)
For an existing job, wait polls until the batch completes, fails, is cancelled, or reaches the optional timeout: 
batch = client.batches.wait("batch_abc123", timeout=300)

​
Read Output

JSON output returns one page with id, status, offset, limit, and results. 
page = client.batches.output("batch_abc123", format="json", offset=0, limit=1000)

for row in page.results:
    print(row["result"])
Iterate all JSON output rows without managing offsets: 
for row in client.batches.iter_output("batch_abc123", page_size=1000):
    print(row["result"])
For jsonl or csv, output returns the response text: 
jsonl_text = client.batches.output("batch_abc123", format="jsonl")
csv_text = client.batches.output("batch_abc123", format="csv")
Write output directly to disk: 
client.batches.download_output("batch_abc123", "batch-output.jsonl", format="jsonl")
client.batches.download_output("batch_abc123", "batch-output.csv", format="csv")

​
Upload a File

Use upload_file when the batch input already lives in a local file: 
batch = client.batches.upload_file(
    "reviews.jsonl",
    operation="run",
    system_prompt="Classify each review as positive, negative, or mixed.",
    user_prompt_template="{content}",
)

​
Cancel

client.batches.cancel("batch_abc123")

​
Methods

MethodDescription
client.batches.submit(...)Submit an async job
client.batches.submit_and_wait(...)Submit a job and poll until completion, failure, or cancellation
client.batches.retrieve(batch_id)Retrieve one job
client.batches.wait(batch_id, ...)Poll an existing job until it reaches a terminal status
client.batches.output(batch_id, ...)Read one JSON page or raw JSONL/CSV export
client.batches.iter_output(batch_id, ...)Iterate all JSON output rows across pages
client.batches.download_output(batch_id, path, ...)Save JSON, JSONL, or CSV output to a local file
client.batches.upload_file(...)Upload a batch input file and create a job
client.batches.cancel(batch_id)Cancel queued or running work
client.completions.run(...)Run a generic synchronous completion
client.completions.classify(...)Run synchronous classification/labeling
client.completions.extract(...)Run synchronous extraction
client.completions.summarize(...)Run synchronous summarization
client.completions.enrich(...)Run synchronous enrichment