{
"$type": "site.standard.document",
"bskyPostRef": {
"cid": "bafyreih6y4tnpyxinmr3yxl34h4yoga5x5rx7ti5chtfnkphv3b3nb3coa",
"uri": "at://did:plc:25rdn5elo5izoxrmtis34zuk/app.bsky.feed.post/3mpszarj6pb32"
},
"coverImage": {
"$type": "blob",
"ref": {
"$link": "bafkreig3jpvezbtgdo3ep5dmf6bpi57ucca3manna4iwvcttatdxlusawu"
},
"mimeType": "image/webp",
"size": 67060
},
"path": "/hiroki-kameyama/observability-tracing-rag-and-agents-with-langfuse-v4-5hc",
"publishedAt": "2026-07-04T11:53:27.000Z",
"site": "https://dev.to",
"tags": [
"ai",
"mlops",
"llm",
"python",
"Chapter 2 (Evals)",
"cloud.langfuse.com",
"@observe"
],
"textContent": "## Introduction\n\nIn Chapter 2 (Evals), we measured answer _quality_. Now we add Observability — making behavior _visible_.\n\n\n\n [Evals]\n Measure answer correctness with scores → quality measurement\n\n [Observability]\n Which tools were called, how many times, how long did each take → behavior visualization\n\n\nWe'll use **Langfuse v4** , an open-source observability tool. It has a free cloud tier and can also be self-hosted.\n\n> **Note: Langfuse v4 (from March 2026) has a significantly changed API.**\n> `langfuse_context`, `update_current_observation`, and `update_current_trace` are deprecated.\n> This tutorial is compatible with v4.9+.\n\n## What Langfuse Can Do\n\nFeature | Description\n---|---\n**Tracing** | Record execution time and I/O for each RAG/Agent step\n**Cost management** | Visualize API usage and costs\n**Dashboard** | Monitor overall quality and latency in real time\n\n## Directory Structure\n\n\n pgvector-tutorial/\n ├── existing files\n ├── evals/\n │ └── ... # already created\n │\n └── observability/\n ├── traced_rag.py # ★ RAG with tracing (add now)\n └── traced_agent.py # ★ Agent with tracing (add now)\n\n\n## Step 1: Langfuse Setup\n\n### 1-1. Install the Library\n\n\n pip install langfuse\n pip freeze > requirements.txt\n\n\n### 1-2. Create a Langfuse Account\n\n 1. Go to cloud.langfuse.com\n 2. Sign up with GitHub (free, no credit card required)\n 3. Create a new project\n 4. Under \"Settings\" → \"API Keys\", retrieve:\n * `LANGFUSE_PUBLIC_KEY` (starts with `pk-lf-...`)\n * `LANGFUSE_SECRET_KEY` (starts with `sk-lf-...`)\n\n\n\n### 1-3. Add to `.env`\n\n\n # Existing settings\n GEMINI_API_KEY=AIza...\n DB_HOST=localhost\n ...\n\n # Langfuse (new)\n LANGFUSE_PUBLIC_KEY=pk-lf-...\n LANGFUSE_SECRET_KEY=sk-lf-...\n LANGFUSE_HOST=https://cloud.langfuse.com\n\n\n> **⚠️ Important: Do not call`get_client()` before `load_dotenv()`**\n> Langfuse reads environment variables at initialization. Always call `get_client()` _after_ `load_dotenv()`.\n\n## Step 2: RAG with Tracing — `observability/traced_rag.py`\n\nSimply add the `@observe()` decorator to automatically record traces.\n\n\n\n # observability/traced_rag.py\n import sys\n import os\n sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))\n\n import psycopg2\n from google import genai\n from google.genai import types\n from dotenv import load_dotenv\n from langfuse import get_client, observe\n import time\n\n # ── Always call load_dotenv() first ──────────────────────────\n load_dotenv()\n langfuse = get_client() # Initialize AFTER load_dotenv()\n\n client = genai.Client(api_key=os.getenv(\"GEMINI_API_KEY\"))\n\n conn = psycopg2.connect(\n host=os.getenv(\"DB_HOST\"),\n port=os.getenv(\"DB_PORT\"),\n dbname=os.getenv(\"DB_NAME\"),\n user=os.getenv(\"DB_USER\"),\n password=os.getenv(\"DB_PASSWORD\"),\n )\n cur = conn.cursor()\n\n\n @observe()\n def get_embedding(text: str) -> list[float]:\n \"\"\"Trace embedding generation\"\"\"\n result = client.models.embed_content(\n model=\"gemini-embedding-001\",\n contents=text,\n config=types.EmbedContentConfig(\n task_type=\"RETRIEVAL_QUERY\",\n output_dimensionality=768,\n ),\n )\n return result.embeddings[0].values\n\n\n @observe()\n def search_documents(query: str, top_k: int = 3) -> list[dict]:\n \"\"\"Trace Vector DB search\"\"\"\n query_embedding = get_embedding(query)\n cur.execute(\"\"\"\n SELECT title, body,\n 1 - (embedding <=> %s::vector) AS similarity\n FROM documents\n ORDER BY embedding <=> %s::vector\n LIMIT %s;\n \"\"\", (query_embedding, query_embedding, top_k))\n rows = cur.fetchall()\n results = [\n {\"title\": r[0], \"body\": r[1], \"similarity\": round(r[2], 4)}\n for r in rows\n ]\n\n # v4: add metadata with update_current_span()\n langfuse.update_current_span(\n metadata={\n \"retrieved_count\": len(results),\n \"top_similarity\": results[0][\"similarity\"] if results else 0,\n }\n )\n return results\n\n\n @observe(name=\"llm_generate\")\n def generate_answer(question: str, context: str) -> str:\n \"\"\"Trace LLM answer generation\"\"\"\n prompt = f\"\"\"Answer the question based on the following documents.\n\n # Reference Documents\n {context}\n\n # Question\n {question}\n\n # Answer (concisely, based on the reference documents)\"\"\"\n\n response = client.models.generate_content(\n model=\"gemini-2.5-flash\",\n contents=prompt,\n )\n return response.text\n\n\n @observe(name=\"rag_pipeline\")\n def rag_answer(question: str) -> str:\n \"\"\"\n Trace the entire RAG pipeline.\n The Langfuse dashboard will show:\n - rag_pipeline (overall)\n ├── search_documents (Vector DB search)\n │ └── get_embedding (Embedding generation)\n └── llm_generate (LLM answer generation)\n \"\"\"\n langfuse.update_current_span(\n metadata={\"question\": question, \"tags\": [\"rag\", \"production\"]}\n )\n\n docs = search_documents(question, top_k=3)\n context = \"\\n\\n\".join([f\"[{d['title']}]\\n{d['body']}\" for d in docs])\n answer = generate_answer(question, context)\n return answer\n\n\n if __name__ == \"__main__\":\n questions = [\n \"How do you calculate the F1 score?\",\n \"How do you optimize AWS costs?\",\n ]\n\n for question in questions:\n print(f\"\\nQuestion: {question}\")\n answer = rag_answer(question)\n print(f\"Answer: {answer[:100]}...\")\n time.sleep(5) # Rate limit safety\n\n langfuse.flush()\n print(\"\\nTraces sent to Langfuse\")\n print(\"Check the dashboard at https://cloud.langfuse.com\")\n\n\n\n mkdir observability\n python observability/traced_rag.py\n\n\n## Step 3: Agent with Tracing — `observability/traced_agent.py`\n\n> **⚠️ Two common gotchas:**\n>\n> 1. Call `get_client()` after `load_dotenv()`\n> 2. Return `candidates` from `agent_step()` (referenced in `run_agent()`)\n>\n\n\n\n # observability/traced_agent.py\n import sys\n import os\n sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))\n\n import psycopg2\n from google import genai\n from google.genai import types\n from dotenv import load_dotenv\n from langfuse import get_client, observe\n import time\n\n # ── Always call load_dotenv() first ──────────────────────────\n load_dotenv()\n langfuse = get_client()\n\n client = genai.Client(api_key=os.getenv(\"GEMINI_API_KEY\"))\n\n conn = psycopg2.connect(\n host=os.getenv(\"DB_HOST\"),\n port=os.getenv(\"DB_PORT\"),\n dbname=os.getenv(\"DB_NAME\"),\n user=os.getenv(\"DB_USER\"),\n password=os.getenv(\"DB_PASSWORD\"),\n )\n cur = conn.cursor()\n\n\n def get_embedding(text: str) -> list[float]:\n result = client.models.embed_content(\n model=\"gemini-embedding-001\",\n contents=text,\n config=types.EmbedContentConfig(\n task_type=\"RETRIEVAL_QUERY\",\n output_dimensionality=768,\n ),\n )\n return result.embeddings[0].values\n\n\n @observe(name=\"tool_search_documents\")\n def search_documents(query: str, top_k: int = 3) -> list[dict]:\n query_embedding = get_embedding(query)\n cur.execute(\"\"\"\n SELECT title, body, category,\n 1 - (embedding <=> %s::vector) AS similarity\n FROM documents\n ORDER BY embedding <=> %s::vector\n LIMIT %s;\n \"\"\", (query_embedding, query_embedding, top_k))\n rows = cur.fetchall()\n return [\n {\"title\": r[0], \"body\": r[1], \"category\": r[2], \"similarity\": round(r[3], 4)}\n for r in rows\n ]\n\n\n @observe(name=\"tool_list_categories\")\n def list_categories() -> list[dict]:\n cur.execute(\"\"\"\n SELECT category, COUNT(*) as count\n FROM documents\n GROUP BY category\n ORDER BY count DESC;\n \"\"\")\n rows = cur.fetchall()\n return [{\"category\": r[0], \"count\": r[1]} for r in rows]\n\n\n tools = types.Tool(\n function_declarations=[\n types.FunctionDeclaration(\n name=\"search_documents\",\n description=\"Search documents from the Vector DB.\",\n parameters=types.Schema(\n type=types.Type.OBJECT,\n properties={\n \"query\": types.Schema(type=types.Type.STRING, description=\"Search query\"),\n \"top_k\": types.Schema(type=types.Type.INTEGER, description=\"Number of results\"),\n },\n required=[\"query\"],\n ),\n ),\n types.FunctionDeclaration(\n name=\"list_categories\",\n description=\"Get the list of categories in the DB.\",\n parameters=types.Schema(type=types.Type.OBJECT, properties={}),\n ),\n ]\n )\n\n\n def dispatch(func_name: str, func_args: dict):\n if func_name == \"search_documents\":\n return search_documents(**func_args)\n elif func_name == \"list_categories\":\n return list_categories()\n return {\"error\": f\"unknown function: {func_name}\"}\n\n\n @observe(name=\"agent_step\")\n def agent_step(contents: list, step_num: int) -> tuple:\n \"\"\"\n Trace a single Agent step.\n Returns: (part, step_type, candidates)\n Note: candidates must be returned for use in run_agent()\n \"\"\"\n for attempt in range(5):\n try:\n response = client.models.generate_content(\n model=\"gemini-2.5-flash\",\n contents=contents,\n config=types.GenerateContentConfig(tools=[tools]),\n )\n break\n except Exception as e:\n if (\"503\" in str(e) or \"429\" in str(e)) and attempt < 4:\n wait = (attempt + 1) * 10\n print(f\" Retry {attempt+1}... waiting {wait}s\")\n time.sleep(wait)\n else:\n raise\n\n candidates = response.candidates\n if not candidates or not candidates[0].content or not candidates[0].content.parts:\n return None, None, None\n\n part = candidates[0].content.parts[0]\n\n if part.function_call:\n func_name = part.function_call.name\n func_args = dict(part.function_call.args)\n langfuse.update_current_span(\n metadata={\"step\": step_num, \"tool\": func_name, \"args\": str(func_args)}\n )\n return part, \"tool_call\", candidates\n else:\n langfuse.update_current_span(\n metadata={\"step\": step_num, \"type\": \"final_answer\"}\n )\n return part, \"final\", candidates\n\n\n @observe(name=\"agent_pipeline\")\n def run_agent(task: str, max_steps: int = 5) -> str:\n \"\"\"Trace the entire Agent pipeline.\"\"\"\n langfuse.update_current_span(\n metadata={\"task\": task, \"tags\": [\"agent\", \"multi-step\"]}\n )\n\n print(f\"\\nTask: {task}\")\n contents = [types.Content(role=\"user\", parts=[types.Part(text=task)])]\n step_count = 0\n\n for step in range(max_steps):\n print(f\"\\n[Step {step + 1}]\")\n\n part, step_type, candidates = agent_step(contents, step + 1)\n\n if part is None:\n break\n\n if step_type == \"tool_call\":\n func_name = part.function_call.name\n func_args = dict(part.function_call.args)\n print(f\" → {func_name}({func_args})\")\n\n result = dispatch(func_name, func_args)\n print(f\" → {len(result) if isinstance(result, list) else result} results\")\n\n contents.append(\n types.Content(role=\"model\", parts=[types.Part(function_call=part.function_call)])\n )\n contents.append(\n types.Content(\n role=\"user\",\n parts=[types.Part(\n function_response=types.FunctionResponse(\n name=func_name,\n response={\"result\": result},\n )\n )]\n )\n )\n step_count += 1\n\n elif step_type == \"final\":\n text_parts = [\n p.text for p in candidates[0].content.parts\n if hasattr(p, 'text') and p.text\n ]\n answer = \"\\n\".join(text_parts) if text_parts else \"\"\n\n langfuse.update_current_span(\n metadata={\"total_steps\": step_count + 1, \"answer_length\": len(answer)}\n )\n print(f\"\\n[Done] Completed in {step + 1} steps\")\n return answer\n\n return \"Reached maximum step limit.\"\n\n\n if __name__ == \"__main__\":\n result = run_agent(\n \"First check the categories, then give me details about ML evaluation metrics.\"\n )\n print(f\"\\nFinal Answer:\\n{result[:200]}...\")\n\n langfuse.flush()\n print(\"\\nTraces sent to Langfuse\")\n print(\"Check the dashboard at https://cloud.langfuse.com\")\n\n\n\n python observability/traced_agent.py\n\n\n## Step 4: What You Can See in the Dashboard\n\nAfter running, open cloud.langfuse.com to see:\n\n**Agent Trace (actual display):**\n\nName | Latency\n---|---\nagent_pipeline | 4.40s\nagent_step [1] | 1.34s (tool: list_categories)\ntool_list_categories | 0.00s\nagent_step [2] | 0.66s (tool: search_documents)\ntool_search_documents | 0.42s\nagent_step [3] | 1.97s (type: final_answer)\n\n## Langfuse v4 Migration Cheatsheet (from v3)\n\nv3 (old) | v4 (new)\n---|---\n`from langfuse.decorators import observe, langfuse_context` | `from langfuse import get_client, observe`\n`from langfuse import Langfuse` → `Langfuse()` | `from langfuse import get_client` → `get_client()`\n`langfuse_context.update_current_observation(...)` | `langfuse.update_current_span(...)`\n`langfuse_context.update_current_trace(...)` | `langfuse.update_current_span(...)`\n\n## Common Errors\n\nError | Cause | Fix\n---|---|---\n`Authentication error: initialized without public_key` | `get_client()` called before `load_dotenv()` | Call `get_client()` after `load_dotenv()`\n`cannot import name 'langfuse_context'` | Deprecated in v4 | Use `from langfuse import get_client, observe`\n`has no attribute 'update_current_observation'` | Deprecated in v4 | Use `langfuse.update_current_span()`\n`NameError: name 'candidates' is not defined` | `agent_step()` not returning `candidates` | Use `return part, step_type, candidates`\nTraces not appearing | `flush()` not called | Add `langfuse.flush()` at end of script\n`429 RESOURCE_EXHAUSTED` | Gemini free tier limit | Re-run the next day\n\n## Next Steps\n\n * **[Chapter 4: Security]** — Prompt injection defense and guardrail design\n * **Integrate with Evals** — Attach Eval scores to traces via Langfuse's Scoring API\n * **Continuous monitoring** — Set up production alerts\n\n",
"title": "Observability — Tracing RAG and Agents with Langfuse v4"
}