Sessions provide a powerful way to group related spans together, making it easier to track and analyze complex workflows, user interactions, or multi-step processes. This guide covers everything you need to know about working with sessions. For complete API documentation, see the Python SDK Reference or TypeScript SDK Reference.

Creating Sessions

Basic Session with ID

The simplest way to create a session is by providing a session ID:
import uuid
import zeroeval as ze

# Generate a unique session ID
session_id = str(uuid.uuid4())

@ze.span(name="process_request", session=session_id)
def process_request(data):
    # This span belongs to the session
    return transform_data(data)

Named Sessions

For better organization in the ZeroEval dashboard, you can provide both an ID and a descriptive name:
@ze.span(
    name="user_interaction",
    session={
        "id": session_id,
        "name": "Customer Support Chat - User #12345"
    }
)
def handle_support_chat(user_id, message):
    # Process the support request
    return generate_response(message)

Session Inheritance

Child spans automatically inherit the session from their parent span:
session_info = {
    "id": str(uuid.uuid4()),
    "name": "Order Processing Pipeline"
}

@ze.span(name="process_order", session=session_info)
def process_order(order_id):
    # These nested calls automatically belong to the same session
    validate_order(order_id)
    charge_payment(order_id)
    fulfill_order(order_id)
    
@ze.span(name="validate_order")
def validate_order(order_id):
    # Automatically part of the parent's session
    return check_inventory(order_id)

@ze.span(name="charge_payment")
def charge_payment(order_id):
    # Also inherits the session
    return process_payment(order_id)

Advanced Session Patterns

Multi-Agent RAG System

Track complex retrieval-augmented generation workflows with multiple specialized agents:
session = {
    "id": str(uuid.uuid4()),
    "name": "Multi-Agent RAG Pipeline"
}

@ze.span(name="rag_coordinator", session=session)
async def process_query(query):
    # Retrieval
    docs = await retrieval_agent(query)
    
    # Reranking  
    ranked = await reranking_agent(query, docs)
    
    # Generation
    response = await generation_agent(query, ranked)
    
    return response

@ze.span(name="retrieval_agent")
async def retrieval_agent(query):
    # Inherits session from parent
    embeddings = await embed(query)
    return await vector_search(embeddings)

@ze.span(name="generation_agent")
async def generation_agent(query, context):
    return await llm.generate(query, context)

Conversational AI Session

Track a complete conversation with an AI assistant:
class ChatSession:
    def __init__(self, user_id):
        self.session = {
            "id": f"chat-{user_id}-{uuid.uuid4()}",
            "name": f"AI Chat - User {user_id}"
        }
        self.history = []
    
    @ze.span(name="process_message", session=lambda self: self.session)
    async def process_message(self, message):
        # Add to history
        self.history.append({"role": "user", "content": message})
        
        # Generate response
        response = await self.generate_response()
        self.history.append({"role": "assistant", "content": response})
        
        return response
    
    @ze.span(name="generate_response", session=lambda self: self.session)
    async def generate_response(self):
        return await llm.chat(self.history)

Batch LLM Processing

Process multiple documents with LLMs in a single session:
async def batch_summarize(documents):
    session = {
        "id": f"batch-{uuid.uuid4()}",
        "name": f"Batch Summarization - {len(documents)} docs"
    }
    
    @ze.span(name="batch_processor", session=session)
    async def process():
        summaries = []
        
        for i, doc in enumerate(documents):
            with ze.span(name=f"summarize_doc_{i}", session=session) as span:
                try:
                    summary = await llm.summarize(doc)
                    span.set_io(
                        input_data=f"Doc: {doc['title']}",
                        output_data=summary[:100]
                    )
                    summaries.append(summary)
                except Exception as e:
                    span.set_error(
                        code=type(e).__name__,
                        message=str(e)
                    )
        
        return summaries
    
    return await process()

Context Manager Sessions

You can also use sessions with the context manager pattern:
session_info = {
    "id": str(uuid.uuid4()),
    "name": "Data Pipeline Run"
}

with ze.span(name="etl_pipeline", session=session_info) as pipeline_span:
    # Extract phase
    with ze.span(name="extract_data") as extract_span:
        raw_data = fetch_from_source()
        extract_span.set_io(output_data=f"Extracted {len(raw_data)} records")
    
    # Transform phase
    with ze.span(name="transform_data") as transform_span:
        clean_data = transform_records(raw_data)
        transform_span.set_io(
            input_data=f"{len(raw_data)} raw records",
            output_data=f"{len(clean_data)} clean records"
        )
    
    # Load phase
    with ze.span(name="load_data") as load_span:
        result = save_to_destination(clean_data)
        load_span.set_io(output_data=f"Loaded to {result['location']}")