# Observations Observations are the **building blocks** inside a trace. While a trace captures the complete end-to-end operation, observations record what happened at **each individual step**: model calls, tool invocations, retrieval operations, and any other discrete action in your workflow. Observations can be nested, forming hierarchies that reflect the structure of your application's pipeline. Each observation contains a start time and its latency, a name and type, attributes (key-value metadata), and parent-child relationships for hierarchical traces. You can navigate to `https://app.interactive.ai/project//observations` to see all observations in your project. {% hint style="info" %} For the full Tracing API reference including all method signatures, parameters, and advanced options, see the [SDK Documentation](https://app.gitbook.com/s/jHEEbkpMbUW2x51XS8Ez/tracing). {% endhint %} ### Why Observations Matter Observations give you granular visibility into each step of your LLM pipeline. With observations, you can: * **Measure latency and cost** at each step, not just end-to-end * **Identify** which specific operation caused a failure or unexpected result * **Compare performance** across different model versions or configurations * Build datasets from specific observations for **targeted evaluation** *** ### Observation Types InteractiveAI supports several observation types, each designed for specific kinds of operations:
TypeDescription
SpanA unit of work with a start and end time; use for any operation you want to measure
GenerationAn LLM call; captures prompts, completions, model details, token usage, and costs
ToolA tool or function invocation; records the tool name, inputs, and outputs
EventA discrete occurrence without duration; use for logging specific moments
. AgentAn orchestration step where an LLM decides what action to take
ChainA sequence of operations linked together
RetrieverA data retrieval operation (e.g., querying a vector database)
EmbeddingA vector embedding generation; captures model, token usage, and costs
EvaluatorAn evaluation function execution
GuardrailA safety check against malicious input or unsafe content
`generation` and `embedding` are special types that accept additional parameters: `model`, `model_parameters`, `usage_details`, `cost_details`, `completion_start_time`, and `prompt`. All other types share the same base parameters. *** ### Creating Observations Observations are always created **within the context of a trace**. When nested inside an existing trace, they automatically become children of the current span. {% hint style="info" %} `@observe` examples on this page assume `from interactiveai import observe` has been imported. {% endhint %} {% tabs %} {% tab title="Context Manager" %} ```python with interactiveai.start_as_current_observation(as_type="span", name="rag-pipeline") as span: span.update(input={"query": "What is RAG?"}) with interactiveai.start_as_current_observation( as_type="retriever", name="vector-search" ) as retriever: results = search_documents("What is RAG?") retriever.update( input={"query": "What is RAG?", "top_k": 5}, output={"documents": results} ) with interactiveai.start_as_current_observation( as_type="generation", name="answer-generation", model="gpt-4" ) as generation: generation.update( input={"prompt": "Based on these documents, what is RAG?"}, output={"response": "RAG is a technique..."} ) span.update(output={"status": "completed"}) interactiveai.flush() ``` For complex pipelines, observations can be nested multiple levels deep: ```python with interactiveai.start_as_current_observation(as_type="span", name="document-processor") as outer_span: outer_span.update(input={"request": "Process document"}) with interactiveai.start_as_current_observation(as_type="generation", name="summarize") as gen1: gen1.update( input={"prompt": "Summarize the document"}, output={"summary": "Document summary here"} ) with interactiveai.start_as_current_observation(as_type="span", name="post-processing") as mid_span: mid_span.update(input={"task": "Extract key points from summary"}) with interactiveai.start_as_current_observation(as_type="generation", name="extract") as gen2: gen2.update( input={"prompt": "Extract key points"}, output={"key_points": ["Point 1", "Point 2"]} ) mid_span.update(output={"status": "done"}) outer_span.update(output={"status": "finished"}) interactiveai.flush() ``` {% endtab %} {% tab title="@observe Decorator" %} When decorated functions call other decorated functions, nesting happens automatically: ```python @observe() def rag_pipeline(query): results = vector_search(query) answer = generate_answer(query, results) return {"results": results, "answer": answer} @observe(as_type="retriever", name="vector-search") def vector_search(query): # Automatically becomes a child of rag_pipeline return search_documents(query) @observe(as_type="generation", name="answer-generation") def generate_answer(query, context): # Automatically becomes a child of rag_pipeline return llm.generate(query=query, context=context) # Creates: rag_pipeline → vector_search + answer_generation rag_pipeline("What is RAG?") interactiveai.flush() ``` This produces the same nested hierarchy as the context manager version, without manual nesting. {% endtab %} {% endtabs %} *** ### Updating Observations After creating an observation, you can update it with additional data that becomes available during execution. {% tabs %} {% tab title="Context Manager" %} Use `update()` on the observation object: ```python with interactiveai.start_as_current_observation(as_type="span", name="process") as span: result = step_one() span.update(metadata={"step": "one", "intermediate_result": result}) final = step_two(result) span.update(output={"final": final}) ``` For generation and embedding observations, `update()` also accepts model-specific parameters: ```python with interactiveai.start_as_current_observation( as_type="generation", name="llm-call", model="gpt-4" ) as generation: response = llm.generate(query) generation.update( output=response.text, usage_details={ "prompt_tokens": response.usage.prompt_tokens, "completion_tokens": response.usage.completion_tokens }, cost_details={"input": 0.0012, "output": 0.0035} ) ``` {% endtab %} {% tab title="@observe Decorator" %} Inside a decorated function, use `update_current_span()` to add data to the current observation: ```python @observe() def process_pipeline(query): result = step_one(query) interactiveai.update_current_span( metadata={"step": "one", "intermediate_result": result} ) final = step_two(result) interactiveai.update_current_span(output={"final": final}) return final ``` For generation and embedding observations, use `update_current_generation()` which accepts model-specific parameters: ```python @observe(as_type="generation", name="llm-call") def generate_response(query): response = llm.generate(query) interactiveai.update_current_generation( model="gpt-4", model_parameters={"temperature": 0.7, "max_tokens": 500}, output=response.text, usage_details={ "prompt_tokens": response.usage.prompt_tokens, "completion_tokens": response.usage.completion_tokens }, cost_details={"input": 0.0012, "output": 0.0035} ) return response.text ``` {% endtab %} {% endtabs %} *** ### Properties of an Observation | Property | Description | | ----------------------- | ----------------------------------------------------------------------------- | | **Name** | The name of the observation (e.g., "llm-generation", "vector-search") | | **Start Time** | Timestamp when the observation began | | **End Time** | Timestamp when the observation completed | | **Input** | JSON payload capturing the request or input data | | **Output** | JSON payload capturing the response or output data | | **Level** | Importance level to control verbosity: `DEBUG`, `DEFAULT`, `WARNING`, `ERROR` | | **Status Message** | Additional information, such as error details when level is `ERROR` | | **Latency** | Duration of the observation from start to end | | **Model** | The model used (for generations and embeddings) | | **Model Cost** | Cost of the operation (for generations and embeddings) | | **Time to First Token** | Time elapsed before the first token was received | | **Tokens** | Token count (input + output) for the observation | | **Prompt** | Link to prompt version in InteractiveAI prompt management | | **Environment** | Deployment context like `production`, `staging`, or `development` | | **Trace Tags** | Tags inherited from the parent trace | | **Metadata** | Free-form JSON for extra context specific to this observation | | **Observation ID** | Unique identifier (16-character lowercase hexadecimal string) | | **Trace Name** | Name of the parent trace containing this observation | | **Trace ID** | Unique identifier of the parent trace | *** ### Observation Type Examples #### Span A generic unit of work. Use when no specialized type fits your operation: {% tabs %} {% tab title="Context Manager" %} ```python with interactiveai.start_as_current_observation( as_type="span", name="data-preprocessing" ) as span: cleaned = clean_data(raw_input) validated = validate_schema(cleaned) span.update( input={"raw_records": len(raw_input)}, output={"valid_records": len(validated), "rejected": len(raw_input) - len(validated)} ) ``` {% endtab %} {% tab title="@observe Decorator" %} ```python @observe(as_type="span", name="data-preprocessing") def preprocess(raw_input): cleaned = clean_data(raw_input) validated = validate_schema(cleaned) return {"valid_records": len(validated), "rejected": len(raw_input) - len(validated)} preprocess(raw_input) ``` {% endtab %} {% endtabs %} Agent, Chain, and Evaluator work identically to Span but are visually differentiated in the InteractiveAI UI for filtering and organization. Use `as_type="agent"`, `as_type="chain"`, or `as_type="evaluator"` respectively. #### Generation Captures model, token usage, and costs: {% tabs %} {% tab title="Context Manager" %} ```python with interactiveai.start_as_current_observation( as_type="generation", name="chat-completion", model="gpt-4", model_parameters={"temperature": 0.7, "max_tokens": 500} ) as generation: generation.update( input={"messages": [{"role": "user", "content": "Explain quantum computing"}]}, output={"response": "Quantum computing uses quantum mechanical phenomena..."}, usage_details={"prompt_tokens": 12, "completion_tokens": 85}, cost_details={"input": 0.0004, "output": 0.0051} ) ``` {% endtab %} {% tab title="@observe Decorator" %} ```python @observe(as_type="generation", name="chat-completion") def chat_completion(messages): response = openai_client.chat.completions.create( model="gpt-4", messages=messages, temperature=0.7, max_tokens=500 ) interactiveai.update_current_generation( model="gpt-4", model_parameters={"temperature": 0.7, "max_tokens": 500}, usage_details={ "prompt_tokens": response.usage.prompt_tokens, "completion_tokens": response.usage.completion_tokens } ) return response.choices[0].message.content chat_completion([{"role": "user", "content": "Explain quantum computing"}]) interactiveai.flush() ``` {% endtab %} {% endtabs %} #### Tool Use for function or tool invocations: {% tabs %} {% tab title="Context Manager" %} ```python with interactiveai.start_as_current_observation( as_type="tool", name="web-search" ) as tool: tool.update( input={"query": "latest AI news"}, output={"results": [{"title": "Article 1", "url": "https://..."}]} ) ``` {% endtab %} {% tab title="@observe Decorator" %} ```python @observe(as_type="tool", name="web-search") def web_search(query): results = search_engine.search(query) return {"results": results} web_search("latest AI news") ``` {% endtab %} {% endtabs %} #### Retriever Use for vector database queries or document retrieval: {% tabs %} {% tab title="Context Manager" %} ```python with interactiveai.start_as_current_observation( as_type="retriever", name="vector-search" ) as retriever: retriever.update( input={"query": "machine learning basics", "top_k": 5}, output={"documents": [{"id": "doc1", "score": 0.95, "content": "..."}]} ) ``` {% endtab %} {% tab title="@observe Decorator" %} ```python @observe(as_type="retriever", name="vector-search") def vector_search(query, top_k=5): results = vector_db.search(query, top_k=top_k) return {"documents": results} vector_search("machine learning basics", top_k=5) ``` {% endtab %} {% endtabs %} #### Embedding Use for embedding generation. Like `generation`, this type captures model, token usage, and costs: {% tabs %} {% tab title="Context Manager" %} ```python with interactiveai.start_as_current_observation( as_type="embedding", name="document-embedding", model="text-embedding-3-small" ) as embedding: embedding.update( input={"text": "Retrieval-Augmented Generation combines retrieval with language model generation..."}, output={"vector": [0.023, -0.041, 0.117, ...]}, usage_details={"prompt_tokens": 14} ) ``` {% endtab %} {% tab title="@observe Decorator" %} ```python @observe(as_type="embedding", name="document-embedding") def embed_document(text): response = openai_client.embeddings.create( model="text-embedding-3-small", input=text ) # Embedding is a generation-like type internally, so use update_current_generation interactiveai.update_current_generation( model="text-embedding-3-small", usage_details={"prompt_tokens": response.usage.prompt_tokens} ) return response.data[0].embedding embed_document("Retrieval-Augmented Generation combines retrieval with language model generation...") ``` {% endtab %} {% endtabs %} #### Event A discrete, zero-duration occurrence. Use for logging specific moments like user feedback, system events, or state changes. Events use the dedicated `create_event()` method — they cannot be created with `@observe` or `start_as_current_observation` because they have no duration to wrap around. ```python interactiveai.create_event( name="user-feedback-received", input={"feedback_type": "thumbs_up", "message_id": "msg-789"}, metadata={"source": "web-app"} ) ``` `create_event()` accepts the same base parameters as other observation types (`name`, `input`, `output`, `metadata`, `version`, `level`, `status_message`) but creates an observation that starts and ends at the same instant. #### Guardrail Use for safety checks: {% tabs %} {% tab title="Context Manager"" %} ```python with interactiveai.start_as_current_observation( as_type="guardrail", name="content-safety-check" ) as guardrail: guardrail.update( input={"content": "User message to check"}, output={"safe": True, "categories": {"toxicity": False, "hate": False}} ) ``` {% endtab %} {% tab title="@observe Decorator" %} ```python @observe(as_type="guardrail", name="content-safety-check") def check_content_safety(content): result = safety_classifier.check(content) return {"safe": result.is_safe, "categories": result.categories} check_content_safety("User message to check") ``` {% endtab %} {% endtabs %}