用户输入 → 意图识别 → 任务规划 → 工具选择 → 工具调用 → 结果解析 → 
二次推理 → 可能的回退/重试 → 最终输出

# ============================================
# 步骤 1：更新系统包
# ============================================
sudo dnf update -y

# ============================================
# 步骤 2：安装 Docker 依赖工具
# ============================================
sudo dnf install -y dnf-plugins-core

# ============================================
# 步骤 3：添加 Docker 官方仓库
# 注意：Fedora 使用 fedora 专用仓库路径
# ============================================
sudo dnf config-manager --add-repo https://download.docker.com/linux/fedora/docker-ce.repo

# ============================================
# 步骤 4：安装 Docker Engine 及插件
# docker-ce: Docker 引擎
# docker-ce-cli: 命令行工具
# containerd.io: 容器运行时
# docker-compose-plugin: Compose V2 插件
# docker-buildx-plugin: 构建工具
# ============================================
sudo dnf install -y docker-ce docker-ce-cli containerd.io docker-compose-plugin docker-buildx-plugin

# ============================================
# 步骤 5：启动 Docker 并设置开机自启
# ============================================
sudo systemctl start docker
sudo systemctl enable docker

# ============================================
# 步骤 6：将当前用户加入 docker 组（免 sudo）
# 注意：需要重新登录才能生效
# ============================================
sudo usermod -aG docker $USER

# ============================================
# 步骤 7：验证安装
# ============================================
docker --version
# 预期输出：Docker version 28.x.x, build xxxxxxx

docker compose version
# 预期输出：Docker Compose version v2.x.x

# ============================================
# 防火墙配置：开放 Langfuse Web 端口（3000）
# ============================================
sudo firewall-cmd --permanent --add-port=3000/tcp
sudo firewall-cmd --permanent --add-port=5432/tcp  # PostgreSQL（仅本地调试需要）
sudo firewall-cmd --permanent --add-port=8123/tcp  # ClickHouse HTTP（仅本地调试需要）
sudo firewall-cmd --reload

# ============================================
# SELinux 配置
# 方案一（推荐开发环境）：设置为 Permissive 模式
# ============================================
sudo setenforce 0
# 永久生效需修改 /etc/selinux/config 中 SELINUX=permissive

# 方案二（生产环境）：保持 Enforcing，为 Docker 添加策略
sudo setsebool -P container_manage_cgroup on

┌─────────────────────────────────────────────────────────┐
│                    Langfuse v3 架构                       │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  ┌──────────────┐    ┌──────────────┐                  │
│  │ langfuse-web │    │langfuse-worker│                  │
│  │  (API + UI)  │    │ (异步处理)    │                  │
│  └──────┬───────┘    └──────┬───────┘                  │
│         │                    │                          │
│  ┌──────┴────────────────────┴───────┐                  │
│  │           PostgreSQL              │  ← 元数据存储     │
│  └───────────────────────────────────┘                  │
│  ┌───────────────────────────────────┐                  │
│  │           ClickHouse              │  ← Trace 数据    │
│  └───────────────────────────────────┘                  │
│  ┌───────────────────────────────────┐                  │
│  │             Redis                 │  ← 缓存/队列     │
│  └───────────────────────────────────┘                  │
│  ┌───────────────────────────────────┐                  │
│  │             MinIO                 │  ← 对象存储      │
│  └───────────────────────────────────┘                  │
│                                                         │
└─────────────────────────────────────────────────────────┘

# 创建项目目录
mkdir -p ~/langfuse-deploy
cd ~/langfuse-deploy

# 下载官方 docker-compose.yml
curl -o docker-compose.yml \
  https://raw.githubusercontent.com/langfuse/langfuse/main/docker-compose.yml

# 创建环境变量文件
touch .env

# ============================================
# .env 文件 - Langfuse 环境变量配置
# ============================================

# ---------- 安全密钥（必须修改！） ----------
# 用于加密 Session 和 JWT Token
# 生成方式: openssl rand -base64 32
NEXTAUTH_SECRET=your-super-secret-nextauth-key-change-me
SALT=your-random-salt-value-change-me
ENCRYPTION_KEY=0000000000000000000000000000000000000000000000000000000000000000

# ---------- 数据库配置 ----------
DATABASE_URL=postgresql://postgres:postgres@db:5432/postgres
DIRECT_URL=postgresql://postgres:postgres@db:5432/postgres

# ---------- ClickHouse 配置 ----------
CLICKHOUSE_URL=http://clickhouse:8123
CLICKHOUSE_MIGRATION_URL=clickhouse://clickhouse:9000
CLICKHOUSE_USER=default
CLICKHOUSE_PASSWORD=clickhouse

# ---------- Redis 配置 ----------
REDIS_HOST=redis
REDIS_PORT=6379
REDIS_AUTH=redis

# ---------- MinIO 配置 ----------
MINIO_ENDPOINT=minio
MINIO_PORT=9000
MINIO_ACCESS_KEY=minio
MINIO_SECRET_KEY=miniosecret
MINIO_BUCKET=langfuse

# ---------- 应用配置 ----------
NEXTAUTH_URL=http://localhost:3000
LANGFUSE_ENABLE_EXPERIMENTAL_FEATURES=true
TELEMETRY_ENABLED=false

# 生成 NEXTAUTH_SECRET
openssl rand -base64 32

# 生成 SALT
openssl rand -base64 24

# 生成 ENCRYPTION_KEY（64位十六进制）
openssl rand -hex 32

# ============================================
# 启动所有服务（后台运行）
# ============================================
cd ~/langfuse-deploy
docker compose up -d

# ============================================
# 查看服务状态
# ============================================
docker compose ps

# 预期输出（所有服务均为 Up/Running）：
# NAME              STATUS
# langfuse-web      Up (healthy)
# langfuse-worker   Up (healthy)
# db                Up (healthy)
# clickhouse        Up (healthy)
# redis             Up
# minio             Up

# ============================================
# 查看启动日志（排查问题用）
# ============================================
docker compose logs -f langfuse-web

# ============================================
# 检查特定服务健康状态
# ============================================
docker compose logs clickhouse | tail -20
docker compose logs db | tail -20

Public Key:  pk-lf-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
Secret Key:  sk-lf-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx

sudo tee /etc/systemd/system/langfuse.service << 'EOF'
[Unit]
Description=Langfuse LLM Observability Platform
Requires=docker.service
After=docker.service network-online.target

[Service]
Type=oneshot
RemainAfterExit=yes
WorkingDirectory=/home/yourusername/langfuse-deploy
ExecStart=/usr/bin/docker compose up -d
ExecStop=/usr/bin/docker compose down
TimeoutStartSec=300

[Install]
WantedBy=multi-user.target
EOF

# 重新加载 systemd 配置
sudo systemctl daemon-reload

# 启用开机自启
sudo systemctl enable langfuse.service

# 手动启动/停止
sudo systemctl start langfuse.service
sudo systemctl stop langfuse.service

# 查看状态
sudo systemctl status langfuse.service

# ============================================
# 方式一：使用 pip 安装（推荐）
# ============================================
pip install opik

# 如果需要特定版本
pip install opik==2.1.13

# ============================================
# 方式二：使用 uv 安装（更快的包管理器）
# ============================================
# 先安装 uv（如果尚未安装）
curl -LsSf https://astral.sh/uv/install.sh | sh

# 使用 uv 安装 opik
uv pip install opik

# ============================================
# 验证安装
# ============================================
python -c "import opik; print(opik.__version__)"
# 预期输出：2.1.13 或更新版本

# 交互式配置
opik configure

# ============================================
# 步骤 1：克隆 Opik 仓库
# ============================================
git clone https://github.com/comet-ml/opik.git
cd opik/deployment/docker-compose

# ============================================
# 步骤 2：查看并配置环境变量（可选）
# ============================================
cat .env  # 查看默认配置

# 如需修改端口或密码，编辑 .env 文件
# 默认 Web UI 端口：5173
# 默认 API 端口：8080

# ============================================
# 步骤 3：启动服务
# ============================================
docker compose up -d

# ============================================
# 步骤 4：验证服务状态
# ============================================
docker compose ps

# ============================================
# 步骤 5：访问 Web UI
# ============================================
# 浏览器打开：http://localhost:5173

# ============================================
# .env - Opik 环境变量配置
# ============================================

# Opik 服务端地址
OPIK_URL_OVERRIDE=http://localhost:5173/api

# API Key（从 Web UI 获取）
OPIK_API_KEY=your-opik-api-key-here

# Workspace 名称
OPIK_WORKSPACE=default

# 项目名称
OPIK_PROJECT_NAME=my-agent-project

"""
Opik 客户端初始化配置
"""
import os
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

# 方式一：通过环境变量自动配置
# Opik SDK 会自动读取 OPIK_URL_OVERRIDE 和 OPIK_API_KEY
import opik

# 方式二：显式初始化客户端
from opik import Opik

client = Opik(
    project_name=os.getenv("OPIK_PROJECT_NAME", "my-agent-project"),
    workspace=os.getenv("OPIK_WORKSPACE", "default"),
)

print(f"✅ Opik 客户端初始化成功")
print(f"   项目: {client.project_name}")
print(f"   工作空间: {client.workspace}")

"""
示例：Langfuse 基础追踪
文件：basic_langfuse_tracing.py
"""
import os
from dotenv import load_dotenv
from langfuse import observe, langfuse_context

# 加载环境变量
load_dotenv()

# ============================================
# 配置 Langfuse 连接（通过环境变量）
# LANGFUSE_PUBLIC_KEY=pk-lf-xxx
# LANGFUSE_SECRET_KEY=sk-lf-xxx
# LANGFUSE_HOST=http://localhost:3000
# ============================================

@observe()  # 自动追踪此函数，创建一个 Trace
def process_user_query(query: str) -> str:
    """
    处理用户查询的入口函数
    @observe() 会自动记录：
    - 函数名（作为 Trace 名称）
    - 输入参数（query）
    - 返回值
    - 执行时间
    """
    # 模拟处理逻辑
    result = f"处理结果：{query.upper()}"
    return result
