)
This library provides tools to instrument and evaluate your AI applications using Gentrace.
The full API documentation can be found in api.md.
# install from PyPI
pip install gentrace-pyThe Gentrace SDK exposes several key functions to help you instrument and evaluate your AI pipelines:
init– Initialise the SDK with your API key and optional base URL.interaction– Decorator to trace a single function that performs your core AI logic. (Requires OpenTelemetry)experiment– Context decorator that groups related evaluation runs. (Requires OpenTelemetry)eval– Decorator that defines a single evaluation (test case) to run inside an experiment. (Requires OpenTelemetry)eval_dataset– Helper that runs an interaction against every test-case in a dataset. (Requires OpenTelemetry)
All of these utilities rely on OpenTelemetry to capture and export spans, which represent units of work or operations within your application. These spans are then sent to Gentrace for visualization and analysis. Make sure you have an OTel SDK running (see OpenTelemetry Integration).
TIP You can get your Gentrace API key at https://gentrace.ai/s/api-keys
importosfromgentraceimportinitGENTRACE_API_KEY=os.environ["GENTRACE_API_KEY"]
init(
api_key=GENTRACE_API_KEY,
# Optional for self-hosted deployments: base_url=os.environ.get("GENTRACE_BASE_URL", "https://gentrace.ai/api")
)
print("Gentrace initialised!")Wrap the function that contains your AI logic so each call is traced.
importopenaifromgentraceimportinteraction, initOPENAI_API_KEY=os.environ["OPENAI_API_KEY"]
GENTRACE_API_KEY=os.environ["GENTRACE_API_KEY"]
GENTRACE_PIPELINE_ID=os.environ["GENTRACE_PIPELINE_ID"]
init(
api_key=GENTRACE_API_KEY,
# Optional for self-hosted deployments: base_url=os.environ.get("GENTRACE_BASE_URL", "https://gentrace.ai/api")
)
client=OpenAI(api_key=OPENAI_API_KEY)
@interaction(pipeline_id=GENTRACE_PIPELINE_ID)asyncdefquery_ai(query: str) ->str|None:
response=client.chat.completions.create(
model="gpt-4o", messages=[{"role": "user", "content": query}]
)
returnresponse.choices[0].message.contentEach call to a function decorated with @interaction (like query_ai above) creates a span, capturing its execution details and any associated metadata, inputs, and outputs. This span is then sent to Gentrace.
If your organization has a default pipeline configured, you can use @interaction without specifying a pipeline ID:
fromgentraceimportinteraction, initinit(api_key=GENTRACE_API_KEY)
# Simplest usage - no pipeline ID required@interaction()defprocess_data(data: str) ->str:
returnf"Processed: {data}"# With custom attributes but no pipeline ID@interaction(attributes={"model": "gpt-4", "temperature": 0.7})defanalyze_data(data: str) ->dict:
return {"analysis": data.upper(), "length": len(data)}
# Async function with custom name@interaction(name="Custom Analysis")asyncdefasync_process(data: str) ->str:
awaitasyncio.sleep(0.1)
returnf"Async processed: {data}"When no pipeline_id is provided, the SDK automatically uses your organization's default pipeline.
Use the traced decorator to wrap any function with OpenTelemetry tracing, creating a span for its execution. This is useful for instrumenting helper functions or specific blocks of code within a larger system. (Requires OpenTelemetry)
importasynciofromgentraceimporttraced, interactionfromopenaiimportOpenAIclient=OpenAI()
USER_ID="<user_id>"PIPELINE_ID="<pipeline_id>"@traced(name="OpenAI Call")asyncdefsummarize_user(user_info: str) ->str:
response=client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": f"Summarize the following user info: {user_info}"}],
)
returnresponse.choices[0].message.content@traced(name="Get User Info DB Call")asyncdefget_user_info(user_id: str) ->str:
# This would be a database call in a real applicationreturnf"User {user_id}: Sample information"@interaction(pipeline_id=GENTRACE_PIPELINE_ID)asyncdefmain_task(input: str) ->str:
user_info=awaitget_user_info(input)
returnawaitsummarize_user(user_info)
asyncio.run(main_task(USER_ID))You can also provide additional attributes to the @traced decorator to add to the span. Like interaction, this also requires OpenTelemetry to be set up properly.
Use experiment to create a testing context and eval for individual test cases.
importasynciofromgentraceimportexperiment, evalimportosGENTRACE_API_KEY=os.environ["GENTRACE_API_KEY"]
GENTRACE_PIPELINE_ID=os.environ["GENTRACE_PIPELINE_ID"]
init(
api_key=GENTRACE_API_KEY,
# Optional for self-hosted deployments: base_url=os.environ.get("GENTRACE_BASE_URL", "https://gentrace.ai/api")
)
@interaction(pipeline_id=GENTRACE_PIPELINE_ID)asyncdefquery_ai(query: str) ->str|None:
# Implementation from previous examplepass@experiment(pipeline_id=GENTRACE_PIPELINE_ID)asyncdefsimple_evals() ->None:
@eval(name="capital-of-france")asyncdefparis_test() ->None:
result=awaitquery_ai("What is the capital of France?")
assertresultand"Paris"inresult# Immediately invoke the evalawaitparis_test()
asyncio.run(simple_evals())The @eval decorator creates a 'test' span for paris_test. When query_ai (an @interaction-decorated function) is called within paris_test, its own interaction span is also created. This interaction span is nested under the 'test' span, creating a trace of the evaluation. Both spans are sent to Gentrace.
importasyncio, osfromgentraceimportTestCase, TestInput, init, experiment, eval_dataset, test_cases_asyncfromtyping_extensionsimportTypedDictfrompydanticimportBaseModelGENTRACE_API_KEY=os.environ["GENTRACE_API_KEY"]
GENTRACE_PIPELINE_ID=os.environ["GENTRACE_PIPELINE_ID"]
GENTRACE_DATASET_ID=os.environ["GENTRACE_DATASET_ID"]
init(
api_key=GENTRACE_API_KEY,
# Optional for self-hosted deployments: base_url=os.environ.get("GENTRACE_BASE_URL", "https://gentrace.ai/api")
)
# Option 1️⃣: Fetch test cases from Gentraceasyncdeffetch_test_cases() ->list[TestCase]:
cases=awaittest_cases_async.list(dataset_id=GENTRACE_DATASET_ID)
# Each test case within cases.data has an attribute "inputs" with the structure: { query: str }returncases.data# Option 2️⃣: Provide locally defined test cases by using TestInput and a typed dict# (in this case QueryInputs)classQueryInputs(TypedDict):
query: strdefcustom_test_cases() ->list[TestInput[QueryInputs]]:
return [
TestInput[QueryInputs](name="Test Case 1", inputs={"query": "Hello, World!"}),
TestInput[QueryInputs](name="Test Case 2", inputs={"query": "How does this work?"}),
]
# Optionally, validate the structure of your inputs with PydanticclassQueryInputsSchema(BaseModel):
query: str@experiment(pipeline_id=GENTRACE_PIPELINE_ID)asyncdefdataset_evals() ->None:
# Option 1️⃣: Use test cases from Gentraceawaiteval_dataset(
data=fetch_test_cases,
interaction=query_ai,
schema=QueryInputsSchema, # Extra validation with Pydantic of the test case structure
)
# Option 2️⃣: Use locally defined test casesawaiteval_dataset(
data=custom_test_cases,
interaction=query_ai,
)
asyncio.run(dataset_evals())The eval_dataset utility creates a 'test' span for each test case processed from the dataset. If the interaction argument (e.g., query_ai) is an @interaction-decorated function, then for each test case, an additional interaction span is created.
This interaction span is nested within its corresponding 'test' span. All these spans are sent to Gentrace, allowing detailed analysis of how the interaction performs across the entire dataset.
[SPAN] Test Case 1
├─── [SPAN] @interaction Function
│ ├─── [SPAN] @traced (LLM call)
│ │ ├─── Model: gpt-4
│ │ ├─── Input: "What is the capital of France?"
│ │ └─── Output: "The capital of France is Paris."
│ │
│ └─── [SPAN] @traced (Tool call)
│ ├─── Tool: search_wikipedia
│ ├─── Input: "Paris"
│ └─── Output: "Paris is the capital and most populous city of France..."
This hierarchical structure allows Gentrace to provide detailed analysis of performance and behavior at different levels of granularity.
OpenTelemetry must be running for spans created by interaction, experiment, eval, and eval_dataset to be exported. The OpenTelemetry SDK is included as a dependency of this package.
Example setup:
fromopentelemetry.sdk.resourcesimportResourcefromopentelemetry.sdk.traceimportTracerProviderfromopentelemetry.sdk.trace.exportimportSimpleSpanProcessorfromopentelemetry.exporter.otlp.proto.http.trace_exporterimportOTLPSpanExporterfromopentelemetryimporttracefromgentraceimportGentraceSampler, GentraceSpanProcessorimportos# In virtually all cases, you should use https://gentrace.ai/api as the base URLGENTRACE_BASE_URL=os.environ.get("GENTRACE_BASE_URL", "https://gentrace.ai/api")
GENTRACE_API_KEY=os.environ["GENTRACE_API_KEY"]
resource=Resource.create({"service.name": "my-gentrace-app"})
provider=TracerProvider(
resource=resource,
sampler=GentraceSampler(), # Use GentraceSampler for selective tracing
)
trace.set_tracer_provider(provider)
exporter=OTLPSpanExporter(
endpoint=f"{GENTRACE_BASE_URL}/otel/v1/traces",
headers={"Authorization": f"Bearer {GENTRACE_API_KEY}"},
)
# Add GentraceSpanProcessor to propagate gentrace.sample attributeprovider.add_span_processor(GentraceSpanProcessor())
provider.add_span_processor(SimpleSpanProcessor(exporter))
processor=SimpleSpanProcessor(exporter)
provider.add_span_processor(processor)
print("OpenTelemetry SDK started – spans will be sent to Gentrace.")Gentrace provides two specialized OpenTelemetry components to help control which spans are sent to Gentrace:
The GentraceSampler is a custom OpenTelemetry sampler that selectively samples spans based on the presence of a gentrace.sample attribute. This helps reduce the volume of telemetry data by only sending relevant spans to Gentrace.
fromgentraceimportGentraceSamplerfromopentelemetry.sdk.traceimportTracerProvider# Create a tracer provider with the GentraceSamplerprovider=TracerProvider(resource=resource, sampler=GentraceSampler())How it works:
- The sampler checks for the
gentrace.samplekey in the OpenTelemetry Baggage or as a span attribute - If
gentrace.sampleis set to"true", the span will be sampled and exported to Gentrace - Otherwise, the span will be dropped and not exported
This is particularly useful for filtering out spans that are not relevant to Gentrace tracing, reducing noise and data volume.
The GentraceSpanProcessor is a specialized span processor that ensures the gentrace.sample attribute is properly propagated from the OpenTelemetry Baggage to span attributes.
fromgentraceimportGentraceSpanProcessor# Add the GentraceSpanProcessor to your tracer providerprovider.add_span_processor(GentraceSpanProcessor())How it works:
- When a span starts, the processor checks for the
gentrace.samplekey in the current OpenTelemetry Baggage - If found, it extracts this value and adds it as an attribute to the span
- This ensures that the sampling attribute is propagated correctly to all spans that need to be tracked by Gentrace
Using both components together provides optimal control over which spans are sent to Gentrace:
# Complete exampleprovider=TracerProvider(resource=resource, sampler=GentraceSampler())
trace.set_tracer_provider(provider)
# Add GentraceSpanProcessor first to ensure proper attribute propagationprovider.add_span_processor(GentraceSpanProcessor())
# Then add your exporter processorprovider.add_span_processor(SimpleSpanProcessor(exporter))See the examples guide for instructions on how to run the examples.
Python 3.8 or newer.
See the contributing guide.
Questions or feedback? support@gentrace.ai