from langfuse.decoratorsimport observefrom langfuse.openaiimport openai# OpenAI integration@observe()def story():    return openai.chat.completions.create(        model="gpt-4o",        messages=[          {"role":"system","content":"You are a great storyteller."},          {"role":"user","content":"Once upon a time in a galaxy far, far away..."}        ],    ).choices[0].message.content@observe()def main():    return story()main()

Install the Langfuse Python SDK

pip install "langfuse<3.0.0"

Add Langfuse API keys

If you haven’t done so yet,sign up to Langfuse and obtain your API keys from the project settings. Alternatively, you can also run Langfuse locally or self-host.

import osos.environ["LANGFUSE_PUBLIC_KEY"]= "pk-lf-..."os.environ["LANGFUSE_SECRET_KEY"]= "sk-lf-..."os.environ["LANGFUSE_HOST"]= "https://cloud.langfuse.com" # 🇪🇺 EU region# os.environ["LANGFUSE_HOST"] = "https://us.cloud.langfuse.com" # 🇺🇸 US region

LANGFUSE_SECRET_KEY = "sk-lf-..."LANGFUSE_PUBLIC_KEY = "pk-lf-..."LANGFUSE_HOST = "https://cloud.langfuse.com" # 🇪🇺 EU region# LANGFUSE_HOST = "https://us.cloud.langfuse.com" # 🇺🇸 US region