@observe()  # 嵌套追踪：此函数的 Trace 会成为父 Trace 的子 Span
def call_llm(prompt: str) -> str:
    """
    调用 LLM 的子步骤
    """
    # 这里实际应该调用 OpenAI/其他 LLM API
    # 为了演示，我们模拟一个响应
    import time
    time.sleep(0.5)  # 模拟 LLM 推理延迟
    return f"LLM 响应：针对 '{prompt}' 的回答"
@observe()
def main_agent(user_input: str) -> str:
    """
    Agent 主函数 - 顶层 Trace
    """
    # 设置 Trace 级别的元数据
    langfuse_context.update_current_trace(
        name="customer-support-agent",  # 自定义 Trace 名称
        user_id="user-12345",          # 关联用户 ID
        session_id="session-abc",      # 关联会话 ID
        tags=["production", "v2.1"],   # 标签
        metadata={                      # 自定义元数据
            "agent_version": "2.1.0",
            "model": "gpt-4o",
            "environment": "production"
        }
    )
   # 调用子步骤（自动形成嵌套 Trace）
    processed = process_user_query(user_input)
    llm_response = call_llm(processed)
  return llm_response
# ============================================
# 运行示例
# ============================================
if __name__ == "__main__":
    result = main_agent("帮我查询订单状态")
    print(f"最终结果: {result}")
    
    # 确保所有数据都已发送（程序退出前）
    langfuse_context.flush()
    print("✅ Trace 数据已发送到 Langfuse")

"""
示例：多层嵌套追踪
展示 Agent → Planner → Tool Call → LLM 的完整链路
"""
from langfuse import observe, langfuse_context
from langfuse.openai import openai  # Langfuse 增强的 OpenAI 客户端
@observe()
def agent_orchestrator(task: str) -> str:
    """Agent 编排器 - 顶层入口"""
    langfuse_context.update_current_trace(
        name="research-agent",
        tags=["multi-step", "research"]
    )
    
    # 步骤 1：规划
    plan = plan_task(task)
    
    # 步骤 2：执行各子任务
    results = []
    for step in plan:
        result = execute_step(step)
        results.append(result)
    
    # 步骤 3：汇总
    final_answer = synthesize_results(results)
    return final_answer


@observe()
def plan_task(task: str) -> list:
    """任务规划器"""
    # 实际场景中这里会调用 LLM 生成计划
    return [
        {"action": "search", "query": f"{task} 相关资料"},
        {"action": "analyze", "data": "搜索结果"},
        {"action": "summarize", "content": "分析结果"}
    ]


@observe()
def execute_step(step: dict) -> str:
    """执行单个步骤"""
    if step["action"] == "search":
        return search_tool(step["query"])
    elif step["action"] == "analyze":
        return analyze_data(step["data"])
    elif step["action"] == "summarize":
        return summarize_content(step["content"])
    return ""


@observe()
def search_tool(query: str) -> str:
    """搜索工具调用"""
    # 记录工具调用的详细信息
    langfuse_context.update_current_observation(
        metadata={"tool": "web_search", "query": query}
    )
    return f"搜索结果：关于 '{query}' 的 5 条结果"


@observe()
def analyze_data(data: str) -> str:
    """数据分析步骤"""
    return f"分析完成：{data}"


@observe()
def summarize_content(content: str) -> str:
    """内容汇总步骤"""
    return f"摘要：{content} 的核心要点是..."


@observe()
def synthesize_results(results: list) -> str:
    """最终汇总"""
    return "综合所有步骤的最终答案"


# 运行
if __name__ == "__main__":
    answer = agent_orchestrator("研究 2026 年 AI Agent 发展趋势")
    print(answer)
    langfuse_context.flush()

📊 research-agent (Trace)
├── 📋 plan_task (Span) - 120ms
├── 🔧 execute_step (Span) - 850ms
│   └── 🔍 search_tool (Span) - 800ms
├── 🔧 execute_step (Span) - 200ms
│   └── 📊 analyze_data (Span) - 180ms
├── 🔧 execute_step (Span) - 300ms
│   └── 📝 summarize_content (Span) - 280ms
└── 🎯 synthesize_results (Span) - 150ms

"""
示例：Opik 基础追踪
文件：basic_opik_tracing.py
"""
import os
from dotenv import load_dotenv
from opik import track, opik_context

load_dotenv()


@track  # Opik 的追踪装饰器（注意：不需要括号）
def research_agent(query: str) -> str:
    """
    Agent 主函数
    @track 自动记录：
    - 函数名
    - 输入参数
    - 返回值
    - 执行时间
    - Token 使用量（如果内部有 LLM 调用）
    """
    # 更新当前 Trace 的元数据
    opik_context.update_current_trace(
        tags=["research", "production"],
        metadata={"agent_type": "research", "version": "1.0"}
    )
    
    # 子步骤会自动成为当前 Trace 的子 Span
    search_results = search_web(query)
    analysis = analyze_results(search_results)
    summary = generate_summary(analysis)
    
    return summary


@track
def search_web(query: str) -> list:
    """网络搜索步骤"""
    # 模拟搜索
    return [
        {"title": "AI Agent 2026 趋势报告", "relevance": 0.95},
        {"title": "多模态 Agent 最新进展", "relevance": 0.88},
    ]


@track
def analyze_results(results: list) -> str:
    """分析搜索结果"""
    return f"分析了 {len(results)} 条结果，发现 3 个关键趋势"


@track
def generate_summary(analysis: str) -> str:
    """生成摘要"""
    return f"研究摘要：{analysis}"


# ============================================
# 运行
# ============================================
if __name__ == "__main__":
    result = research_agent("2026年AI Agent发展趋势")
    print(f"结果: {result}")
    # Opik 会自动在程序退出时 flush 数据
    # 也可以手动调用：opik_context.flush()

"""
示例：Opik 与 OpenAI 自动集成
无需手动添加 @track，自动捕获 LLM 调用详情
"""
from opik.integrations.openai import openai  # 导入 Opik 增强的 openai 模块
from opik import track


@track
def chat_with_model(user_message: str) -> str:
    """
    使用 OpenAI 兼容接口
    Opik 会自动记录：
    - 模型名称
    - 输入/输出 Token 数
    - 请求延迟
    - 完整的 messages 内容
    """
    response = openai.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "你是一个有帮助的助手。"},
            {"role": "user", "content": user_message}
        ],
        temperature=0.7,
        max_tokens=1000
    )
    return response.choices[0].message.content


@track
def langchain_agent_example(query: str) -> str:
    """
    LangChain 集成示例
    """
    from opik.integrations.langchain import OpikTracer
    
    # 创建 Opik 追踪器
    opik_tracer = OpikTracer(tags=["langchain", "agent"])
    
    # 在 LangChain 链中使用
    from langchain_openai import ChatOpenAI
    from langchain.chains import LLMChain
    from langchain.prompts import PromptTemplate
    
    llm = ChatOpenAI(model="gpt-4o")
    prompt = PromptTemplate(
        input_variables=["query"],
        template="请详细回答以下问题：{query}"
    )
    
    chain = LLMChain(llm=llm, prompt=prompt)
    result = chain.invoke({"query": query}, config={"callbacks": [opik_tracer]})
    
    return result["text"]
                              
                                  

"""
完整 ReAct Agent 示例（带 Langfuse + Opik 双平台追踪）
文件：react_agent_with_observability.py

依赖安装：
pip install langfuse opik openai python-dotenv
"""
import os
import json
import time
from typing import Optional
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

# ============================================
# 双平台追踪导入
# ============================================
from langfuse import observe as langfuse_observe
from langfuse import langfuse_context
from opik import track as opik_track
from opik import opik_context

# 使用 Langfuse 增强的 OpenAI 客户端
from langfuse.openai import openai


# ============================================
# 工具定义
# ============================================
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "web_search",
            "description": "搜索互联网获取最新信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "搜索查询关键词"
                    }
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculator",
            "description": "执行数学计算",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "数学表达式，如 '2 + 3 * 4'"
                    }
                },
                "required": ["expression"]
            }
        }
    }
]


# ============================================
# 工具实现（带追踪）
# ============================================
@langfuse_observe(name="tool:web_search")
@opik_track(name="tool:web_search")
def web_search(query: str) -> str:
    """
    网络搜索工具
    实际项目中这里调用搜索 API
    """
    # 模拟搜索延迟
    time.sleep(0.3)
    
    # 模拟搜索结果
    results = {
        "status": "success",
        "results": [
            {"title": f"关于'{query}'的最新资讯", "snippet": "这是搜索结果摘要..."},
            {"title": f"'{query}'深度分析", "snippet": "详细分析内容..."}
        ]
    }
    return json.dumps(results, ensure_ascii=False)


@langfuse_observe(name="tool:calculator")
@opik_track(name="tool:calculator")
def calculator(expression: str) -> str:
    """
    计算器工具
    """
    try:
        # 安全计算（生产环境应使用更安全的方案）
        result = eval(expression, {"__builtins__": {}}, {})
        return json.dumps({"status": "success", "result": result})
    except Exception as e:
        return json.dumps({"status": "error", "message": str(e)})


# 工具路由
TOOL_REGISTRY = {
    "web_search": web_search,
    "calculator": calculator,
}


