Xây dựng một quy trình hoàn chỉnh về khả năng quan sát và đánh giá Langfuse để theo dõi, quản lý lời nhắc, chấm điểm và thử nghiệm

Trong hướng dẫn này, chúng tôi triển khai quy trình Langfuse (một nền tảng kỹ thuật LLM mã nguồn mở) để theo dõi, quản lý lời nhắc (prompt), chấm điểm, tập dữ liệu và thử nghiệm. Chúng tôi xây dựng một quy trình làm việc hoàn chỉnh hoạt động với khóa OpenAI thực hoặc LLM giả lập xác định, để có thể hiểu mọi tính năng chính của Langfuse mà không phụ thuộc vào quyền truy cập mô hình trả phí. Chúng tôi bắt đầu bằng cách thiết lập thông tin xác thực và kết nối với Langfuse. Chúng tôi theo dõi các lệnh gọi hàm đơn giản, thiết lập một quy trình RAG nhỏ, quản lý lời nhắc tập trung, đính kèm điểm đánh giá và chạy các thử nghiệm dựa trên tập dữ liệu. Ngoài ra, chúng tôi sẽ xem cách Langfuse giúp quan sát, đánh giá và cải thiện các ứng dụng LLM một cách có cấu trúc và sẵn sàng cho sản xuất. Copy CodeCopiedUse a different Browser import subprocess, sys def pip_install(*pkgs): subprocess.run([sys.executable, "-m", "pip", "install", "-qU", *pkgs], check=True) pip_install("langfuse", "openai") import os from getpass import getpass def _ask(var, prompt, secret=True, default=None): if os.environ.get(var): return os.environ[var] val = (getpass(prompt) if secret else input(prompt)).strip() if not val and default is not None: val = default os.environ[var] = val return val print("Enter your Langfuse credentials (input is hidden):") _ask("LANGFUSE_PUBLIC_KEY", " Langfuse PUBLIC key (pk-lf-...): ") _ask("LANGFUSE_SECRET_KEY", " Langfuse SECRET key (sk-lf-...): ") region = (input(" Region — EU (default) / US / or paste a self-hosted URL: ") .strip().lower()) if region.startswith("http"): HOST = region elif region in ("2", "us"): HOST = "https://us.cloud.langfuse.com" else: HOST = "https://cloud.langfuse.com" os.environ["LANGFUSE_HOST"] = HOST OPENAI_API_KEY = (getpass(" OpenAI key (optional, press Enter to skip): ").strip()) if OPENAI_API_KEY: os.environ["OPENAI_API_KEY"] = OPENAI_API_KEY USE_OPENAI = bool(OPENAI_API_KEY) DEFAULT_MODEL = "gpt-4o-mini" if USE_OPENAI else "mock-llm-v1" from langfuse import get_client, observe, propagate_attributes, Evaluation langfuse = get_client() assert langfuse.auth_check(), "Auth failed — double-check keys/region." print(f"\n Connected to Langfuse at {HOST}") print(f" LLM backend: {'OpenAI (' + DEFAULT_MODEL + ')' if USE_OPENAI else 'built-in mock'}\n") Chúng tôi bắt đầu bằng cách cài đặt các gói Langfuse và OpenAI cần thiết trong môi trường Colab. Sau đó, chúng tôi thu thập thông tin xác thực Langfuse, chọn khu vực Langfuse hoặc URL tự lưu trữ chính xác, và tùy chọn chấp nhận khóa API OpenAI. Cuối cùng, chúng tôi khởi tạo máy khách Langfuse, xác minh xác thực và xác nhận liệu chúng tôi đang sử dụng OpenAI hay LLM giả lập tích hợp. Copy CodeCopiedUse a different Browser Nếu USE_OPENAI: từ langfuse.openai nhập openai _MOCK_FACTS = { "pháp": "Paris", "đức": "Berlin", "nhật bản": "Tokyo", "ý": "Rome", "tây ban nha": "Madrid", "ấn độ": "New Delhi", } def _mock_answer(user_text: str) -> str: t = user_text.lower() for country, capital in _MOCK_FACTS.items(): if country in t: return capital if "langfuse" in t: return ("Langfuse là một nền tảng kỹ thuật LLM mã nguồn mở dành cho " "khả năng quan sát, quản lý lời nhắc, đánh giá và tập dữ liệu.") return "Đây là một phản hồi giả. Vui lòng cung cấp khóa OpenAI để tạo phản hồi thực." def llm_chat(messages, *, model=DEFAULT_MODEL, temperature=0.3, name=None, langfuse_prompt=None) -> str: """Trả về văn bản trợ lý; lệnh gọi được theo dõi dưới dạng một thế hệ Langfuse.""" if USE_OPENAI: kwargs = dict(model=model, messages=messages, temperature=temperature) if name: kwargs["name"] = name if langfuse_prompt: kwargs["langfuse_prompt"] = langfuse_prompt resp = openai.chat.completions.create(**kwargs) return resp.choices[0].message.content last_user = next((m["content"] for m in reversed(messages) if m["role"] == "user"), "") answer = _mock_answer(last_user) gen_kwargs = dict(as_type="generation", name=name or "mock-llm", model=model, input=messages) if langfuse_prompt is not None: gen_kwargs["prompt"] = langfuse_prompt with langfuse.start_as_current_observation(**gen_kwargs) as gen: gen.update(output=answer, usage_details={"input_tokens": 24, "output_tokens": 12}) return answer print("PHẦN 1 ── Theo dõi Decorator -------------------------------------------") @observe() def write_story(topic: str) -> str: return llm_chat( [{"role": "user", "content": f"Viết một câu chuyện một câu về {topic}."}], name="story-generation", ) @observe() def story_pipeline(topic: str) -> str: return write_story(topic) print(" →", story_pipeline("một robot gỡ lỗi")) Chúng tôi định nghĩa trình trợ giúp LLM hỗ trợ cả thế hệ OpenAI thực và phản hồi giả định. Chúng tôi cũng đảm bảo rằng ngay cả đường dẫn giả định cũng tạo ra một quan sát thế hệ Langfuse phù hợp, để hướng dẫn vẫn có thể theo dõi đầy đủ mà không cần khóa OpenAI. Sau đó, chúng tôi trình bày việc theo dõi dựa trên decorator cơ bản bằng cách gói một quy trình tạo câu chuyện đơn giản với @observe. Sao chép mã đã sao chépSử dụng trình duyệt khác print("\nPHẦN 2 ── Theo dõi RAG thủ công --------------------------------------------") _KB = { "hoàn tiền": "Việc hoàn tiền được xử lý trong vòng 5–7 ngày làm việc theo phương thức ban đầu.", "bảo hành": "Tất cả các sản phẩm đều có bảo hành giới hạn của nhà sản xuất 1 năm.", } @observe(name="retrieve") def retrieve(question: str): q = question.lower() hits = [v for k, v in _KB.items() if k in q] or list(_KB.values()) return hits[:2] @observe(name="rag-pipeline") def rag_pipeline(question: str, user_id="user-42", session_id="sess-001") -> str: with propagate_attributes(user_id=user_id, session_id=session_id, tags=["rag", "support-bot", "tutorial"]): context = "\n".join(retrieve(question)) return llm_chat( [{"role": "system", "content": "Trả lời câu hỏi CHỈ sử dụng ngữ cảnh được cung cấp."}, {"role": "user", "content": f"Ngữ cảnh:\n{context}\n\nCâu hỏi: {question}"}], name="rag-answer", ) rag_answer = rag_pipeline("Hoàn tiền mất bao lâu?") rag_trace_id = langfuse.get_current_trace_id() print(" →", rag_answer) Chúng tôi xây dựng một quy trình RAG thủ công nhỏ sử dụng cơ sở kiến thức trong bộ nhớ đơn giản để hoàn tiền, sh