The Respan tracing SDK provides decorators and wrappers to instrument your code. This page covers the detailed features.
Decorators & span types
Use decorators (Python) or wrapper methods (JS/TS) to create spans. Each decorator creates a different span type in the trace tree.
| Decorator | Span type | Use for |
|---|
@workflow / withWorkflow | Workflow | Top-level parent span |
@task / withTask | Task | Individual steps within a workflow |
@agent / withAgent | Agent | Agent-level operations |
@tool / withTool | Tool | Tool/function calls |
from openai import OpenAI
from respan_tracing.decorators import workflow, task
from respan_tracing.main import RespanTelemetry
k_tl = RespanTelemetry()
client = OpenAI()
@task(name="joke_creation")
def create_joke():
completion = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "Tell me a joke about opentelemetry"}],
temperature=0.5,
max_tokens=100,
)
return completion.choices[0].message.content
@task(name="pirate_joke_translation")
def translate_joke_to_pirate(joke: str):
completion = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "translate to pirate language:\n\n" + joke}],
)
return completion.choices[0].message.content
@workflow(name="joke_workflow")
def joke_workflow():
joke = create_joke()
pirate_joke = translate_joke_to_pirate(joke)
return pirate_joke
result = joke_workflow()
print(result)
For the full SDK API reference, see the Python SDK or TypeScript SDK.
Class-based workflows
Apply @workflow to a class with method_name to designate the entry point. Use @task on individual methods.
from openai import OpenAI
from respan_tracing import RespanTelemetry
from respan_tracing.decorators import workflow, task
k_tl = RespanTelemetry()
client = OpenAI()
@workflow(name="joke_agent", method_name="run")
class JokeAgent:
@task(name="joke_creation")
def create_joke(self):
completion = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "Tell me a joke"}],
)
return completion.choices[0].message.content
def run(self):
return self.create_joke()
agent = JokeAgent()
result = agent.run()
Control what appears as a span’s input/output in the trace UI using OpenTelemetry attributes:
import json
from opentelemetry.semconv_ai import SpanAttributes
from respan_tracing.main import RespanTelemetry
from respan_tracing.decorators import workflow
k_tl = RespanTelemetry()
client = k_tl.get_client()
@workflow(name="update_attributes_test")
def update_attributes_test(input: str):
force_set_attributes = {
SpanAttributes.TRACELOOP_ENTITY_INPUT: json.dumps({
"args": [],
"kwargs": {"text": "custom input"}
}),
}
client.update_current_span(
attributes=force_set_attributes,
name="update_attributes_test",
respan_params={"metadata": {"test": "test"}},
)
return "Some desired output"
if __name__ == "__main__":
update_attributes_test("Some input")
Use update_current_span(attributes=...) for overriding displayed input/output. Values must be JSON-serializable strings via json.dumps(...).