Galtea’s tracing feature captures every operation your agent performs—tool calls, retrieval operations, LLM invocations—with minimal code changes. This tutorial shows you how to instrument your agent and collect traces.
For detailed information about trace properties, node types, and hierarchy, see the Trace concept page. Setup
There are two primary ways to set up tracing in your agent. Choose the option that fits your needs.
a) The @trace Decorator
Add the @trace decorator to any function you want to track. It automatically captures: name, inputs, outputs, timing, errors, and parent-child relationships.
@trace(name="db_call", type=TraceType.TOOL)
def my_function(query: str) -> str:
result = db.query(query)
return result
b) The start_trace Context Manager
For fine-grained control over specific code blocks, use start_trace.
def get_user(user_id: str) -> str:
with start_trace(
"database_query", type=TraceType.TOOL, input={"user_id": user_id}
) as span:
query = f"SELECT * FROM users WHERE id = {user_id}"
result = db.query(query)
span.update(output=result, metadata={"query": query})
return result
span.update() method lets you add output, metadata, or change the type after execution.
Both @trace and start_trace automatically capture parent-child relationships between operations when they are nested inside each other, giving you a full hierarchical view of your agent’s behavior.
Collection
Traces are built locally. To send them to Galtea, you need to associate them with an inference_result_id. There are two approaches:
a) Automatic Collection
Use inference_results.generate() or simulator.simulate() for hands-free trace management. These methods automatically:
- Set the trace context (with the appropriate setup)
- Execute your agent
- Flush all collected traces to Galtea
- Clean up the context
To achieve this, implement the Agent abstract class and decorate your methods with @trace:
@trace(type=TraceType.RETRIEVER)
def search(query: str) -> list[dict]:
return [{"id": "doc_1", "content": "..."}]
@trace(type=TraceType.GENERATION)
def generate_response(context: list, query: str) -> str:
return "Based on the context..."
@trace(type=TraceType.AGENT)
def my_agent(input_data: AgentInput) -> AgentResponse:
query = input_data.last_user_message_str()
docs = search(query)
response = generate_response(docs, query)
return AgentResponse(content=response, retrieval_context=str(docs))
# Setup
session = galtea.sessions.create(version_id=version.id, is_production=True)
Single-Turn with generate()
When using generate(), the trace context is automatically set for the entire duration of the agent’s execution. Just call generate() with your agent and session:
inference_result = galtea.inference_results.generate(
agent=my_agent, session=session, user_input="What's the price?"
)
# Traces are collected, associated with inference_result.id, and flushed automatically
Multi-Turn with simulate()
When using the Conversation Simulator, tracing works out-of-the-box. Decorate your agent methods with @trace and run:
result = galtea.simulator.simulate(
session_id=simulation_session.id, agent=my_agent, max_turns=5
)
# Traces are saved automatically for each turn
b) Manual Collection
For full control, use set_context() and clear_context() to manually manage the trace lifecycle:
# Define traced functions
@trace(type=TraceType.RETRIEVER)
def search(query: str) -> list[dict]:
return [{"id": "doc_1", "content": "..."}]
@trace(type=TraceType.GENERATION)
def generate(context: list, query: str) -> str:
return "Based on the context..."
@trace(type=TraceType.AGENT)
def run_agent(query: str) -> str:
docs = search(query)
return generate(docs, query)
# Setup
manual_session = galtea.sessions.create(version_id=version.id, is_production=True)
user_input = "What's the price?"
# 1. Create inference result first (to get the ID)
manual_inference_result = galtea.inference_results.create(
session_id=manual_session.id,
input=user_input,
output=None, # Will update later
)
# 2. Set trace context with the inference result ID
token = set_context(inference_result_id=manual_inference_result.id)
try:
# 3. Run your logic - all @trace calls will be associated with this inference result
response = run_agent(user_input)
# 4. Update inference result with the output
galtea.inference_results.update(
inference_result_id=manual_inference_result.id, output=response
)
finally:
# 5. Clear context and flush traces to Galtea
clear_context(token) # flush=True by default
clear_context(token, flush=True) automatically flushes all pending traces for the inference result before clearing. Set flush=False if you want to discard traces without sending them.
Remote Agent Tracing
When your agent runs on a remote server (e.g., deployed as a FastAPI service), OpenTelemetry’s thread-local context does not cross the HTTP boundary. The remote server cannot discover the inference_result_id to correlate traces.
To solve this, AgentInput includes an inference_result_id field that is automatically populated during generate() and simulate() calls. Forward this ID to your remote server so it can attach traces to the same inference result.
Agent / Client Side
In your agent function, read input_data.inference_result_id and send it alongside the request payload:
import httpx
from galtea import AgentInput, AgentResponse, trace, TraceType
REMOTE_URL = "https://my-remote-agent.example.com/invoke"
@trace(type=TraceType.AGENT)
def remote_agent(input_data: AgentInput) -> AgentResponse:
"""Forward execution to a remote server, passing the inference_result_id for trace correlation."""
response = httpx.post(
REMOTE_URL,
json={
"message": input_data.last_user_message_str(),
"session_id": input_data.session_id,
"inference_result_id": input_data.inference_result_id,
},
)
return AgentResponse(content=response.json()["content"])
Remote Server Side
On the remote server, use set_context() and clear_context() with the received inference_result_id:
# On the remote server (e.g. FastAPI endpoint):
from galtea import set_context, clear_context
def handle_request(message: str, session_id: str, inference_result_id: str) -> str:
# Attach traces to the same inference result
token = set_context(inference_result_id=inference_result_id)
try:
# All @trace calls here will be associated with the inference result
response = run_agent_logic(message)
return response
finally:
clear_context(token)
The remote server must have the Galtea SDK installed (pip install galtea) to use set_context() and clear_context().
Next Steps
