Python SDK

pip install weavel

# .env
WEAVEL_API_KEY=<API_KEY>

import os
from fastapi import FastAPI
from openai import AsyncOpenAI
from weavel import create_client
 
app = FastAPI()
 
openai = AsyncOpenAI()
weavel = create_client() # Create Weavel Client
 
@app.post("signup")
async def signup(
    properties
):
    # create a new user
    user_id = create_user()
    weavel.identify(user_id, properties) # Identify the user
 
@app.post("/response")
async def response(
    user_id, conversation_id, content, metadata
):
    system_prompt = "You are a helpful assistant"
    if conversation_id is None:
        trace = weavel.open_trace(  # For new conversation, open a new trace
            user_id=user_id,
            trace_id=conversation_id,
            metadata=metadata
        )
        messages = [{"role": "system", "content": system_prompt}]
    else:
        trace = weavel.resume_trace( # For existing conversation, resume the trace
            user_id=user_id,
            trace_id=conversation_id,
            metadata=metadata
        )
        messages = get_context(conversation_id)
    messages += [{"role": "user", "content": content}]
 
    trace.log_message("user", content) # Log the user message
 
    res = await openai.chat.completions.create(
        model=metadata["model"]
        messages=messages
    )
 
    message_id = save_response(res)
    trace.log_message(
      "assistant",
      res,
      trace_data_id=message_id) # Log the assistant message
    return res, message_id
 
 
@app.post("/feedback")
async def feedback(
    user_id, message_id, feedback
):
    weavel.log_message_metadata( # Log the feedback for the message
        trace_data_id=message_id,
        metadata={"feedback": feedback}
    )
    save_feedback(message_id, feedback)
    return
 
 
@app.post("/subscribe")
async def subscribe(
    user_id, plan
):
    weavel.track( # Log track_event:subscribe
        user_id=user_id,
        event_name="subscribe",
        properties={"plan": plan}
    )
    subscribe_user(user_id, plan)
    return
 
@app.post("/user/update")
async def user_update(
    user_id, properties
):
    weavel.identify(user_id, properties) # Update user properties
    update_user(user_id, properties)
    return