# ============================================
# Agent 核心逻辑（带完整追踪）
# ============================================
@langfuse_observe(name="react-agent")
@opik_track(name="react-agent")
def react_agent(
    user_query: str,
    max_steps: int = 5,
    model: str = "gpt-4o"
) -> str:
    """
    ReAct Agent 主函数
    
    参数:
        user_query: 用户输入
        max_steps: 最大推理步数（防止死循环）
        model: 使用的 LLM 模型
    
    返回:
        Agent 的最终回答
    """
    # ---- 设置 Trace 元数据 ----
    langfuse_context.update_current_trace(
        name=f"react-agent: {user_query[:50]}",
        user_id=os.getenv("USER_ID", "anonymous"),
        tags=["react", "multi-step"],
        metadata={
            "model": model,
            "max_steps": max_steps,
            "agent_version": "1.0.0"
        }
    )
    
    # ---- 初始化对话历史 ----
    messages = [
        {
            "role": "system",
            "content": (
                "你是一个 ReAct Agent。你可以使用工具来帮助用户。"
                "每一步先思考（Thought），再决定是否调用工具（Action），"
                "最后根据工具结果继续推理或给出最终答案。"
            )
        },
        {"role": "user", "content": user_query}
    ]
    
    # ---- ReAct 循环 ----
    for step in range(max_steps):
        # 记录当前步骤
        step_trace_name = f"step-{step + 1}"
        
        # 调用 LLM（Langfuse 自动追踪 OpenAI 调用）
        response = openai.chat.completions.create(
            model=model,
            messages=messages,
            tools=TOOLS,
            tool_choice="auto",
            temperature=0.1
        )
        
        assistant_message = response.choices[0].message
        
        # 检查是否有工具调用
        if assistant_message.tool_calls:
            # 将助手消息加入历史
            messages.append(assistant_message)
            
            # 执行每个工具调用
            for tool_call in assistant_message.tool_calls:
                func_name = tool_call.function.name
                func_args = json.loads(tool_call.function.arguments)
                
                # 执行工具
                tool_func = TOOL_REGISTRY.get(func_name)
                if tool_func:
                    tool_result = tool_func(**func_args)
                else:
                    tool_result = json.dumps({
                        "status": "error",
                        "message": f"未知工具: {func_name}"
                    })
                
                # 将工具结果加入对话历史
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": tool_result
                })
        else:
            # 没有工具调用 → Agent 给出最终答案
            final_answer = assistant_message.content
            
            # 记录最终答案到 Trace
            langfuse_context.update_current_trace(
                output=final_answer,
                metadata={"total_steps": step + 1}
            )
            
            return final_answer
    
    # 超过最大步数
    fallback = "抱歉，我无法在限定步数内完成此任务。请尝试简化问题。"
    langfuse_context.update_current_trace(
        output=fallback,
        metadata={"total_steps": max_steps, "hit_max_steps": True}
    )
    return fallback


# ============================================
# 运行入口
# ============================================
if __name__ == "__main__":
    # 示例查询
    queries = [
        "帮我计算 (15 + 27) * 3 的结果",
        "搜索一下 2026 年最流行的 AI Agent 框架",
    ]
    
    for query in queries:
        print(f"\n{'='*60}")
        print(f"🤖 用户查询: {query}")
        print(f"{'='*60}")
        
        result = react_agent(query)
        print(f"\n📝 Agent 回答: {result}")
    
    # 确保数据发送完毕
    langfuse_context.flush()
    print("\n✅ 所有 Trace 数据已发送")

@langfuse_observe(name="react-agent")  # Langfuse 追踪
@opik_track(name="react-agent")         # Opik 追踪（叠加使用）
def react_agent(user_query: str, ...) -> str:

┌─────────────────────────────────────────────────────────────┐
│ Trace: react-agent: 帮我计算 (15+27)*3 的结果               │
│ 时间: 2026-08-06 09:30:15  耗时: 2.3s  状态: ✅ Success     │
│ User: user-12345  Session: session-abc                      │
│ Tags: [react] [multi-step]                                  │
│ Metadata: {"model": "gpt-4o", "total_steps": 2}            │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  📊 Timeline View                                           │
│  ┌─────────────────────────────────────────────────────┐    │
│  │ react-agent          ████████████████████████ 2.3s  │    │
│  │ ├─ LLM Call #1      ████████ 0.8s                  │    │
│  │ ├─ tool:calculator   ██ 0.1s                        │    │
│  │ └─ LLM Call #2      ████████████ 1.2s              │    │
│  └─────────────────────────────────────────────────────┘    │
│                                                             │
│  📋 Detail View                                             │
│  • LLM Call #1:                                            │
│    Input: [system + user messages]                          │
│    Output: tool_call(calculator, "(15+27)*3")              │
│    Tokens: 156 in / 23 out                                  │
│    Cost: $0.0012                                            │
│                                                             │
│  • tool:calculator:                                         │
│    Input: {"expression": "(15+27)*3"}                       │
│    Output: {"status": "success", "result": 126}            │
│    Duration: 100ms                                          │
│                                                             │
│  • LLM Call #2:                                            │
│    Input: [previous + tool result]                          │
│    Output: "计算结果是 126"                                  │
│    Tokens: 203 in / 15 out                                  │
│    Cost: $0.0015                                            │
│                                                             │
└─────────────────────────────────────────────────────────────┘

react-agent (32.5s) ⚠️ 超时
├─ LLM Call #1 (0.8s) → 决定调用 web_search
├─ tool:web_search (0.3s) → 返回结果
├─ LLM Call #2 (0.9s) → 再次调用 web_search（同样参数！）
├─ tool:web_search (0.3s) → 返回相同结果
├─ LLM Call #3 (0.8s) → 又调用 web_search...
├─ ... (重复 5 次)
└─ LLM Call #6 (1.0s) → 放弃，输出"无法完成"

@langfuse_observe(name="tool:web_search")
@opik_track(name="tool:web_search")
def web_search(query: str) -> str:
    """修复后的搜索工具 - 返回格式与 Agent 期望一致"""
    time.sleep(0.3)
    
    results = {
        "status": "success",
        "results": [
            {
                "title": f"关于'{query}'的最新资讯",
                "content": "这是搜索结果详细内容...",  # ← 修复：使用 content 而非 snippet
                "url": "https://example.com/article1"
            }
        ]
    }
    return json.dumps(results, ensure_ascii=False)
                              
                                  

