Instructor (gateway)

1
import os
2
from typing import TypedDict
3
4
import instructor
5
from respan_tracing import RespanTelemetry, workflow
6
from respan_tracing.exporters import propagate_attributes
7
from respan_instrumentation_instructor import InstructorInstrumentor
8
9
respan_api_key = os.environ["RESPAN_API_KEY"]
10
respan_base_url = os.getenv("RESPAN_BASE_URL", "https://api.respan.ai/api")
11
12
telemetry = RespanTelemetry(
13
    api_key=respan_api_key,
14
    base_url=respan_base_url,
15
    app_name="instructor-gateway",
16
    is_auto_instrument=False,
17
)
18
InstructorInstrumentor().activate()
19
20
client = instructor.from_provider(
21
    "openai/gpt-4o-mini",
22
    api_key=respan_api_key,
23
    base_url=respan_base_url,
24
    mode=instructor.Mode.TOOLS,
25
)
26
27
28
class Person(TypedDict):
29
    name: str
30
    role: str
31
32
33
@workflow(name="person_extraction")
34
def extract_person() -> Person:
35
    return client.create(
36
        response_model=Person,
37
        messages=[
38
            {
39
                "role": "user",
40
                "content": "Grace Hopper was a pioneering computer scientist.",
41
            }
42
        ],
43
    )
44
45
46
with propagate_attributes(
47
    thread_identifier="instructor_gateway_person_extraction",
48
    metadata={"integration": "instructor", "route": "gateway"},
49
):
50
    person = extract_person()
51
    print(dict(person))
52
53
telemetry.flush()