Langfuse for Enterprise AI: Architecture & Integration

import os
from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context
from openai import OpenAI

# Initialize Langfuse client with environment credentials
langfuse = Langfuse(
  public_key=os.environ["LANGFUSE_PUBLIC_KEY"],
  secret_key=os.environ["LANGFUSE_SECRET_KEY"],
  host=os.environ.get("LANGFUSE_HOST", "https://cloud.langfuse.com")
)

client = OpenAI()

@observe(name="enterprise_rag_query")
def execute_rag_pipeline(user_query: str):
  # Fetch managed prompt version from Langfuse registry
  prompt = langfuse.get_prompt("rag_system_prompt", version=1)
  compiled_prompt = prompt.compile(query=user_query)

  # Attach custom metadata to the active trace
  langfuse_context.update_current_trace(
      user_id="usr_94820",
      tags=["production", "financial_rag"],
      metadata={"environment": "k8s_prod"}
  )

  # Execute LLM completion with automatic span tracking
  response = client.chat.completions.create(
      model="gpt-4o",
      messages=[{"role": "user", "content": compiled_prompt}],
      temperature=0.2
  )
  answer = response.choices[0].message.content

  # Log evaluation score to active span
  langfuse_context.score_current_observation(
      name="user_feedback",
      value=1.0,
      comment="Accurate answer generated"
  )

  return answer

if __name__ == "__main__":
  result = execute_rag_pipeline("What is the Q3 revenue target?")
  langfuse.flush()
  print("Answer:", result)