一、docker镜像无法下载时如果无法访问docker官方镜像源 https://registry-1.docker.io那么需要替换国内镜像源root用户执行vi /etc/docker/daemon.json添加“registry-mirrors” :[“https://docker.domys.cc”,“https://hub.domys.cc”,“https://docker.m.daocloud.io”,“https://docker.xuanyuan.me”,“https://docker.1ms.run”]重启Docker服务sudo systemctl daemon-reloadsudo systemctl restart docker二、demo#!/usr/bin/env python # -*- coding: utf-8 -*- LangGraph LangFuse 集成演示 LangFuse 基于 OpenTelemetry 架构提供两种追踪方式 方式1推荐observe 装饰器 - 最简洁直接修饰函数 - 自动追踪函数内的所有 LLM 调用 - 支持嵌套被装饰的函数互相调用时自动形成父子 span 方式2observe OTEL span 属性 - 适用于需要动态传入 user_id / session_id / tags 的场景 - 通过 OpenTelemetry span 属性注入用户上下文 环境变量配置Windows PowerShell $env:OPENAI_API_KEYsk-xxx $env:OPENAI_BASE_URLhttps://api.openai.com/v1 $env:LANGFUSE_PUBLIC_KEYpk-lf-xxx $env:LANGFUSE_SECRET_KEYsk-lf-xxx $env:LANGFUSE_BASE_URLhttp://172.31.234.9:3000 import os import uuid import logging from typing import TypedDict, Optional import warnings warnings.filterwarnings(ignore) # 修复 langchain 版本兼容性 import langchain for attr in (verbose, debug, llm_cache): if not hasattr(langchain, attr): setattr(langchain, attr, False if attr ! llm_cache else None) # 替换为 langchain_openai from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langgraph.graph import StateGraph, END # LangFuse 导入 from langfuse import observe, get_client os.environ[LANGFUSE_PUBLIC_KEY] pk-lf-xxx os.environ[LANGFUSE_SECRET_KEY] sk-lf-xxx os.environ[LANGFUSE_BASE_URL] http://ip:3000 # 抑制 OpenTelemetry 网络超时日志 logging.getLogger(opentelemetry).setLevel(logging.CRITICAL) # 全局配置 LANGFUSE_ENABLED bool( os.getenv(LANGFUSE_PUBLIC_KEY) and os.getenv(LANGFUSE_SECRET_KEY) ) # 使用 ChatOpenAI可兼容 OpenAI / 通义 / DeepSeek 等 OpenAI‑兼容接口 llm ChatOpenAI( modelqwen3.8-27b, api_keysk-xxxxx, base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, temperature0, ) # 方式1observe 装饰器推荐 observe(as_typegeneration) def call_llm(prompt_text: str, model_name: str gpt-3.5-turbo) - str: prompt ChatPromptTemplate.from_template({text}) chain prompt | llm | StrOutputParser() result chain.invoke({text: prompt_text}) get_client().update_current_generation( modelmodel_name, inputprompt_text, outputresult, ) return result observe() def translate(text: str) - str: return call_llm(f请将以下内容翻译成英文只返回译文\n{text}) observe() def summarize(text: str) - str: return call_llm(f请用一句话总结以下内容\n{text}) observe() def qa(question: str) - str: return call_llm(f请用一句话回答{question}) observe(namemulti_tool_agent) def agent_entry(user_input: str): answer qa(user_input) summary summarize(answer) translated translate(summary) return {answer: answer, summary: summary, translated: translated} # 方式2observe 动态设置 trace 属性 class State(TypedDict): question: str answer: Optional[str] def chat_node(state: State) - dict: answer call_llm(f请用一句话回答{state[question]}) return {answer: answer} def build_graph(): graph StateGraph(State) graph.add_node(chat, chat_node) graph.set_entry_point(chat) graph.add_edge(chat, END) return graph.compile() observe(as_typechain) def traced_graph_invoke(question, user_id, session_id, tagsNone, metadataNone): from opentelemetry import trace as otel_trace span otel_trace.get_current_span() if span.is_recording(): span.set_attribute(langfuse.user.id, user_id) span.set_attribute(langfuse.session.id, session_id) if tags: span.set_attribute(langfuse.tags, tags) if metadata: get_client().update_current_span(metadatametadata) app build_graph() return app.invoke({question: question}) # 运行演示 def demo_observe_decorator(): print(\n * 60) print(方式1observe 装饰器) print( * 60) result agent_entry(什么是大语言模型) print(f 回答: {result[answer]}) print(f 摘要: {result[summary]}) print(f 译文: {result[translated]}) def demo_observe_with_context(): print(\n * 60) print(方式2observe 动态上下文) print( * 60) session_id fsession-{uuid.uuid4().hex[:8]} questions [ (user_001, 什么是 Python), (user_001, 它和 Java 有什么区别), (user_002, 推荐一个入门编程语言), ] for user_id, question in questions: print(f\n [{user_id}] {question}) result traced_graph_invoke( questionquestion, user_iduser_id, session_idsession_id, tags[langgraph, multi-turn], metadata{app_version: 1.0.0}, ) print(f 助手: {result[answer]}) if LANGFUSE_ENABLED: get_client().flush() if __name__ __main__: print(fLangFuse: {已启用 if LANGFUSE_ENABLED else 未配置跳过追踪}) demo_observe_decorator() demo_observe_with_context() if LANGFUSE_ENABLED: get_client().flush() print(\n运行完毕。在 LangFuse 控制台可查看追踪数据。)参考资料1.https://blog.csdn.net/leonhongliang806/article/details/148711169