SYSTEM_PROMPT = """
你是一个 ReAct Agent。

重要规则：
1. 如果工具调用返回了结果，不要以相同参数重复调用同一工具
2. 如果工具返回错误，最多重试 1 次，然后尝试其他方案
3. 如果连续 2 次无法获取有效信息，直接向用户说明情况

"""
获取 Langfuse 延迟统计数据
"""
from langfuse import Langfuse

# 初始化客户端
langfuse = Langfuse(
    public_key=os.getenv("LANGFUSE_PUBLIC_KEY"),
    secret_key=os.getenv("LANGFUSE_SECRET_KEY"),
    host=os.getenv("LANGFUSE_HOST", "http://localhost:3000")
)

# 获取最近的 Traces
traces = langfuse.fetch_traces(
    limit=100,
    tags=["production"]
)

# 计算延迟统计
latencies = [t.latency for t in traces.data if t.latency]
latencies.sort()

print(f"样本数: {len(latencies)}")
print(f"P50: {latencies[len(latencies)//2]:.2f}s")
print(f"P95: {latencies[int(len(latencies)*0.95)]:.2f}s")
print(f"P99: {latencies[int(len(latencies)*0.99)]:.2f}s")
print(f"最大: {latencies[-1]:.2f}s")
                              
                                  

"""
Token 消耗分析脚本
"""
from langfuse import Langfuse
from datetime import datetime, timedelta

langfuse = Langfuse()

# 获取最近 24 小时的 Traces
since = datetime.now() - timedelta(hours=24)
traces = langfuse.fetch_traces(limit=500, from_timestamp=since)

# 汇总 Token 消耗
total_input_tokens = 0
total_output_tokens = 0
total_cost = 0.0
model_usage = {}

for trace in traces.data:
    observations = langfuse.fetch_observations(trace_id=trace.id)
    for obs in observations.data:
        if obs.type == "GENERATION" and obs.usage:
            input_t = obs.usage.input or 0
            output_t = obs.usage.output or 0
            total_input_tokens += input_t
            total_output_tokens += output_t
            
            # 按模型统计
            model = obs.model or "unknown"
            if model not in model_usage:
                model_usage[model] = {"input": 0, "output": 0, "count": 0}
            model_usage[model]["input"] += input_t
            model_usage[model]["output"] += output_t
            model_usage[model]["count"] += 1

print("=" * 50)
print("📊 过去 24 小时 Token 消耗报告")
print("=" * 50)
print(f"总输入 Tokens: {total_input_tokens:,}")
print(f"总输出 Tokens: {total_output_tokens:,}")
print(f"总 Tokens: {total_input_tokens + total_output_tokens:,}")
print(f"\n按模型分布:")
for model, usage in model_usage.items():
    print(f"  {model}: {usage['count']} 次调用, "
          f"输入 {usage['input']:,}, 输出 {usage['output']:,}")
                                  

"""
错误率监控脚本
"""
from langfuse import Langfuse
from datetime import datetime, timedelta

langfuse = Langfuse()

since = datetime.now() - timedelta(hours=24)
traces = langfuse.fetch_traces(limit=1000, from_timestamp=since)

total = len(traces.data)
errors = [t for t in traces.data if t.level == "ERROR"]
error_rate = len(errors) / total * 100 if total > 0 else 0

print(f"总请求数: {total}")
print(f"错误数: {len(errors)}")
print(f"错误率: {error_rate:.2f}%")

# 错误分类
if errors:
    print(f"\n最近错误详情:")
    for err in errors[:5]:
        print(f"  - [{err.timestamp}] {err.name}: {err.status_message}")

"""
自定义指标上报到 Langfuse
"""
from langfuse import Langfuse

langfuse = Langfuse()

# 创建自定义 Score（可作为指标）
langfuse.score(
    trace_id="your-trace-id",
    name="response_quality",      # 指标名称
    value=0.85,                    # 数值（0-1）
    comment="回答相关但不完整",     # 备注
    data_type="NUMERIC"           # 数据类型
)

# 分类指标
langfuse.score(
    trace_id="your-trace-id",
    name="task_completion",
    value="completed",            # 分类值
    data_type="CATEGORICAL"
)

"""
每日 Agent 运行报告生成器
文件：daily_report.py
建议通过 cron 每日执行
"""
from langfuse import Langfuse
from datetime import datetime, timedelta
import json

def generate_daily_report():
    """生成每日运行报告"""
    langfuse = Langfuse()
    
    # 时间范围：过去 24 小时
    end_time = datetime.now()
    start_time = end_time - timedelta(hours=24)
    
    # 获取所有 Traces
    traces = langfuse.fetch_traces(
        limit=2000,
        from_timestamp=start_time,
        to_timestamp=end_time
    )
    
    # ---- 统计指标 ----
    total_traces = len(traces.data)
    
    # 延迟统计
    latencies = [t.latency for t in traces.data if t.latency]
    avg_latency = sum(latencies) / len(latencies) if latencies else 0
    
    # 成功/失败统计
    success_count = sum(1 for t in traces.data if t.level != "ERROR")
    error_count = total_traces - success_count
    
    # Token 统计（简化版）
    total_tokens = 0
    for trace in traces.data:
        if trace.usage:
            total_tokens += (trace.usage.get("input", 0) + 
                           trace.usage.get("output", 0))
    
    # ---- 生成报告 ----
    report = {
        "report_date": start_time.strftime("%Y-%m-%d"),
        "generated_at": end_time.isoformat(),
        "summary": {
            "total_requests": total_traces,
            "success_rate": f"{success_count/total_traces*100:.1f}%" if total_traces else "N/A",
            "error_count": error_count,
            "avg_latency_ms": round(avg_latency * 1000, 1),
            "total_tokens": total_tokens,
        },
        "top_errors": [],  # 可进一步填充
        "recommendations": []
    }
    
    # 自动生成建议
    if error_count / max(total_traces, 1) > 0.05:
        report["recommendations"].append("⚠️ 错误率超过 5%，建议检查工具调用稳定性")
    if avg_latency > 5:
        report["recommendations"].append("⚠️ 平均延迟超过 5s，建议优化 LLM 调用或减少推理步骤")
    
    # 输出报告
    print(json.dumps(report, ensure_ascii=False, indent=2))
    
    # 可选：发送到邮件/Slack/企业微信
    # send_report_to_slack(report)
    
    return report


if __name__ == "__main__":
    generate_daily_report()

# 每天早上 9 点执行
0 9 * * * cd /path/to/project && python daily_report.py >> /var/log/agent_report.log 2>&1

# ✅ 检查 1：环境变量是否正确设置
echo $LANGFUSE_PUBLIC_KEY
echo $LANGFUSE_SECRET_KEY
echo $LANGFUSE_HOST

# ✅ 检查 2：Langfuse 服务是否正常运行
curl http://localhost:3000/api/public/health
# 预期返回：{"status": "OK"}

# ✅ 检查 3：网络连通性
curl -v http://localhost:3000

# ✅ 检查 4：Docker 容器日志
docker compose logs langfuse-worker --tail=50

"""
确保数据不丢失的正确写法
"""
from langfuse import Langfuse, observe, langfuse_context
import atexit

# 方法 1：注册退出钩子（推荐）
atexit.register(lambda: langfuse_context.flush())

# 方法 2：使用 try-finally
@observe()
def my_agent(query: str) -> str:
    try:
        # ... Agent 逻辑 ...
        return result
    finally:
        langfuse_context.flush()  # 确保数据发送

# 方法 3：使用上下文管理器（Langfuse v3+）
with Langfuse() as langfuse:
    trace = langfuse.trace(name="my-trace")
    trace.generation(name="llm-call", input="hello", output="world")
# 退出 with 块时自动 flush

langfuse.api.error.ApiError: Connection timed out
# 或
requests.exceptions.ConnectTimeout: HTTPConnectionPool(host='localhost', port=3000)

"""
处理连接超时的配置
"""
import os
from langfuse import Langfuse

# 增加超时时间（默认 15s）
os.environ["LANGFUSE_FLUSH_INTERVAL"] = "10"  # 批量发送间隔（秒）
os.environ["LANGFUSE_FLUSH_AT"] = "100"       # 批量发送阈值（条数）

# 如果使用代理
os.environ["HTTP_PROXY"] = "http://proxy:8080"
os.environ["HTTPS_PROXY"] = "http://proxy:8080"

# 初始化时指定超时
langfuse = Langfuse(
    public_key="pk-lf-xxx",
    secret_key="sk-lf-xxx",
    host="http://localhost:3000",
    timeout=30,  # 增加超时时间
)

# docker-compose.yml - Agent 服务与 Langfuse 在同一网络
services:
  my-agent:
    build: .
    environment:
      # 使用 Docker 服务名而非 localhost
      LANGFUSE_HOST: http://langfuse-web:3000
    depends_on:
      - langfuse-web
    networks:
      - langfuse-net

  langfuse-web:
    image: langfuse/langfuse:3
    # ... 其他配置
    networks:
      - langfuse-net

networks:
  langfuse-net:
    driver: bridge
                              
                                  

"""
Opik 连接诊断脚本
"""
import os
from opik import Opik

def diagnose_opik():
    """诊断 Opik 连接问题"""
    print("🔍 Opik 连接诊断")
    print("-" * 40)
    
    # 检查环境变量
    url = os.getenv("OPIK_URL_OVERRIDE", "未设置")
    api_key = os.getenv("OPIK_API_KEY", "未设置")
    workspace = os.getenv("OPIK_WORKSPACE", "default")
    
    print(f"  URL: {url}")
    print(f"  API Key: {'✅ 已设置' if api_key != '未设置' else '❌ 未设置'}")
    print(f"  Workspace: {workspace}")
    
    # 尝试连接
    try:
        client = Opik(project_name="diagnostic-test")
        print(f"\n  ✅ 连接成功！")
        print(f"  项目: {client.project_name}")
    except Exception as e:
        print(f"\n  ❌ 连接失败: {type(e).__name__}: {e}")
        
        # 给出建议
        if "Connection" in str(e):
            print("  💡 建议：检查 Opik 服务是否启动")
            print("     运行: docker compose ps")
        elif "401" in str(e) or "Unauthorized" in str(e):
            print("  💡 建议：检查 API Key 是否正确")
        elif "timeout" in str(e).lower():
            print("  💡 建议：检查网络连接或增加超时时间")

if __name__ == "__main__":
    diagnose_opik()

┌─────────────────────────────────────────────────────────┐
│           Agent 可观测性排查 Checklist                    │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  □ 1. 服务状态检查                                       │
│    □ Docker 容器全部 Up (healthy)                        │
│    □ Web UI 可正常访问                                   │
│    □ /api/public/health 返回 200                         │
│                                                         │
│  □ 2. 网络连通性                                         │
│    □ Agent → Langfuse 端口可达                           │
│    □ Agent → Opik 端口可达                               │
│    □ DNS 解析正常（非 localhost 场景）                    │
│                                                         │
│  □ 3. 认证配置                                           │
│    □ Public Key / Secret Key 正确                        │
│    □ API Key 未过期                                      │
│    □ 项目名/工作空间名匹配                                │
│                                                         │
│  □ 4. SDK 配置                                           │
│    □ 环境变量已加载（load_dotenv）                        │
│    □ flush() 在退出前被调用                               │
│    □ SDK 版本与服务端兼容                                 │
│                                                         │
│  □ 5. 资源检查                                           │
│    □ 磁盘空间充足（>1GB）                                │
│    □ 内存充足（>2GB 可用）                               │
│    □ ClickHouse / PostgreSQL 连接数未耗尽                 │
│                                                         │
│  □ 6. 日志检查                                           │
│    □ langfuse-worker 日志无 ERROR                        │
│    □ Agent 应用日志无 SDK 异常                            │
│    □ Docker 日志无 OOM Kill                              │
│                                                         │
└─────────────────────────────────────────────────────────┘

"""
Langfuse 评分示例
"""
from langfuse import Langfuse

langfuse = Langfuse()

# ---- 方式 1：数值评分（0-1 或自定义范围） ----
langfuse.score(
    trace_id="trace-abc-123",
    name="answer_relevance",     # 评分维度名称
    value=0.92,                  # 数值
    comment="回答高度相关，但缺少具体数据",
    data_type="NUMERIC"
)

# ---- 方式 2：分类评分 ----
langfuse.score(
    trace_id="trace-abc-123",
    name="task_status",
    value="completed",           # 分类值
    data_type="CATEGORICAL"
)

# ---- 方式 3：布尔评分 ----
langfuse.score(
    trace_id="trace-abc-123",
    name="user_satisfied",
    value=True,
    data_type="BOOLEAN"
)

"""
LLM-as-Judge 自动评估流水线
用 GPT-4o 评估 Agent 输出质量
"""
from langfuse import Langfuse, observe
from langfuse.openai import openai
import json

langfuse = Langfuse()

EVALUATION_PROMPT = """
你是一个 AI 输出质量评估专家。请评估以下 Agent 回答的质量。

用户问题：{question}
Agent 回答：{answer}

请从以下维度评分（每个维度 1-5 分）：
1. 相关性（relevance）：回答是否切题
2. 完整性（completeness）：是否覆盖了问题的所有方面
3. 准确性（accuracy）：信息是否准确
4. 清晰度（clarity）：表达是否清晰易懂

