Python SDK

%pip install langfuse --upgrade

import os
 
# get keys for your project from https://cloud.langfuse.com
os.environ["LANGFUSE_PUBLIC_KEY"] = ""
os.environ["LANGFUSE_SECRET_KEY"] = ""
 
# your openai key
os.environ["OPENAI_API_KEY"] = ""
 
# Your host, defaults to https://cloud.langfuse.com
# For US data region, set to "https://us.cloud.langfuse.com"
# os.environ["LANGFUSE_HOST"] = "http://localhost:3000"

from langfuse import Langfuse
 
langfuse = Langfuse()

# checks the SDK connection with the server.
langfuse.auth_check()

from datetime import datetime
from langfuse.model import InitialGeneration, Usage
 
generationStartTime = datetime.now()
 
# call to an LLM API
 
generation = langfuse.generation(InitialGeneration(
    name="summary-generation",
    startTime=generationStartTime,
    endTime=datetime.now(),
    model="gpt-3.5-turbo",
    modelParameters={"maxTokens": "1000", "temperature": "0.9"},
    prompt=[{"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Please generate a summary of the following documents \nThe engineering department defined the following OKR goals...\nThe marketing department defined the following OKR goals..."}],
    completion="The Q3 OKRs contain goals for multiple teams...",
    usage=Usage(promptTokens=50, completionTokens = 49),
    metadata={"interface": "whatsapp"}
))

TRACE
|
|-- SPAN: Retrieval
|   |
|   |-- GENERATION: Vector DB Query Creation
|   |
|   |-- SPAN: Data Fetching
|   |
|   |-- EVENT: Data Summary Creation
|
|-- GENERATION: Output Generation

from langfuse.model import CreateTrace, CreateSpan, CreateGeneration, CreateEvent
 
trace = langfuse.trace(CreateTrace(name = "llm-feature"))
retrieval = trace.span(CreateSpan(name = "retrieval"))
retrieval.generation(CreateGeneration(name = "query-creation"))
retrieval.span(CreateSpan(name = "vector-db-search"))
retrieval.event(CreateEvent(name = "db-summary"))
trace.generation(CreateGeneration(name = "user-output"));

from langfuse.model import CreateTrace
 
trace = langfuse.trace(CreateTrace(
    name = "docs-retrieval",
    userId = "user__935d7d1d-8625-4ef4-8651-544613e7bd22",
    metadata = {
        "env": "production",
        "email": "user@langfuse.com",
    }
))

from datetime import datetime
from langfuse.model import CreateSpan, UpdateSpan
 
retrievalStartTime = datetime.now()
 
# retrieveDocs = retrieveDoc()
# ...
 
span = trace.span(CreateSpan(
        name="embedding-search",
        startTime=retrievalStartTime,
        endTime=datetime.now(),
        metadata={"database": "pinecone"},
        input = {'query': 'This document entails the OKR goals for ACME'},
    )
)

span = span.update(UpdateSpan(
        output = {"response": "[{'name': 'OKR Engineering', 'content': 'The engineering department defined the following OKR goals...'},{'name': 'OKR Marketing', 'content': 'The marketing department defined the following OKR goals...'}]"}
    )
)

from langfuse.model import CreateGeneration, Usage, UpdateGeneration
 
from datetime import datetime
 
generationStartTime = datetime.now()
 
# chat_completion = openai.ChatCompletion.create(model="gpt-3.5-turbo", messages=[{"role": "user", "content": "Hello world"}])
# ...
 
generation = trace.generation(CreateGeneration(
    name="summary-generation",
    startTime=generationStartTime,
    endTime=datetime.now(),
    model="gpt-3.5-turbo",
    modelParameters={"maxTokens": "1000", "temperature": "0.9"},
    prompt=[{"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Please generate a summary of the following documents \nThe engineering department defined the following OKR goals...\nThe marketing department defined the following OKR goals..."}],
    metadata={"interface": "whatsapp"}
))

generation.update(UpdateGeneration(
    completion="The Q3 OKRs contain goals for multiple teams...",
    usage=Usage(promptTokens=50, completionTokens = 49),
));

from langfuse.model import CreateEvent
from datetime import datetime
 
event = span.event(CreateEvent(
        name="chat-docs-retrieval",
        startTime=datetime.now(),
        metadata={"key": "value"},
        input = {"key": "value"},
        output = {"key": "value"}
    )
)

from langfuse.model import CreateScore, InitialScore
 
# via {trace, span, event, generation}.score
trace.score(CreateScore(
    name="user-explicit-feedback",
    value=1,
    comment="I like how personalized the response is"
));
 
# using the trace_id
trace_id = trace.id
langfuse.score(InitialScore(
    traceId=trace.id,
    name="user-explicit-feedback",
    value=1,
    comment="I like how personalized the response is"
));

langfuse.flush()

# The SDK will automatically include the env variable.
os.environ["LANGFUSE_RELEASE"] = "ba7816b..." # <- example, github sha
 
# Alternatively, use the constructor of the SDK
langfuse = Langfuse(release='ba7816b')

langfuse.span(CreateSpan(name = "retrieval", version="<version>"));

langfuse = Langfuse(debug=True)
 
# Deactivating for the rest of the notebook
langfuse = Langfuse()

%pip install fastapi --upgrade

from contextlib import asynccontextmanager
from fastapi import FastAPI, Query, BackgroundTasks
from langfuse.model import InitialGeneration
 
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    # Operation on startup
 
    yield  # wait until shutdown
 
    # Flush all events to be sent to Langfuse on shutdown and terminate all Threads gracefully. This operation is blocking.
    langfuse.flush()
 
 
app = FastAPI(lifespan=lifespan)

langfuse = Langfuse()
 
@app.get("/generate/",tags=["APIs"])
async def campaign(prompt: str = Query(..., max_length=20)):
  # call to a LLM
  generation = langfuse.generation(
      InitialGeneration(name="llm-feature", metadata="test", prompt=prompt)
  )
  return True