from langfuse.decoratorsimport langfuse_contextlangfuse_context.configure(    secret_key="sk-lf-...",    public_key="pk-lf-...",    host="https://cloud.langfuse.com",# 🇪🇺 EU region    # host="https://us.cloud.langfuse.com" # 🇺🇸 US region)

When no API keys are provided, a single warning is logged, and no traces are sent to Langfuse.

Add the Langfuse decorator

Import the@observe() decorator and apply it to the functions you want to trace. By default it captures:

• nesting via context vars
• timings/durations
• function name
• args and kwargs as input dict
• returned values as output

The decorator will automatically create a trace for the top-level function and spans for any nested functions. Learn more about the tracing data modelhere.

from langfuse.decoratorsimport observe@observe()def fn():    pass@observe()def main():    fn()main()

from langfuse.decoratorsimport observe,langfuse_context@observe()def main():    print("Hello, from the main function!")main()langfuse_context.flush()

from langfuse.decoratorsimport observe, langfuse_contextimport anthropicanthopic_client= anthropic.Anthropic()# Wrap LLM function with decorator@observe(as_type="generation")def anthropic_completion(**kwargs):  # optional, 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  )  response= anthopic_client.messages.create(**kwargs)  # See docs for more details on token counts and usd cost in Langfuse  # https://langfuse.com/docs/model-usage-and-cost  langfuse_context.update_current_observation(      usage_details={          "input": response.usage.input_tokens,          "output": response.usage.output_tokens      }  )  # return result  return response.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.decoratorsimport observe@observe(capture_input=False,capture_output=False)def fn(secret_arg):    return "super secret output"fn("my secret arg")

from langfuse.decoratorsimport langfuse_context, observe@observe()def fn(secret_arg):langfuse_context.update_current_observation(        input="sanitized input",# any serializable object        output="sanitized output",# any serializable object    )    return "super secret output"fn("my secret arg")

from langfuse.decoratorsimport langfuse_context# Configure the Langfuse clientlangfuse_context.configure(    secret_key="sk-lf-...",    public_key="pk-lf-...",    httpx_client=custom_httpx_client,    host=custom_host,    enabled=True,)

from langfuse.decoratorsimport langfuse_context, observe@observe()def deeply_nested_fn():    # Enrich the current observation with a custom name, input, and output    # All of these parameters override the default values captured by the decoratorlangfuse_context.update_current_observation(        name="Deeply nested LLM call",        input="Ping?",        output="Pong!"    )    # Updates the trace, overriding the default trace name `main` (function name)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    )    return "output" # This output will not be captured as we have overridden it@observe()def nested_fn():    # Update the current span with a custom name and level    # Overrides the default span namelangfuse_context.update_current_observation(        name="Nested Span",        level="WARNING"    )    deeply_nested_fn()@observe()def main():    # This will be the trace as it is the highest level function    nested_fn()# Execute the main function to generate the enriched tracemain()

from langfuse.decoratorsimport langfuse_context, observe@observe(as_type="generation")def nested_generation():    prompt= langfuse.get_prompt("movie-critic")    langfuse_context.update_current_observation(        prompt=prompt,    )@observe()def main():  nested_generation()main()

from langfuse.decoratorsimport langfuse_context, observe@observe()def main():    print(langfuse_context.get_current_trace_url())main()

from langfuse.decoratorsimport langfuse_context, observe@observe()def fn():    print(langfuse_context.get_current_trace_id())    print(langfuse_context.get_current_observation_id())fn()

from langfuse.decoratorsimport langfuse_context, observe@observe()def process_user_request(user_id, request_data,**kwargs):    # Function logic here    pass@observe(**kwargs)def main():    process_user_request(        "user_id",        "request",langfuse_observation_id="my-custom-request-id",    )main(langfuse_observation_id="my-custom-request-id")

from langfuse.decoratorsimport langfuse_context, observe@observe()def process_user_request(user_id, request_data,**kwargs):    # Function logic here    pass@observe(**kwargs)def main():    process_user_request(        "user_id",        "request",        langfuse_observation_id="my-custom-request-id",    )# Set a parent trace idmain(    langfuse_observation_id="my-custom-request-id",langfuse_parent_trace_id="some_existing_trace_id")# Set a parent span id. Note that parent_trace_id must also be passed.main(    langfuse_observation_id="my-custom-request-id",langfuse_parent_trace_id="some_existing_trace_id",langfuse_parent_observation_id="some_existing_span_id",)

for itemin dataset.items:    # Make sure your application function is decorated with @observe decorator to automatically link the trace    with item.observe(        run_name="<run_name>",        run_description="My first run",        run_metadata={"model":"llama3"},    )as trace_id:        # run your @observe() decorated application on the dataset item input        output= my_llm_application.run(item.input)        # optionally, evaluate the output to compare different runs more easily        langfuse.score(            trace_id=trace_id,            name="<example_eval>",            # any float value            value=my_eval_fn(item.input, output, item.expected_output),            comment="This is a comment",# optional, useful to add reasoning        )# Flush the langfuse client to ensure all data is sent to the server at the end of the experiment runlangfuse.flush()

from langfuse.decoratorsimport observefrom langfuse.openaiimportopenai@observe()def story():    returnopenai.chat.completions.create(        model="gpt-3.5-turbo",        max_tokens=100,        messages=[          {"role":"system","content":"You are a great storyteller."},          {"role":"user","content":"Once upon a time in a galaxy far, far away..."}        ],    ).choices[0].message.content@observe()def main():    return story()main()

from operatorimport itemgetterfrom langchain_openaiimport ChatOpenAIfrom langchain.promptsimport ChatPromptTemplatefrom langchain.schemaimport StrOutputParserfrom langfuse.decoratorsimport langfuse_context, observeprompt= 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 contextlangfuse_handler= langfuse_context.get_current_langchain_handler()    # Pass handler to invoke of your langchain chain/agent    chain.invoke({"person": person},config={"callbacks":[langfuse_handler]})langchain_fn("John Doe")

from langfuse.decoratorsimport langfuse_context, observefrom llama_index.coreimport Document, VectorStoreIndexfrom llama_index.coreimport Settingsfrom llama_index.core.callbacksimport CallbackManagerdoc1= 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 functionlangfuse_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 concurrent.futuresimport ThreadPoolExecutor, as_completedfrom langfuse.decoratorsimport langfuse_context, observe@observe()def execute_task(*args):    return args@observe()def execute_groups(task_args):    trace_id= langfuse_context.get_current_trace_id()    observation_id= langfuse_context.get_current_observation_id()    with ThreadPoolExecutor(3)as executor:        futures= [            executor.submit(                execute_task,                *task_arg,langfuse_parent_trace_id=trace_id,langfuse_parent_observation_id=observation_id,            )            for task_argin task_args        ]        for futurein as_completed(futures):            future.result()    return [f.result()for fin futures]@observe()def main():    task_args= [["a","b"], ["c","d"]]    execute_groups(task_args)main()langfuse_context.flush()

from concurrent.futuresimport ThreadPoolExecutor, as_completedfrom contextvarsimport copy_contextfrom langfuse.decoratorsimport langfuse_context, observe@observe()def execute_task(*args):    return argstask_args= [["a","b"], ["c","d"]]@observe()def execute_groups(task_args):    with ThreadPoolExecutor(3)as executor:        futures= []        for task_argin task_args:            ctx= copy_context()            task= lambda p=task_arg, ctx=ctx: ctx.run(execute_task,*p)            futures.append(executor.submit(task))    return [f.result()for fin as_completed(futures)]execute_groups(task_args)langfuse_context.flush()