请以 JSON 格式返回：
{{"relevance": <1-5>, "completeness": <1-5>, "accuracy": <1-5>, "clarity": <1-5>, "overall": <1-5>, "reasoning": "<简要说明>"}}
"""


@observe(name="evaluate_response")
def evaluate_agent_output(trace_id: str, question: str, answer: str):
    """
    使用 LLM 自动评估 Agent 输出
    并将评分写回 Langfuse
    """
    # 调用评估 LLM
    response = openai.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": EVALUATION_PROMPT.format(
                question=question,
                answer=answer
            )
        }],
        temperature=0,  # 评估需要确定性
        response_format={"type": "json_object"}
    )
    
    # 解析评估结果
    eval_result = json.loads(response.choices[0].message.content)
    
    # 将各维度评分写入 Langfuse
    for dimension, score_value in eval_result.items():
        if dimension != "reasoning" and isinstance(score_value, (int, float)):
            langfuse.score(
                trace_id=trace_id,
                name=f"eval_{dimension}",
                value=score_value / 5.0,  # 归一化到 0-1
                comment=eval_result.get("reasoning", ""),
                data_type="NUMERIC"
            )
    
    return eval_result


# ---- 批量评估示例 ----
def batch_evaluate(limit=50):
    """批量评估最近的 Traces"""
    traces = langfuse.fetch_traces(limit=limit, tags=["needs-evaluation"])
    
    for trace in traces.data:
        question = trace.input  # 用户输入
        answer = trace.output   # Agent 输出
        
        if question and answer:
            result = evaluate_agent_output(
                trace_id=trace.id,
                question=str(question),
                answer=str(answer)
            )
            print(f"✅ 评估完成: {trace.id} → overall={result['overall']}")
    
    langfuse.flush()

"""
Opik 内置评估指标使用示例
"""
from opik.evaluation import evaluate
from opik.evaluation.metrics import (
    HallucinationMetric,      # 幻觉检测
    AnswerRelevanceMetric,    # 回答相关性
    ContextPrecisionMetric,   # 上下文精确度（RAG）
    ContextRecallMetric,      # 上下文召回率（RAG）
    ModerationMetric,         # 内容审核
    GEqualsReference,         # 与参考答案对比
)

# 定义评估数据集
eval_dataset = [
    {
        "input": "什么是量子计算？",
        "output": "量子计算是利用量子力学原理进行计算的技术...",
        "expected_output": "量子计算是基于量子比特的计算范式...",
        "context": ["量子计算相关资料..."]
    },
    # ... 更多样本
]

# 运行评估
results = evaluate(
    dataset=eval_dataset,
    metrics=[
        HallucinationMetric(),
        AnswerRelevanceMetric(),
        ContextPrecisionMetric(),
    ],
    project_name="agent-evaluation",
    experiment_name="v2.1-baseline"
)

print(f"评估完成！结果已保存到 Opik 项目")
print(f"平均幻觉分数: {results['hallucination']['mean']}")
print(f"平均相关性分数: {results['answer_relevance']['mean']}")

"""
自定义 Opik 评估指标
"""
from opik.evaluation.metrics import BaseMetric, ScoreResult


class AgentEfficiencyMetric(BaseMetric):
    """
    自定义指标：Agent 效率评分
    评估 Agent 是否用最少步骤完成任务
    """
    
    def __init__(self, max_expected_steps: int = 3):
        self.max_expected_steps = max_expected_steps
    
    def score(self, output: str, metadata: dict = None, **kwargs) -> ScoreResult:
        """
        计算效率分数
        """
        # 从 metadata 中获取实际步数
        actual_steps = metadata.get("total_steps", 1) if metadata else 1
        
        # 效率 = 期望步数 / 实际步数（越高越好）
        if actual_steps <= self.max_expected_steps:
            score = 1.0
        else:
            score = self.max_expected_steps / actual_steps
        
        return ScoreResult(
            name="agent_efficiency",
            value=score,
            reason=f"实际 {actual_steps} 步，期望 ≤{self.max_expected_steps} 步"
        )


class ToolUsageCorrectness(BaseMetric):
    """
    自定义指标：工具使用正确性
    检查 Agent 是否选择了正确的工具
    """
    
    def score(self, output: str, input: str, metadata: dict = None, **kwargs) -> ScoreResult:
        # 实际场景中这里可以调用 LLM 判断
        tools_used = metadata.get("tools_used", []) if metadata else []
        expected_tools = metadata.get("expected_tools", []) if metadata else []
        
        if not expected_tools:
            return ScoreResult(name="tool_correctness", value=1.0, reason="无预期工具")
        
        # 计算匹配度
        correct = sum(1 for t in tools_used if t in expected_tools)
        score = correct / len(expected_tools)
        
        return ScoreResult(
            name="tool_correctness",
            value=score,
            reason=f"使用了 {tools_used}，期望 {expected_tools}"
        )

"""
客服 Agent 质量评分流水线
文件：cs_agent_evaluation_pipeline.py
"""
from langfuse import Langfuse, observe
from opik import track
from opik.evaluation.metrics import HallucinationMetric, AnswerRelevanceMetric
import json

langfuse_client = Langfuse()


@observe(name="cs-evaluation-pipeline")
@track(name="cs-evaluation-pipeline")
def run_cs_evaluation(date_range: str = "last_24h"):
    """
    客服 Agent 评估流水线
    
    流程：
    1. 拉取指定时间范围的客服对话 Traces
    2. 对每条 Trace 运行多维度评估
    3. 将评分写回 Langfuse
    4. 生成汇总报告
    """
    # 步骤 1：获取待评估的 Traces
    traces = langfuse_client.fetch_traces(
        limit=200,
        tags=["customer-support"]
    )
    
    results_summary = {
        "total_evaluated": 0,
        "avg_relevance": 0,
        "avg_hallucination": 0,
        "avg_satisfaction": 0,
        "flagged_for_review": []
    }
    
    for trace in traces.data:
        if not trace.output:
            continue
        
        # 步骤 2：运行评估
        question = str(trace.input) if trace.input else ""
        answer = str(trace.output) if trace.output else ""
        
        # LLM 评估（相关性）
        relevance_score = evaluate_relevance(question, answer)
        
        # 幻觉检测
        hallucination_score = detect_hallucination(answer)
        
        # 步骤 3：写回评分
        langfuse_client.score(
            trace_id=trace.id,
            name="auto_relevance",
            value=relevance_score,
            data_type="NUMERIC"
        )
        
        langfuse_client.score(
            trace_id=trace.id,
            name="auto_hallucination",
            value=hallucination_score,
            data_type="NUMERIC"
        )
        
        # 标记需要人工审核的低分 Trace
        if relevance_score < 0.5 or hallucination_score > 0.7:
            results_summary["flagged_for_review"].append(trace.id)
            langfuse_client.score(
                trace_id=trace.id,
                name="needs_human_review",
                value=True,
                data_type="BOOLEAN"
            )
        
        results_summary["total_evaluated"] += 1
    
    # 步骤 4：输出报告
    print(json.dumps(results_summary, ensure_ascii=False, indent=2))
    langfuse_client.flush()
    
    return results_summary


def evaluate_relevance(question: str, answer: str) -> float:
    """评估回答相关性（0-1）"""
    # 简化实现，实际使用 LLM 评估
    if not question or not answer:
        return 0.0
    # 这里调用评估 LLM
    return 0.85  # 示例返回值


def detect_hallucination(answer: str) -> float:
    """检测幻觉程度（0=无幻觉，1=严重幻觉）"""
    # 简化实现
    return 0.1  # 示例返回值


if __name__ == "__main__":
    run_cs_evaluation()

"""
PII（个人身份信息）过滤中间件
在数据发送到 Langfuse/Opik 之前进行脱敏
文件：pii_filter.py
"""
import re
from typing import Any, Dict


class PIIFilter:
    """
    PII 过滤器
    在 Trace 数据发送到可观测平台之前进行脱敏处理
    """
    
    # 正则表达式模式
    PATTERNS = {
        "phone": re.compile(r'1[3-9]\d{9}'),
        "email": re.compile(r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}'),
        "id_card": re.compile(r'\d{17}[\dXx]'),
        "bank_card": re.compile(r'\d{16,19}'),
        "ip_address": re.compile(r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}'),
    }
    
    def __init__(self, enabled: bool = True, custom_patterns: Dict = None):
        self.enabled = enabled
        if custom_patterns:
            self.PATTERNS.update(custom_patterns)
    
    def sanitize(self, data: Any) -> Any:
        """
        递归清洗数据中的 PII
        """
        if not self.enabled:
            return data
        
        if isinstance(data, str):
            return self._sanitize_string(data)
        elif isinstance(data, dict):
            return {k: self.sanitize(v) for k, v in data.items()}
        elif isinstance(data, list):
            return [self.sanitize(item) for item in data]
        return data
    
    def _sanitize_string(self, text: str) -> str:
        """清洗字符串中的 PII"""
        # 手机号：138****5678
        text = self.PATTERNS["phone"].sub(
            lambda m: m.group()[:3] + "****" + m.group()[7:], text
        )
        
        # 邮箱：j***@example.com
        text = self.PATTERNS["email"].sub(
            lambda m: m.group()[0] + "***@" + m.group().split("@")[1], text
        )
        
        # 身份证：110***********1234
        text = self.PATTERNS["id_card"].sub(
            lambda m: m.group()[:3] + "*" * 11 + m.group()[-4:], text
        )
        
        # 银行卡：************5678
        text = self.PATTERNS["bank_card"].sub(
            lambda m: "*" * (len(m.group()) - 4) + m.group()[-4:], text
        )
        
        return text


# ---- 使用示例 ----
pii_filter = PIIFilter(enabled=True)

# 在发送到 Langfuse 之前过滤
original_data = {
    "user_query": "我的手机号是13812345678，请帮我查询订单",
    "user_email": "zhangsan@example.com",
    "context": "用户身份证：110101199001011234"
}

sanitized_data = pii_filter.sanitize(original_data)
print(sanitized_data)
# 输出：
# {
#   "user_query": "我的手机号是138****5678，请帮我查询订单",
#   "user_email": "z***@example.com",
#   "context": "用户身份证：110***********1234"
# }

from langfuse import observe, langfuse_context
from pii_filter import PIIFilter

pii_filter = PIIFilter(enabled=True)

@observe()
def traced_agent(user_input: str) -> str:
    """带 PII 过滤的 Agent 追踪"""
    
    # 脱敏后再记录到 Trace
    sanitized_input = pii_filter.sanitize(user_input)
    langfuse_context.update_current_observation(
        input=sanitized_input  # 只记录脱敏后的数据
    )
    
    # 实际处理使用原始数据
    result = process_query(user_input)
    
    # 输出也脱敏
    sanitized_output = pii_filter.sanitize(result)
    langfuse_context.update_current_observation(
        output=sanitized_output
    )
    
    return result

"""
性能优化配置
"""
import os

# Langfuse 批量发送配置
os.environ["LANGFUSE_FLUSH_AT"] = "50"       # 累积 50 条后批量发送
os.environ["LANGFUSE_FLUSH_INTERVAL"] = "5"  # 或每 5 秒发送一次

# Opik 配置
os.environ["OPIK_BATCH_SIZE"] = "100"        # 批量大小
os.environ["OPIK_FLUSH_INTERVAL"] = "3"      # 刷新间隔

"""
采样策略：只记录部分请求
"""
import random
from langfuse import observe, langfuse_context

SAMPLE_RATE = 0.1  # 10% 采样率


@observe()
def agent_with_sampling(user_input: str) -> str:
    """带采样的 Agent"""
    
    # 决定是否采样
    should_trace = random.random() < SAMPLE_RATE
    
    if not should_trace:
        # 不追踪：直接执行，不记录
        # 注意：需要临时禁用追踪
        langfuse_context.update_current_trace(
            metadata={"sampled": False}
        )
    
    result = process_query(user_input)
    
    if should_trace:
        langfuse_context.update_current_trace(
            metadata={"sampled": True, "sample_rate": SAMPLE_RATE}
        )
    
    return result

生产环境推荐架构：

                    ┌─────────────┐
                    │   Nginx     │  ← 负载均衡 + SSL
                    │  (反向代理)  │
                    └──────┬──────┘
                           │
              ┌────────────┼────────────┐
              │            │            │
     ┌────────┴───┐ ┌─────┴─────┐ ┌───┴────────┐
     │langfuse-web│ │langfuse-web│ │langfuse-web│  ← 多副本
     │  (实例 1)  │ │  (实例 2)  │ │  (实例 3)  │
     └────────────┘ └───────────┘ └────────────┘
              │            │            │
     ┌────────┴────────────┴────────────┴────────┐
     │              PostgreSQL (主从)              │
     │              ClickHouse (集群)             │
     │              Redis (Sentinel)              │
     └───────────────────────────────────────────┘

"""
使用 Langfuse 管理 Prompt 版本
"""
from langfuse import Langfuse

langfuse = Langfuse()

# ---- 创建 Prompt 版本 ----
# 版本 1：初始版本
langfuse.create_prompt(
    name="customer-support-agent",
    prompt="你是一个客服助手。请回答用户的问题。",
    labels=["v1"],
    config={
        "model": "gpt-4o",
        "temperature": 0.7,
        "max_tokens": 1000
    }
)

# 版本 2：优化后（基于观测数据）
langfuse.create_prompt(
    name="customer-support-agent",
    prompt="""你是一个专业的客服助手。

