Cookbook: Python Decorators

%pip install langfuse

import os
 
# Get keys for your project from the project settings page
# https://cloud.langfuse.com
os.environ["LANGFUSE_PUBLIC_KEY"] = ""
os.environ["LANGFUSE_SECRET_KEY"] = ""
os.environ["LANGFUSE_HOST"] = "https://cloud.langfuse.com" # 🇪🇺 EU region
# os.environ["LANGFUSE_HOST"] = "https://us.cloud.langfuse.com" # 🇺🇸 US region
 
# Your openai key
os.environ["OPENAI_API_KEY"] = ""

from langfuse.decorators import langfuse_context, observe
import time
 
@observe()
def wait():
    time.sleep(1)
 
@observe()
def capitalize(input: str):
    return input.upper()
 
@observe()
def main_fn(query: str):
    wait()
    capitalized = capitalize(query)
    return f"Q:{capitalized}; A: nice too meet you!"
 
main_fn("hi there");

from langfuse.decorators import langfuse_context, observe
 
@observe(as_type="generation")
def deeply_nested_llm_call():
    # Enrich the current observation with a custom name, input, and output
    langfuse_context.update_current_observation(
        name="Deeply nested LLM call", input="Ping?", output="Pong!"
    )
    # Set the parent trace's name from within a nested observation
    langfuse_context.update_current_trace(
        name="Trace name set from deeply_nested_llm_call",
        session_id="1234",
        user_id="5678",
        tags=["tag1", "tag2"],
        public=True
    )
 
@observe()
def nested_span():
    # Update the current span with a custom name and level
    langfuse_context.update_current_observation(name="Nested Span", level="WARNING")
    deeply_nested_llm_call()
 
@observe()
def main():
    nested_span()
 
# Execute the main function to generate the enriched trace
main()

%pip install anthropic

os.environ["ANTHROPIC_API_KEY"] = ""
 
import anthropic
anthopic_client = anthropic.Anthropic()

# Wrap LLM function with decorator
@observe(as_type="generation")
def anthropic_completion(**kwargs):
  # extract some fields from kwargs
  kwargs_clone = kwargs.copy()
  input = kwargs_clone.pop('messages', None)
  model = kwargs_clone.pop('model', None)
  langfuse_context.update_current_observation(
      input=input,
      model=model,
      metadata=kwargs_clone
  )
  
  # return result
  return anthopic_client.messages.create(**kwargs).content[0].text
 
@observe()
def main():
  return anthropic_completion(
      model="claude-3-opus-20240229",
      max_tokens=1024,
      messages=[
          {"role": "user", "content": "Hello, Claude"}
      ]
  )
 
main()

from langfuse.decorators import observe
 
@observe(capture_input=False, capture_output=False)
def stealth_fn(input: str):
    return input
 
stealth_fn("Super secret content")

from langfuse.decorators import langfuse_context, observe
 
@observe()
def fn_2():
    langfuse_context.update_current_observation(
        input="Table?", output="Tennis!"
    )
    # Logic for a deeply nested LLM call
    pass
 
@observe()
def main_fn():
    langfuse_context.update_current_observation(
        input="Ping?", output="Pong!"
    )
    fn_2()
 
main_fn()

%pip install llama-index langchain langchain_openai --upgrade

from langfuse.openai import openai
from langfuse.decorators import observe
 
@observe()
def openai_fn(calc: str):
    res = openai.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[
          {"role": "system", "content": "You are a very accurate calculator. You output only the result of the calculation."},
          {"role": "user", "content": calc}],
    )
    return res.choices[0].message.content

from langfuse.decorators import langfuse_context, observe
from llama_index.core import Document, VectorStoreIndex
from llama_index.core import Settings
from llama_index.core.callbacks import CallbackManager
 
