Instructor (tracing)

1
from respan_tracing import RespanTelemetry
2
from respan_instrumentation_instructor import InstructorInstrumentor
3
4
telemetry = RespanTelemetry(
5
    app_name="structured-output-api",
6
    is_auto_instrument=False,
7
    customer_identifier="user_123",
8
    metadata={"service": "extraction-api", "version": "1.0.0"},
9
)
10
InstructorInstrumentor().activate()

1
from respan_tracing.exporters import propagate_attributes
2
3
with propagate_attributes(
4
    customer_identifier="user_456",
5
    thread_identifier="conversation_abc",
6
    metadata={
7
        "route": "invoice-extraction",
8
        "example_script": "01_create.py",
9
    },
10
):
11
    invoice = client.create(
12
        response_model=Invoice,
13
        messages=[{"role": "user", "content": "Extract this invoice..."}],
14
        model="gpt-4o-mini",
15
    )

1
from typing import TypedDict
2
3
from respan_tracing import task, workflow
4
5
class Invoice(TypedDict):
6
    vendor: str
7
    invoice_id: str
8
    total_usd: float
9
10
@task(name="extract_invoice")
11
def extract_invoice(text: str) -> Invoice:
12
    return client.create(
13
        response_model=Invoice,
14
        messages=[{"role": "user", "content": text}],
15
        model="gpt-4o-mini",
16
    )
17
18
@workflow(name="invoice_processing")
19
def invoice_processing(raw_invoice: str) -> None:
20
    invoice = extract_invoice(raw_invoice)
21
    print(dict(invoice))
22
23
invoice_processing("Northwind invoice NW-1042 totals 298.00 USD.")
24
telemetry.flush()

1
from typing import TypedDict
2
3
class InvoiceLine(TypedDict):
4
    sku: str
5
    quantity: int
6
    unit_price_usd: float
7
8
class Invoice(TypedDict):
9
    vendor: str
10
    invoice_id: str
11
    line_items: list[InvoiceLine]
12
    total_usd: float
13
14
invoice = client.create(
15
    response_model=Invoice,
16
    messages=[
17
        {
18
            "role": "user",
19
            "content": "Northwind invoice NW-1042 has 2 LOG-001 at 49.50 and 1 SEC-010 at 199.00. Total is 298.00.",
20
        }
21
    ],
22
    model="gpt-4o-mini",
23
)
24
print(dict(invoice))

1
from typing import Literal, TypedDict
2
3
class ReleaseNote(TypedDict):
4
    title: str
5
    category: Literal["feature", "fix", "docs"]
6
    bullets: list[str]
7
8
release_note, completion = client.create_with_completion(
9
    response_model=ReleaseNote,
10
    messages=[
11
        {
12
            "role": "user",
13
            "content": "Create a release note for preserving Instructor tool schemas in traces.",
14
        }
15
    ],
16
    model="gpt-4o-mini",
17
)
18
19
print(dict(release_note))
20
print(completion.id)

1
from typing import Literal, TypedDict
2
3
class ActionItem(TypedDict):
4
    owner: str
5
    task: str
6
    status: Literal["new", "blocked", "done"]
7
8
items = list(
9
    client.create_iterable(
10
        response_model=ActionItem,
11
        messages=[
12
            {
13
                "role": "user",
14
                "content": "Maya will send the launch checklist. Noah is blocked on legal approval.",
15
            }
16
        ],
17
        model="gpt-4o-mini",
18
    )
19
)
20
print([dict(item) for item in items])

1
import asyncio
2
from typing import TypedDict
3
4
import instructor
5
from openai import AsyncOpenAI
6
7
async_client = instructor.from_openai(
8
    AsyncOpenAI(),
9
    mode=instructor.Mode.TOOLS,
10
)
11
12
class ProjectBrief(TypedDict):
13
    title: str
14
    owner: str
15
    milestones: list[str]
16
17
@workflow(name="project_brief")
18
async def create_project_brief() -> ProjectBrief:
19
    return await async_client.create(
20
        response_model=ProjectBrief,
21
        messages=[
22
            {
23
                "role": "user",
24
                "content": "Build a project brief for the Respan Instructor tracing launch.",
25
            }
26
        ],
27
        model="gpt-4o-mini",
28
    )
29
30
asyncio.run(create_project_brief())
31
telemetry.flush()

1
from instructor.core.hooks import HookName
2
3
def on_completion_kwargs(**kwargs):
4
    print({"model": kwargs.get("model"), "tool_count": len(kwargs.get("tools", []))})
5
6
client.on(HookName.COMPLETION_KWARGS, on_completion_kwargs)
7
8
try:
9
    ticket = client.create(
10
        response_model=SupportTicket,
11
        max_retries=2,
12
        messages=[{"role": "user", "content": "ACME needs urgent login help."}],
13
        model="gpt-4o-mini",
14
    )
15
finally:
16
    client.off(HookName.COMPLETION_KWARGS, on_completion_kwargs)