规则：
1. 先确认理解用户问题，再给出解答
2. 如果不确定答案，明确告知用户并建议转人工
3. 回答简洁，不超过 200 字
4. 不要编造政策或承诺
5. 如果用户情绪激动，先表示理解再解决问题

禁止：
- 不要重复用户的完整问题
- 不要使用"作为AI"等措辞
- 不要一次给出超过 3 个建议
""",
    labels=["v2", "production"],  # 标记为生产版本
    config={
        "model": "gpt-4o",
        "temperature": 0.3,  # 降低随机性
        "max_tokens": 500    # 限制输出长度
    }
)

# ---- 在 Agent 中使用版本化 Prompt ----
@observe()
def agent_with_managed_prompt(user_query: str) -> str:
    """使用 Langfuse 管理的 Prompt"""
    
    # 获取当前生产版本的 Prompt
    prompt = langfuse.get_prompt(
        name="customer-support-agent",
        label="production"  # 获取标记为 production 的版本
    )
    
    # 使用 Prompt
    system_message = prompt.prompt  # 获取 Prompt 文本
    config = prompt.config          # 获取配置
    
    response = openai.chat.completions.create(
        model=config.get("model", "gpt-4o"),
        messages=[
            {"role": "system", "content": system_message},
            {"role": "user", "content": user_query}
        ],
        temperature=config.get("temperature", 0.7),
        max_tokens=config.get("max_tokens", 1000)
    )
    
    return response.choices[0].message.content

"""
Prompt A/B 测试框架
"""
from langfuse import Langfuse, observe
import random

langfuse = Langfuse()


@observe()
def agent_with_ab_test(user_query: str, user_id: str) -> str:
    """
    A/B 测试：50% 用户使用 v1 Prompt，50% 使用 v2
    """
    # 基于 user_id 哈希决定分组（保证同一用户始终在同一组）
    group = "A" if hash(user_id) % 2 == 0 else "B"
    
    # 获取对应版本的 Prompt
    prompt_label = "v1" if group == "A" else "v2"
    prompt = langfuse.get_prompt(
        name="customer-support-agent",
        label=prompt_label
    )
    
    # 记录分组信息到 Trace
    langfuse_context.update_current_trace(
        metadata={
            "ab_group": group,
            "prompt_version": prompt_label
        },
        tags=[f"ab-test", f"group-{group}"]
    )
    
    # 执行 Agent
    response = openai.chat.completions.create(
        model=prompt.config.get("model", "gpt-4o"),
        messages=[
            {"role": "system", "content": prompt.prompt},
            {"role": "user", "content": user_query}
        ]
    )
    
    return response.choices[0].message.content


# ---- 分析 A/B 测试结果 ----
def analyze_ab_results():
    """分析 A/B 测试数据"""
    traces_a = langfuse.fetch_traces(tags=["group-A"], limit=500)
    traces_b = langfuse.fetch_traces(tags=["group-B"], limit=500)
    
    # 计算各组的平均评分
    scores_a = [t.scores for t in traces_a.data if t.scores]
    scores_b = [t.scores for t in traces_b.data if t.scores]
    
    print(f"Group A (v1): {len(traces_a.data)} traces")
    print(f"Group B (v2): {len(traces_b.data)} traces")
    # 进一步分析评分差异...

┌─────────────────────────────────────────────────────────────┐
│ 第 1 步：观测发现问题                                         │
│ • Langfuse Dashboard 显示 tool_call_error 占比 12%          │
│ • 主要集中在 "order_lookup" 工具                             │
├─────────────────────────────────────────────────────────────┤
│ 第 2 步：Trace 分析定位根因                                   │
│ • 查看失败 Trace 的 LLM 输出                                 │
│ • 发现 Agent 生成的参数格式错误：                             │
│   期望 {"order_id": "ORD-12345"}                            │
│   实际 {"order_id": 12345}（数字而非字符串）                  │
├─────────────────────────────────────────────────────────────┤
│ 第 3 步：修改 Prompt                                          │
│ • 在工具描述中增加参数格式示例                                 │
│ • 添加 "order_id 必须是字符串格式，如 'ORD-12345'"           │
├─────────────────────────────────────────────────────────────┤
│ 第 4 步：A/B 测试验证                                         │
│ • 50% 流量使用新 Prompt                                      │
│ • 运行 3 天收集数据                                          │
├─────────────────────────────────────────────────────────────┤
│ 第 5 步：评估效果                                             │
│ • 工具调用失败率从 12% 降至 2.3%                             │
│ • 整体用户满意度评分提升 0.15                                 │
├─────────────────────────────────────────────────────────────┤
│ 第 6 步：全量发布                                             │
│ • 将新 Prompt 标记为 production                              │
│ • 持续监控 1 周确认稳定                                      │
└─────────────────────────────────────────────────────────────┘

"""
提示词迭代 CI/CD 流水线（概念代码）
文件：prompt_ci_cd.py
"""

def prompt_optimization_pipeline():
    """
    提示词优化流水线
    建议每周执行一次
    """
    
    # 阶段 1：数据收集
    print("📊 阶段 1：收集过去一周的观测数据...")
    weekly_traces = fetch_weekly_traces()
    
    # 阶段 2：问题识别
    print("🔍 阶段 2：识别 Top 5 问题模式...")
    issues = identify_top_issues(weekly_traces)
    for issue in issues:
        print(f"  - {issue['type']}: {issue['count']} 次, 影响 {issue['impact']}")
    
    # 阶段 3：生成优化建议
    print("💡 阶段 3：生成 Prompt 优化建议...")
    suggestions = generate_prompt_suggestions(issues)
    
    # 阶段 4：自动评估新 Prompt
    print("🧪 阶段 4：在评估数据集上测试新 Prompt...")
    eval_results = evaluate_new_prompt(suggestions)
    
    # 阶段 5：决策
    if eval_results["improvement"] > 0.05:  # 提升超过 5%
        print("✅ 阶段 5：效果显著提升，建议发布新版本")
        publish_new_prompt(suggestions)
    else:
        print("⚠️ 阶段 5：提升不显著，继续观察")
    
    print("\n🎉 流水线执行完毕！")

# ❌ 错误：@opik_track 在外层，Langfuse 无法正确嵌套
@opik_track
@langfuse_observe()
def my_function():
    ...

# ✅ 正确：@langfuse_observe 在外层
@langfuse_observe()
@opik_track
def my_function():
    ...

# ❌ 错误：asyncio 中直接调用可能丢失上下文
import asyncio

@observe()
async def async_agent(query: str):
    result = await some_async_call()  # 上下文可能丢失
    return result

# ✅ 正确：使用 Langfuse 的异步支持
from langfuse import observe

@observe()  # Langfuse v3 原生支持 async
async def async_agent(query: str):
    result = await some_async_call()
    return result
# 注意：确保 langfuse SDK 版本 >= 2.0

# ❌ 错误：先 import langfuse，后加载 .env
from langfuse import Langfuse  # 此时环境变量还未加载！
from dotenv import load_dotenv
load_dotenv()  # 太晚了！

# ✅ 正确：先加载环境变量
from dotenv import load_dotenv
load_dotenv()  # 必须在 import langfuse 之前

from langfuse import Langfuse, observe

# ❌ 错误：容器内使用 localhost
LANGFUSE_HOST = "http://localhost:3000"  # 容器内的 localhost 不是宿主机！

# ✅ 正确：使用宿主机 IP 或 Docker 网络别名
LANGFUSE_HOST = "http://host.docker.internal:3000"  # Docker Desktop
LANGFUSE_HOST = "http://172.17.0.1:3000"            # Linux Docker 默认网桥
LANGFUSE_HOST = "http://langfuse-web:3000"          # 同一 Compose 网络

# ❌ 错误：一次性获取所有 Traces
traces = langfuse.fetch_traces(limit=100000)  # 内存爆炸！

# ✅ 正确：分页获取
page = 1
while True:
    traces = langfuse.fetch_traces(limit=100, page=page)
    if not traces.data:
        break
    process(traces.data)
    page += 1

# 检查 ClickHouse 磁盘使用
docker exec clickhouse df -h

# 清理旧数据（保留最近 30 天）
docker exec clickhouse clickhouse-client --query "
    ALTER TABLE traces DELETE WHERE timestamp < now() - INTERVAL 30 DAY
"

# 在 docker-compose.yml 中限制数据量
# 添加 ClickHouse 配置：
# <merge_tree>
#   <max_bytes_to_keep>10737418240</max_bytes_to_keep>  <!-- 10GB -->
# </merge_tree>

"""
使用统一的 request_id 关联双平台数据
"""
import uuid
from langfuse import observe, langfuse_context
from opik import track, opik_context


@observe()
@track
def unified_agent(query: str) -> str:
    # 生成统一的请求 ID
    request_id = str(uuid.uuid4())
    
    # 在两个平台都记录此 ID
    langfuse_context.update_current_trace(
        metadata={"unified_request_id": request_id}
    )
    opik_context.update_current_trace(
        metadata={"unified_request_id": request_id}
    )
    
    # ... Agent 逻辑 ...
    return result

第 1 周：基础搭建
├── 部署 Langfuse 服务端
├── 安装 Opik SDK
└── 在现有 Agent 中嵌入基础追踪

第 2 周：深度集成
├── 完善嵌套追踪（工具调用、多步推理）
├── 配置关键指标监控
└── 建立每日运行报告

第 3 周：评估体系
├── 构建 LLM-as-Judge 评估流水线
├── 配置 Opik 自动评估指标
└── 建立质量基线

第 4 周：闭环优化
├── 基于观测数据优化 Prompt
├── A/B 测试验证效果
└── 建立持续迭代机制

# ============================================
# Langfuse v3 Docker Compose 配置
# 文件：docker-compose.yml
# 适用：Fedora 39/40/41
# ============================================
version: "3.9"

services:
  # ---- Langfuse Web 服务（API + UI） ----
  langfuse-web:
    image: langfuse/langfuse:3
    container_name: langfuse-web
    restart: unless-stopped
    depends_on:
      db:
        condition: service_healthy
      clickhouse:
        condition: service_healthy
      redis:
        condition: service_healthy
    ports:
      - "3000:3000"
    environment:
      # 数据库
      DATABASE_URL: postgresql://postgres:postgres@db:5432/postgres
      DIRECT_URL: postgresql://postgres:postgres@db:5432/postgres
      # ClickHouse
      CLICKHOUSE_URL: http://clickhouse:8123
      CLICKHOUSE_MIGRATION_URL: clickhouse://clickhouse:9000
      CLICKHOUSE_USER: default
      CLICKHOUSE_PASSWORD: clickhouse
      # Redis
      REDIS_HOST: redis
      REDIS_PORT: "6379"
      REDIS_AUTH: redis
      # MinIO
      MINIO_ENDPOINT: minio
      MINIO_PORT: "9000"
      MINIO_ACCESS_KEY: minio
      MINIO_SECRET_KEY: miniosecret
      MINIO_BUCKET: langfuse
      # 安全密钥（生产环境必须修改！）
      NEXTAUTH_SECRET: ${NEXTAUTH_SECRET:-change-me-in-production}
      SALT: ${SALT:-change-me-in-production}
      ENCRYPTION_KEY: ${ENCRYPTION_KEY:-0000000000000000000000000000000000000000000000000000000000000000}
      # 应用配置
      NEXTAUTH_URL: http://localhost:3000
      TELEMETRY_ENABLED: "false"
      LANGFUSE_ENABLE_EXPERIMENTAL_FEATURES: "true"
    networks:
      - langfuse-net

  # ---- Langfuse Worker（异步处理） ----
  langfuse-worker:
    image: langfuse/langfuse-worker:3
    container_name: langfuse-worker
    restart: unless-stopped
    depends_on:
      db:
        condition: service_healthy
      clickhouse:
        condition: service_healthy
      redis:
        condition: service_healthy
    environment:
      DATABASE_URL: postgresql://postgres:postgres@db:5432/postgres
      DIRECT_URL: postgresql://postgres:postgres@db:5432/postgres
      CLICKHOUSE_URL: http://clickhouse:8123
      CLICKHOUSE_MIGRATION_URL: clickhouse://clickhouse:9000
      CLICKHOUSE_USER: default
      CLICKHOUSE_PASSWORD: clickhouse
      REDIS_HOST: redis
      REDIS_PORT: "6379"
      REDIS_AUTH: redis
      MINIO_ENDPOINT: minio
      MINIO_PORT: "9000"
      MINIO_ACCESS_KEY: minio
      MINIO_SECRET_KEY: miniosecret
      MINIO_BUCKET: langfuse
      NEXTAUTH_SECRET: ${NEXTAUTH_SECRET:-change-me-in-production}
      SALT: ${SALT:-change-me-in-production}
      ENCRYPTION_KEY: ${ENCRYPTION_KEY:-0000000000000000000000000000000000000000000000000000000000000000}
    networks:
      - langfuse-net

  # ---- PostgreSQL（元数据存储） ----
  db:
    image: postgres:16-alpine
    container_name: langfuse-db
    restart: unless-stopped
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: postgres
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres"]
      interval: 5s
      timeout: 5s
      retries: 5
    networks:
      - langfuse-net

  # ---- ClickHouse（Trace 数据存储） ----
  clickhouse:
    image: clickhouse/clickhouse-server:24
    container_name: langfuse-clickhouse
    restart: unless-stopped
    environment:
      CLICKHOUSE_USER: default
      CLICKHOUSE_PASSWORD: clickhouse
    volumes:
      - clickhouse_data:/var/lib/clickhouse
    healthcheck:
      test: ["CMD", "wget", "--no-verbose", "--tries=1", "--spider", "http://localhost:8123/ping"]
      interval: 5s
      timeout: 5s
      retries: 5
    networks:
      - langfuse-net

  # ---- Redis（缓存 + 消息队列） ----
  redis:
    image: redis:7-alpine
    container_name: langfuse-redis
    restart: unless-stopped
    command: redis-server --requirepass redis
    volumes:
      - redis_data:/data
    healthcheck:
      test: ["CMD", "redis-cli", "-a", "redis", "ping"]
      interval: 5s
      timeout: 5s
      retries: 5
    networks:
      - langfuse-net

  # ---- MinIO（对象存储） ----
  minio:
    image: minio/minio:latest
    container_name: langfuse-minio
    restart: unless-stopped
    command: server /data --console-address ":9001"
    environment:
      MINIO_ROOT_USER: minio
      MINIO_ROOT_PASSWORD: miniosecret
    volumes:
      - minio_data:/data
    ports:
      - "9090:9000"
      - "9091:9001"
    networks:
      - langfuse-net

# ---- 数据卷 ----
volumes:
  postgres_data:
  clickhouse_data:
  redis_data:
  minio_data:

# ---- 网络 ----
networks:
  langfuse-net:
    driver: bridge

"""
============================================================
完整示例：带双平台追踪的 ReAct Agent
文件：complete_agent_example.py

