import subprocess, sys\ndef pip_install(*pkgs):\n subprocess.run([sys.executable, \"-m\", \"pip\", \"install\", \"-qU\", *pkgs], check=True)\npip_install(\"langfuse\", \"openai\")\nimport os\nfrom getpass import getpass\ndef _ask(var, prompt, secret=True, default=None):\n if os.environ.get(var):\n return os.environ[var]\n val = (getpass(prompt) if secret else input(prompt)).strip()\n if not val and default is not None:\n val = default\n os.environ[var] = val\n return val\nprint(\"Enter your Langfuse credentials (input is hidden):\")\n_ask(\"LANGFUSE_PUBLIC_KEY\", \" Langfuse PUBLIC key (pk-lf-...): \")\n_ask(\"LANGFUSE_SECRET_KEY\", \" Langfuse SECRET key (sk-lf-...): \")\nregion = (input(\" Region \u2014 EU (default) \/ US \/ or paste a self-hosted URL: \")\n .strip().lower())\nif region.startswith(\"http\"):\n HOST = region\nelif region in (\"2\", \"us\"):\n HOST = \"https:\/\/us.cloud.langfuse.com\"\nelse:\n HOST = \"https:\/\/cloud.langfuse.com\"\nos.environ[\"LANGFUSE_HOST\"] = HOST\nOPENAI_API_KEY = (getpass(\" OpenAI key (optional, press Enter to skip): \").strip())\nif OPENAI_API_KEY:\n os.environ[\"OPENAI_API_KEY\"] = OPENAI_API_KEY\nUSE_OPENAI = bool(OPENAI_API_KEY)\nDEFAULT_MODEL = \"gpt-4o-mini\" if USE_OPENAI else \"mock-llm-v1\"\nfrom langfuse import get_client, observe, propagate_attributes, Evaluation\nlangfuse = get_client()\nassert langfuse.auth_check(), \"Auth failed \u2014 double-check keys\/region.\"\nprint(f\"n![\"\u2705\"](\"https:\/\/s.w.org\/images\/core\/emoji\/17.0.2\/72x72\/2705.png\")
Connected to Langfuse at {HOST}\")\nprint(f\" LLM backend: {'OpenAI (' + DEFAULT_MODEL + ')' if USE_OPENAI else 'built-in mock'}n\")<\/code><\/pre>\n<\/div>\n<\/div>\nWe begin by installing the required Langfuse and OpenAI packages inside the Colab environment. We then collect Langfuse credentials, choose the correct Langfuse region or self-hosted URL, and optionally accept an OpenAI API key. We finally initialize the Langfuse client, verify authentication, and confirm whether we are using OpenAI or the built-in mock LLM.<\/p>\n
\n\n\n\n<\/div>\n<\/div>\n<\/div>\n<\/div>\nCopy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\nif USE_OPENAI:\n from langfuse.openai import openai\n_MOCK_FACTS = {\n \"france\": \"Paris\", \"germany\": \"Berlin\", \"japan\": \"Tokyo\",\n \"italy\": \"Rome\", \"spain\": \"Madrid\", \"india\": \"New Delhi\",\n}\ndef _mock_answer(user_text: str) -> str:\n t = user_text.lower()\n for country, capital in _MOCK_FACTS.items():\n if country in t:\n return capital\n if \"langfuse\" in t:\n return (\"Langfuse is an open-source LLM engineering platform for \"\n \"observability, prompt management, evaluation and datasets.\")\n return \"This is a mock response. Provide an OpenAI key for real generations.\"\ndef llm_chat(messages, *, model=DEFAULT_MODEL, temperature=0.3,\n name=None, langfuse_prompt=None) -> str:\n \"\"\"Return assistant text; the call is traced as a Langfuse generation.\"\"\"\n if USE_OPENAI:\n kwargs = dict(model=model, messages=messages, temperature=temperature)\n if name: kwargs[\"name\"] = name\n if langfuse_prompt: kwargs[\"langfuse_prompt\"] = langfuse_prompt\n resp = openai.chat.completions.create(**kwargs)\n return resp.choices[0].message.content\n last_user = next((m[\"content\"] for m in reversed(messages)\n if m[\"role\"] == \"user\"), \"\")\n answer = _mock_answer(last_user)\n gen_kwargs = dict(as_type=\"generation\", name=name or \"mock-llm\",\n model=model, input=messages)\n if langfuse_prompt is not None:\n gen_kwargs[\"prompt\"] = langfuse_prompt\n with langfuse.start_as_current_observation(**gen_kwargs) as gen:\n gen.update(output=answer,\n usage_details={\"input_tokens\": 24, \"output_tokens\": 12})\n return answer\nprint(\"PART 1 \u2500\u2500 Decorator tracing -------------------------------------------\")\n@observe()\ndef write_story(topic: str) -> str:\n return llm_chat(\n [{\"role\": \"user\", \"content\": f\"Write a one-sentence story about {topic}.\"}],\n name=\"story-generation\",\n )\n@observe()\ndef story_pipeline(topic: str) -> str:\n return write_story(topic)\nprint(\" \u2192\", story_pipeline(\"a debugging robot\"))<\/code><\/pre>\n<\/div>\n<\/div>\nWe define the LLM helper that supports both real OpenAI generations and deterministic mock responses. We also make sure that even the mock path creates a proper Langfuse generation observation, so the tutorial remains fully traceable without an OpenAI key. We then demonstrate basic decorator-based tracing by wrapping a simple story-generation pipeline with @observe.<\/p>\n
\n\n\n\n<\/div>\n<\/div>\n<\/div>\n<\/div>\nCopy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\nprint(\"nPART 2 \u2500\u2500 Manual RAG trace --------------------------------------------\")\n_KB = {\n \"refund\": \"Refunds are processed within 5\u20137 business days to the original method.\",\n \"warranty\": \"All products carry a 1-year limited manufacturer warranty.\",\n}\n@observe(name=\"retrieve\")\ndef retrieve(question: str):\n q = question.lower()\n hits = [v for k, v in _KB.items() if k in q] or list(_KB.values())\n return hits[:2]\n@observe(name=\"rag-pipeline\")\ndef rag_pipeline(question: str, user_id=\"user-42\", session_id=\"sess-001\") -> str:\n with propagate_attributes(user_id=user_id, session_id=session_id,\n tags=[\"rag\", \"support-bot\", \"tutorial\"]):\n context = \"n\".join(retrieve(question))\n return llm_chat(\n [{\"role\": \"system\",\n \"content\": \"Answer the question using ONLY the provided context.\"},\n {\"role\": \"user\",\n \"content\": f\"Context:n{context}nnQuestion: {question}\"}],\n name=\"rag-answer\",\n )\nrag_answer = rag_pipeline(\"How long do refunds take?\")\nrag_trace_id = langfuse.get_current_trace_id()\nprint(\" \u2192\", rag_answer)<\/code><\/pre>\n<\/div>\n<\/div>\nWe build a small manual RAG pipeline using a simple in-memory knowledge base for refunds, shipping, and warranty information. We trace the retrieval step separately and use propagate_attributes to attach user ID, session ID, and tags across the full trace. We then run a refund-related question and capture the trace ID so we can attach scores to it later.<\/p>\n
\n\n\n\n<\/div>\n<\/div>\n<\/div>\n<\/div>\nCopy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\nprint(\"nPART 3 \u2500\u2500 Prompt management -------------------------------------------\")\nlangfuse.create_prompt(\n name=\"support-agent\",\n type=\"chat\",\n prompt=[\n {\"role\": \"system\",\n \"content\": \"You are a {{tone}} customer-support agent for {{company}}. \"\n \"Be concise.\"},\n {\"role\": \"user\", \"content\": \"{{question}}\"},\n ],\n labels=[\"production\"],\n config={\"model\": DEFAULT_MODEL, \"temperature\": 0.2},\n)\nprompt = langfuse.get_prompt(\"support-agent\", type=\"chat\")\ncompiled = prompt.compile(tone=\"friendly\", company=\"Acme\",\n question=\"Do you offer express shipping?\")\nprint(\" compiled prompt:\", compiled)\n@observe(name=\"prompt-managed-call\")\ndef answer_with_managed_prompt():\n return llm_chat(compiled, name=\"support-reply\", langfuse_prompt=prompt)\nprint(\" \u2192\", answer_with_managed_prompt())\nprint(\"nPART 4 \u2500\u2500 Scoring -----------------------------------------------------\")\ndef keyword_overlap(answer: str, expected_keyword: str) -> float:\n return 1.0 if expected_keyword.lower() in (answer or \"\").lower() else 0.0\nlangfuse.create_score(\n name=\"groundedness\",\n value=keyword_overlap(rag_answer, \"5\"),\n trace_id=rag_trace_id,\n data_type=\"NUMERIC\",\n comment=\"Heuristic: mentions the documented refund window.\",\n)\nlangfuse.create_score(name=\"user_feedback\", value=\"helpful\",\n trace_id=rag_trace_id, data_type=\"CATEGORICAL\")\nlangfuse.create_score(name=\"resolved\", value=1,\n trace_id=rag_trace_id, data_type=\"BOOLEAN\")\n@observe(name=\"scored-call\")\ndef scored_call():\n out = llm_chat([{\"role\": \"user\", \"content\": \"What is the capital of Japan?\"}],\n name=\"capital-q\")\n with langfuse.start_as_current_observation(as_type=\"span\", name=\"grade\") as span:\n span.score(name=\"correct\", value=keyword_overlap(out, \"Tokyo\"),\n data_type=\"NUMERIC\")\n span.score_trace(name=\"trace_quality\", value=0.9, data_type=\"NUMERIC\")\n return out\nprint(\" \u2192\", scored_call(), \"(scores attached)\")<\/code><\/pre>\n<\/div>\n<\/div>\nWe create a managed Langfuse chat prompt, compile it with runtime variables, and link the prompt version to a traced generation. We then add different score types to the earlier RAG trace, including numeric, categorical, and boolean scores. We also demonstrate inline scoring by grading a capital-city answer inside the current observed span and trace.<\/p>\n
\n\n\n\n<\/div>\n<\/div>\n<\/div>\n<\/div>\nCopy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\nprint(\"nPART 5 \u2500\u2500 Datasets & experiments --------------------------------------\")\nDATASET = \"capital-cities-tutorial\"\nlangfuse.create_dataset(name=DATASET, description=\"Capital-city QA benchmark\")\n_items = [\n (\"What is the capital of France?\", \"Paris\"),\n (\"What is the capital of Germany?\", \"Berlin\"),\n (\"What is the capital of Japan?\", \"Tokyo\"),\n (\"What is the capital of Italy?\", \"Rome\"),\n]\nfor i, (q, a) in enumerate(_items):\n langfuse.create_dataset_item(dataset_name=DATASET, id=f\"cap-{i}\",\n input={\"question\": q}, expected_output=a)\ndef capital_task(*, item, **kwargs):\n question = item.input[\"question\"] if isinstance(item.input, dict) else item.input\n return llm_chat([{\"role\": \"user\", \"content\": question}], name=\"experiment-answer\")\ndef accuracy(*, input, output, expected_output, metadata=None, **kwargs):\n hit = bool(expected_output) and expected_output.lower() in (output or \"\").lower()\n return Evaluation(name=\"accuracy\", value=1.0 if hit else 0.0,\n comment=\"exact-match contains check\")\ndef conciseness(*, input, output, **kwargs):\n return Evaluation(name=\"char_length\", value=float(len(output or \"\")))\ndef mean_accuracy(*, item_results, **kwargs):\n vals = [e.value for r in item_results for e in r.evaluations if e.name == \"accuracy\"]\n avg = sum(vals) \/ len(vals) if vals else 0.0\n return Evaluation(name=\"mean_accuracy\", value=avg, comment=f\"{avg:.0%} correct\")\ndataset = langfuse.get_dataset(DATASET)\nresult = dataset.run_experiment(\n name=\"capitals-baseline\",\n description=\"Baseline run from the Colab tutorial\",\n task=capital_task,\n evaluators=[accuracy, conciseness],\n run_evaluators=[mean_accuracy],\n max_concurrency=4,\n)\nprint(result.format())<\/code><\/pre>\n<\/div>\n<\/div>\nWe create a Langfuse dataset for capital-city questions and add deterministic items to ensure repeated runs remain idempotent. We define a task function that answers each item, along with item-level evaluators for accuracy and response length. We then run an experiment on the dataset and print a formatted summary of item-level and aggregate results.<\/p>\n
\n\n\n\n<\/div>\n<\/div>\n<\/div>\n<\/div>\nCopy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\nif USE_OPENAI:\n print(\"nPART 6 \u2500\u2500 LangChain integration ---------------------------------------\")\n pip_install(\"langchain-core\", \"langchain-openai\")\n from langchain_openai import ChatOpenAI\n from langchain_core.prompts import ChatPromptTemplate\n from langfuse.langchain import CallbackHandler\n handler = CallbackHandler()\n chain = (ChatPromptTemplate.from_template(\"Explain {concept} in one sentence.\")\n | ChatOpenAI(model=\"gpt-4o-mini\", temperature=0))\n lc_out = chain.invoke({\"concept\": \"observability\"},\n config={\"callbacks\": [handler]})\n print(\" \u2192\", lc_out.content)\nelse:\n print(\"nPART 6 \u2500\u2500 LangChain integration skipped (no OpenAI key).\")\nlangfuse.flush()\nprint(\"Open your project at\", HOST)\nprint(\" \u2022 Tracing tab .... Parts 1\u20134 traces (incl. user\/session\/tags)\")\nprint(\" \u2022 Prompts tab .... the versioned 'support-agent' prompt\")\nprint(\" \u2022 Scores ......... groundedness \/ user_feedback \/ resolved \/ accuracy\")\nprint(\" \u2022 Datasets tab ... '%s' with the 'capitals-baseline' experiment run\" % DATASET)<\/code><\/pre>\n<\/div>\n<\/div>\nWe optionally demonstrate the LangChain integration when an OpenAI key is available, using the Langfuse callback handler to trace chain execution. If no OpenAI key is provided, we skip this section while keeping the rest of the tutorial fully functional. We finally flush all buffered events to Langfuse and print where to inspect traces, prompts, scores, and dataset experiment results.<\/p>\n
In conclusion, we created a practical end-to-end Langfuse workflow that covers the most important parts of LLM observability and evaluation. We learned how to trace both automatic and manual operations, link prompt versions to generations, score outputs, and benchmark an application using datasets and experiments. We also kept the tutorial flexible by supporting both OpenAI-powered generation and a mock LLM path, making it easier to test the full pipeline in any environment. Also, we gained an understanding of how Langfuse helps us monitor LLM behavior, compare experiment runs, and build more reliable AI applications.<\/p>\n
\n
\n<\/p>\n<\/p>
Check out\u00a0the\u00a0Full Codes with Notebook here<\/strong><\/a>.\u00a0<\/strong>Also,\u00a0feel free to follow us on\u00a0Twitter<\/mark><\/a><\/strong>\u00a0and don\u2019t forget to join our\u00a0150k+ ML SubReddit<\/a><\/strong>\u00a0and Subscribe to\u00a0our Newsletter<\/a><\/strong>. Wait! are you on telegram?\u00a0now you can join us on telegram as well.<\/a><\/strong><\/p>\nNeed to partner with us for promoting your GitHub Repo OR Hugging Face Page OR Product Release OR Webinar etc.?\u00a0Connect with us<\/mark><\/a><\/strong><\/p>\nThe post Build a Complete Langfuse Observability and Evaluation Pipeline for Tracing, Prompt Management, Scoring, and Experiments<\/a> appeared first on MarkTechPost<\/a>.<\/p>","protected":false},"excerpt":{"rendered":"In this tutorial, we implement…<\/p>\n","protected":false},"author":1,"featured_media":29,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-972","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-uncategorized"],"_links":{"self":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/972","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=972"}],"version-history":[{"count":0,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/972\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/media\/29"}],"wp:attachment":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=972"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=972"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=972"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}