doc1 = Document(text="""
Maxwell "Max" Silverstein, a lauded movie director, screenwriter, and producer, was born on October 25, 1978, in Boston, Massachusetts. A film enthusiast from a young age, his journey began with home movies shot on a Super 8 camera. His passion led him to the University of Southern California (USC), majoring in Film Production. Eventually, he started his career as an assistant director at Paramount Pictures. Silverstein's directorial debut, “Doors Unseen,” a psychological thriller, earned him recognition at the Sundance Film Festival and marked the beginning of a successful directing career.
""")
doc2 = Document(text="""
Throughout his career, Silverstein has been celebrated for his diverse range of filmography and unique narrative technique. He masterfully blends suspense, human emotion, and subtle humor in his storylines. Among his notable works are "Fleeting Echoes," "Halcyon Dusk," and the Academy Award-winning sci-fi epic, "Event Horizon's Brink." His contribution to cinema revolves around examining human nature, the complexity of relationships, and probing reality and perception. Off-camera, he is a dedicated philanthropist living in Los Angeles with his wife and two children.
""")
 
@observe()
def llama_index_fn(question: str):
    # Set callback manager for LlamaIndex, will apply to all LlamaIndex executions in this function
    langfuse_handler = langfuse_context.get_current_llama_index_handler()
    Settings.callback_manager = CallbackManager([langfuse_handler])
 
    # Run application
    index = VectorStoreIndex.from_documents([doc1,doc2])
    response = index.as_query_engine().query(question)
    return response

from operator import itemgetter
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema import StrOutputParser
from langfuse.decorators import observe
 
prompt = ChatPromptTemplate.from_template("what is the city {person} is from?")
model = ChatOpenAI()
chain = prompt | model | StrOutputParser()
 
@observe()
def langchain_fn(person: str):
    # Get Langchain Callback Handler scoped to the current trace context
    langfuse_handler = langfuse_context.get_current_langchain_handler()
 
    # Pass handler to invoke
    chain.invoke({"person": person}, config={"callbacks":[langfuse_handler]})

from langfuse.decorators import observe
 
@observe()
def main():
    output_openai = openai_fn("5+7")
    output_llamaindex = llama_index_fn("What did he do growing up?")
    output_langchain = langchain_fn("Feynman")
 
    return output_openai, output_llamaindex, output_langchain
 
main();

from langfuse.decorators import langfuse_context, observe
 
@observe()
def nested_span():
    langfuse_context.score_current_observation(
        name="feedback-on-span",
        value=1,
        comment="I like how personalized the response is",
    )
 
    langfuse_context.score_current_trace(
        name="feedback-on-trace-from-nested-span",
        value=1,
        comment="I like how personalized the response is",
    )
 
 
# This will create a new trace
@observe()
def main():
    langfuse_context.score_current_trace(
        name="feedback-on-trace",
        value=1,
        comment="I like how personalized the response is",
    )
    nested_span()
 
main()

from langfuse import Langfuse
from langfuse.decorators import langfuse_context, observe
 
# Initialize the Langfuse client
langfuse_client = Langfuse()
 
@observe()
def nested_fn():
    span_id = langfuse_context.get_current_observation_id()
 
    # can also be accessed in main
    trace_id = langfuse_context.get_current_trace_id()
 
    return "foo_bar", trace_id, span_id
 
# Create a new trace
@observe()
def main():
 
    _, trace_id, span_id = nested_fn()
 
    return "main_result", trace_id, span_id
 
 
# Flush the trace to send it to the Langfuse platform
langfuse_context.flush()
 
# Execute the main function to generate a trace
_, trace_id, span_id = main()
 
# Score the trace from outside the trace context
langfuse_client.score(
    trace_id=trace_id,
    name="trace-score",
    value=1,
    comment="I like how personalized the response is"
)
 
# Score the specific span/function from outside the trace context
langfuse_client.score(
    trace_id=trace_id,
    observation_id=span_id,
    name="span-score",
    value=1,
    comment="I like how personalized the response is"
);

from langfuse.decorators import langfuse_context, observe
import uuid
 
@observe()
def process_user_request(user_id, request_data, **kwargs):
    # Function logic here
    pass
 
 
def main():
    user_id = "user123"
    request_data = {"action": "login"}
 
    # Custom ID for tracking
    custom_observation_id = "custom-" + str(uuid.uuid4())
 
    # Pass id as kwarg
    process_user_request(
        user_id=user_id,
        request_data=request_data,
        # Pass the custom observation ID to the function
        langfuse_observation_id=custom_observation_id,
    )
 
main()

from langfuse.decorators import langfuse_context
 
assert langfuse_context.auth_check()