运行前准备：
1. 安装依赖：pip install langfuse opik openai python-dotenv
2. 创建 .env 文件（见附录 C）
3. 确保 Langfuse 和 Opik 服务已启动

运行：python complete_agent_example.py
============================================================
"""
import os
import json
import time
import uuid
from typing import Optional, Dict, Any
from dotenv import load_dotenv

# 必须在 import langfuse 之前加载环境变量
load_dotenv()

# ---- 双平台 SDK 导入 ----
from langfuse import observe as lf_observe, langfuse_context
from langfuse.openai import openai  # Langfuse 增强的 OpenAI 客户端
from opik import track as opik_track, opik_context

# ---- PII 过滤器 ----
import re

class SimplePIIFilter:
    """简易 PII 过滤器"""
    
    PATTERNS = {
        "phone": (re.compile(r'1[3-9]\d{9}'), lambda m: m.group()[:3] + "****" + m.group()[7:]),
        "email": (re.compile(r'[\w.+-]+@[\w-]+\.[\w.]+'), lambda m: m.group()[0] + "***@" + m.group().split("@")[1]),
    }
    
    def sanitize(self, text: str) -> str:
        if not isinstance(text, str):
            return text
        for pattern, replacer in self.PATTERNS.values():
            text = pattern.sub(replacer, text)
        return text

pii_filter = SimplePIIFilter()


# ---- 工具定义 ----
AVAILABLE_TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "search_knowledge_base",
            "description": "在知识库中搜索相关信息。当用户询问事实性问题时使用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "搜索查询，必须是字符串格式"
                    },
                    "top_k": {
                        "type": "integer",
                        "description": "返回结果数量，默认 3",
                        "default": 3
                    }
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "执行数学计算。当用户需要数值计算时使用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "数学表达式，如 '(15 + 27) * 3'"
                    }
                },
                "required": ["expression"]
            }
        }
    }
]


# ---- 工具实现（带双平台追踪） ----
@lf_observe(name="tool:search_knowledge_base")
@opik_track(name="tool:search_knowledge_base")
def search_knowledge_base(query: str, top_k: int = 3) -> str:
    """知识库搜索工具"""
    time.sleep(0.2)  # 模拟搜索延迟
    
    # 模拟搜索结果
    results = [
        {
            "title": f"文档 {i+1}：关于 {query} 的说明",
            "content": f"这是关于 '{query}' 的第 {i+1} 条相关知识...",
            "relevance_score": round(0.9 - i * 0.1, 2)
        }
        for i in range(top_k)
    ]
    
    return json.dumps({"status": "success", "results": results}, ensure_ascii=False)


@lf_observe(name="tool:calculate")
@opik_track(name="tool:calculate")
def calculate(expression: str) -> str:
    """数学计算工具"""
    try:
        # 安全计算
        allowed_chars = set("0123456789+-*/.() ")
        if not all(c in allowed_chars for c in expression):
            return json.dumps({"status": "error", "message": "包含不允许的字符"})
        
        result = eval(expression, {"__builtins__": {}}, {})
        return json.dumps({"status": "success", "result": result})
    except Exception as e:
        return json.dumps({"status": "error", "message": str(e)})


# 工具注册表
TOOL_REGISTRY = {
    "search_knowledge_base": search_knowledge_base,
    "calculate": calculate,
}


# ---- Agent 主函数 ----
@lf_observe(name="react-agent")
@opik_track(name="react-agent")
def react_agent(
    user_query: str,
    user_id: str = "anonymous",
    session_id: Optional[str] = None,
    max_steps: int = 5,
    model: str = "gpt-4o"
) -> Dict[str, Any]:
    """
    ReAct Agent 主函数（带完整双平台追踪）
    
    Args:
        user_query: 用户输入
        user_id: 用户标识
        session_id: 会话 ID（多轮对话）
        max_steps: 最大推理步数
        model: LLM 模型名称
    
    Returns:
        包含 answer、metadata 的字典
    """
    # 生成统一请求 ID
    request_id = str(uuid.uuid4())
    
    # ---- 设置 Langfuse Trace 元数据 ----
    langfuse_context.update_current_trace(
        name=f"agent: {user_query[:50]}",
        user_id=user_id,
        session_id=session_id,
        tags=["react-agent", "production"],
        metadata={
            "unified_request_id": request_id,
            "model": model,
            "max_steps": max_steps,
            "agent_version": "2.0.0"
        }
    )
    
    # ---- 设置 Opik Trace 元数据 ----
    opik_context.update_current_trace(
        tags=["react-agent", "production"],
        metadata={"unified_request_id": request_id}
    )
    
    # ---- 脱敏用户输入（用于记录） ----
    sanitized_query = pii_filter.sanitize(user_query)
    
    # ---- 初始化对话 ----
    system_prompt = """你是一个智能助手 Agent。你可以使用工具来帮助用户。

规则：
1. 分析用户意图，决定是否需要使用工具
2. 如果需要工具，一次只调用一个工具
3. 根据工具返回结果决定下一步
4. 不要以相同参数重复调用同一工具
5. 最多使用 {max_steps} 步完成任务
6. 如果无法完成，诚实告知用户""".format(max_steps=max_steps)
    
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_query}
    ]
    
    # ---- ReAct 循环 ----
    steps_taken = 0
    tools_used = []
    
    for step in range(max_steps):
        steps_taken += 1
        
        # 调用 LLM
        response = openai.chat.completions.create(
            model=model,
            messages=messages,
            tools=AVAILABLE_TOOLS,
            tool_choice="auto",
            temperature=0.1,
            max_tokens=2000
        )
        
        assistant_msg = response.choices[0].message
        
        # 检查工具调用
        if assistant_msg.tool_calls:
            messages.append(assistant_msg)
            
            for tool_call in assistant_msg.tool_calls:
                func_name = tool_call.function.name
                func_args = json.loads(tool_call.function.arguments)
                
                # 记录工具使用
                tools_used.append(func_name)
                
                # 执行工具
                tool_func = TOOL_REGISTRY.get(func_name)
                if tool_func:
                    tool_result = tool_func(**func_args)
                else:
                    tool_result = json.dumps({
                        "status": "error",
                        "message": f"未知工具: {func_name}"
                    })
                
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": tool_result
                })
        else:
            # 最终答案
            final_answer = assistant_msg.content
            
            # 更新 Trace 元数据
            langfuse_context.update_current_trace(
                output=pii_filter.sanitize(final_answer),
                metadata={
                    "total_steps": steps_taken,
                    "tools_used": tools_used
                }
            )
            
            return {
                "answer": final_answer,
                "metadata": {
                    "request_id": request_id,
                    "steps": steps_taken,
                    "tools_used": tools_used,
                    "model": model
                }
            }
    
    # 超过最大步数
    fallback_answer = "抱歉，我在限定步数内未能完成此任务。请尝试简化问题或联系人工支持。"
    
    langfuse_context.update_current_trace(
        output=fallback_answer,
        metadata={"total_steps": steps_taken, "hit_limit": True}
    )
    
    return {
        "answer": fallback_answer,
        "metadata": {
            "request_id": request_id,
            "steps": steps_taken,
            "hit_limit": True
        }
    }


# ---- 运行示例 ----
if __name__ == "__main__":
    print("=" * 60)
    print("🤖 AI Agent 可观测性实战演示")
    print("=" * 60)
    
    # 测试用例
    test_queries = [
        "帮我计算 (15 + 27) * 3 等于多少",
        "搜索一下关于 Python 异步编程的最佳实践",
        "我的手机号是 13812345678，帮我查一下账户余额",
    ]
    
    session_id = f"session-{uuid.uuid4().hex[:8]}"
    
    for i, query in enumerate(test_queries):
        print(f"\n{'─'*60}")
        print(f"📝 查询 {i+1}: {query}")
        print(f"{'─'*60}")
        
        result = react_agent(
            user_query=query,
            user_id="demo-user",
            session_id=session_id
        )
        
        print(f"✅ 回答: {result['answer'][:100]}...")
        print(f"📊 元数据: 步数={result['metadata']['steps']}, "
              f"工具={result['metadata'].get('tools_used', [])}")
    
    # 确保所有数据发送完毕
    langfuse_context.flush()
    
    print(f"\n{'='*60}")
    print("✅ 所有 Trace 已发送到 Langfuse 和 Opik")
    print(f"📍 Langfuse: http://localhost:3000")
    print(f"📍 Opik: http://localhost:5173")
    print(f"{'='*60}")

# ============================================
# .env 完整模板
# ============================================

# ---- Langfuse 配置 ----
LANGFUSE_PUBLIC_KEY=pk-lf-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
LANGFUSE_SECRET_KEY=sk-lf-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
LANGFUSE_HOST=http://localhost:3000
LANGFUSE_FLUSH_AT=50
LANGFUSE_FLUSH_INTERVAL=5

# ---- Opik 配置 ----
OPIK_URL_OVERRIDE=http://localhost:5173/api
OPIK_API_KEY=your-opik-api-key
OPIK_WORKSPACE=default
OPIK_PROJECT_NAME=my-agent-project

# ---- OpenAI / LLM 配置 ----
OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxx
OPENAI_BASE_URL=https://api.openai.com/v1  # 或自定义兼容端点

# ---- 应用配置 ----
USER_ID=demo-user
AGENT_ENV=development  # development / staging / production
LOG_LEVEL=INFO

# ---- 采样配置（生产环境） ----
TRACE_SAMPLE_RATE=1.0  # 1.0 = 100% 采样，0.1 = 10% 采样

# ============================================
# Docker 服务管理
# ============================================

# 启动所有服务
cd ~/langfuse-deploy && docker compose up -d

# 停止所有服务
docker compose down

# 重启特定服务
docker compose restart langfuse-web

# 查看服务状态
docker compose ps

# 查看日志（实时）
docker compose logs -f langfuse-web
docker compose logs -f langfuse-worker
docker compose logs -f clickhouse

# 查看最近 100 行日志
docker compose logs --tail=100 langfuse-web

# 进入容器调试
docker exec -it langfuse-web /bin/sh
docker exec -it langfuse-clickhouse clickhouse-client

# 清理并重建
docker compose down -v  # 删除数据卷（慎用！）
docker compose up -d --build

# ============================================
# 健康检查
# ============================================

# Langfuse 健康检查
curl -s http://localhost:3000/api/public/health | jq

# ClickHouse 健康检查
curl -s http://localhost:8123/ping

# Redis 健康检查
docker exec langfuse-redis redis-cli -a redis ping

# PostgreSQL 健康检查
docker exec langfuse-db pg_isready -U postgres

# ============================================
# 数据管理
# ============================================

# 查看 ClickHouse 表大小
docker exec langfuse-clickhouse clickhouse-client --query "
    SELECT table, formatReadableSize(sum(bytes)) as size
    FROM system.parts
    WHERE database = 'default'
    GROUP BY table
    ORDER BY sum(bytes) DESC
"

# 查看 PostgreSQL 数据库大小
docker exec langfuse-db psql -U postgres -c "
    SELECT pg_size_pretty(pg_database_size('postgres'));
"

# 备份 PostgreSQL
docker exec langfuse-db pg_dump -U postgres postgres > backup_$(date +%Y%m%d).sql

# ============================================
# Python SDK 诊断
# ============================================

# 检查 Langfuse 连接
python -c "
from langfuse import Langfuse
lf = Langfuse()
print('✅ Langfuse 连接成功')
print(f'   Host: {lf.base_url}')
"

# 检查 Opik 连接
python -c "
from opik import Opik
client = Opik(project_name='test')
print('✅ Opik 连接成功')
"

# 查看已安装的 SDK 版本
pip show langfuse opik

# 升级 SDK
pip install --upgrade langfuse opik

# ============================================
# Fedora 系统管理
# ============================================

# 防火墙状态
sudo firewall-cmd --list-all

# SELinux 状态
getenforce

# Docker 资源使用
docker system df

# 清理未使用的 Docker 资源
docker system prune -f