http://192.168.110.207:3000

http://localhost:3000
https://cloud.langfuse.com
http://langfuse-web:3000          # Dify 和 Langfuse 不在同一个 docker network

POST /api/public/ingestion
Authorization: Basic <public_key:secret_key>

Rejected ... event(s) from the legacy /api/public/ingestion endpoint
... this Langfuse v4 deployment runs in events_only mode

浏览器
  |  问一句「公司请假流程是什么」
  v
Dify Nginx :80
  |  POST /console/api/agent/<agent_id>/chat-messages
  v
docker-api-1
  |  新 Agent 实际推理交给 agent_backend
  |  对话结束后，看 apps.tracing 是否 enabled + langfuse
  v
把 trace 任务丢进 Redis 队列 ops_trace
  v
docker-worker-1   (Celery: tasks.ops_trace_task.process_trace_tasks)
  |  从 trace_app_config 取出并解密 public_key / secret_key / host
  |  LangFuseDataTrace.add_trace()        -> type=trace-create
  |  LangFuseDataTrace.add_generation()   -> type=generation-create
  v
HTTP Basic Auth
POST http://192.168.110.207:3000/api/public/ingestion
  v
langfuse-langfuse-web-1
  |  dual 模式下写入旧表 traces / observations
  v
ClickHouse
  v
浏览器打开 Langfuse Traces 页面看到这条链路

{"enabled": true, "tracing_provider": "langfuse"}

GET  /console/api/apps/<app_id>/trace
POST /console/api/apps/<app_id>/trace
Body: {"enabled": true, "tracing_provider": "langfuse"}

GET    /console/api/apps/<app_id>/trace-config?tracing_provider=langfuse
POST   /console/api/apps/<app_id>/trace-config
PATCH  /console/api/apps/<app_id>/trace-config
DELETE /console/api/apps/<app_id>/trace-config?tracing_provider=langfuse

{
  "tracing_provider": "langfuse",
  "tracing_config": {
    "public_key": "pk-lf-...",
    "secret_key": "sk-lf-...",
    "host": "http://192.168.110.207:3000"
  }
}

/app/api/.venv/lib/python3.12/site-packages/dify_trace_langfuse/langfuse_trace.py

LANGFUSE_MIGRATION_V4_ALLOW_PREVIEW_OPT_IN=false

LANGFUSE_INIT_PROJECT_PUBLIC_KEY=pk-lf-...
LANGFUSE_INIT_PROJECT_SECRET_KEY=sk-lf-...

#!/usr/bin/env bash
set -euo pipefail

curl -s http://127.0.0.1:3000/api/public/health
# 期望: {"status":"OK","version":"4.12.0"}

#!/usr/bin/env bash
set -euo pipefail

docker ps --format 'table {{.Names}}\t{{.Status}}\t{{.Ports}}' | grep -E 'langfuse|docker-api-1|docker-worker-1'

#!/usr/bin/env bash
set -euo pipefail

docker exec docker-api-1 curl -s -m 8 http://192.168.110.207:3000/api/public/health
docker exec docker-worker-1 curl -s -m 8 http://192.168.110.207:3000/api/public/health

#!/usr/bin/env bash
set -euo pipefail

docker exec langfuse-postgres-1 psql -U postgres -d postgres -c \
  "SELECT id, public_key, note, project_id, display_secret_key FROM api_keys;"

#!/usr/bin/env bash
set -euo pipefail

python3 - <<'PY'
import uuid
print("pk-lf-" + str(uuid.uuid4()))
print("sk-lf-" + str(uuid.uuid4()))
PY

pk-lf-84a76d89-b6b4-4943-b4ea-2dee48d4cda7
sk-lf-24aa6d73-ed36-4447-9eb6-3ebd446fb298

LANGFUSE_INIT_PROJECT_PUBLIC_KEY=pk-lf-84a76d89-b6b4-4943-b4ea-2dee48d4cda7
LANGFUSE_INIT_PROJECT_SECRET_KEY=sk-lf-24aa6d73-ed36-4447-9eb6-3ebd446fb298

#!/usr/bin/env bash
set -euo pipefail

cd /Langfuse/langfuse
docker compose -p langfuse up -d langfuse-web

LANGFUSE_MIGRATION_V4_WRITE_MODE=dual
LANGFUSE_MIGRATION_V4_ALLOW_PREVIEW_OPT_IN=false

      LANGFUSE_MIGRATION_V4_WRITE_MODE: ${LANGFUSE_MIGRATION_V4_WRITE_MODE:-dual}
      LANGFUSE_MIGRATION_V4_ALLOW_PREVIEW_OPT_IN: ${LANGFUSE_MIGRATION_V4_ALLOW_PREVIEW_OPT_IN:-false}

#!/usr/bin/env bash
set -euo pipefail

cd /Langfuse/langfuse
docker compose -p langfuse up -d langfuse-web langfuse-worker

#!/usr/bin/env bash
set -euo pipefail

docker exec langfuse-langfuse-web-1 printenv LANGFUSE_MIGRATION_V4_WRITE_MODE
docker exec langfuse-langfuse-worker-1 printenv LANGFUSE_MIGRATION_V4_WRITE_MODE

from dify_api import Dify, pp

APP_ID = "40e95fd3-e7f5-44f2-a03b-dc3e60413540"
PK = "pk-lf-84a76d89-b6b4-4943-b4ea-2dee48d4cda7"
SK = "sk-lf-24aa6d73-ed36-4447-9eb6-3ebd446fb298"
HOST = "http://192.168.110.207:3000"

d = Dify("38205036@qq.com", "Difyai123456")
ok, _, _ = d.login()
print("login", ok)

# 1) 看当前开关
pp("current trace", d.get("/apps/%s/trace" % APP_ID))
pp("current config", d.get("/apps/%s/trace-config?tracing_provider=langfuse" % APP_ID))

# 2) 写入密钥。没有配置则 POST，已有则 PATCH
body = {
    "tracing_provider": "langfuse",
    "tracing_config": {
        "public_key": PK,
        "secret_key": SK,
        "host": HOST,
    },
}
cfg = d.get("/apps/%s/trace-config?tracing_provider=langfuse" % APP_ID)
if (cfg.get("json") or {}).get("has_not_configured"):
    r = d.post("/apps/%s/trace-config" % APP_ID, body=body)
    pp("create trace-config", r)
else:
    r = d.req("PATCH", "/apps/%s/trace-config" % APP_ID, body=body)
    pp("update trace-config", r)
    if r["code"] not in ("200", "201"):
        r = d.post("/apps/%s/trace-config" % APP_ID, body=body)
        pp("create after failed patch", r)

# 3) 打开总开关
r = d.post(
    "/apps/%s/trace" % APP_ID,
    body={"enabled": True, "tracing_provider": "langfuse"},
)
pp("enable tracing", r)

# 4) 回读：密钥会被打码，并出现 project_url
pp("trace after", d.get("/apps/%s/trace" % APP_ID))
pp("config after", d.get("/apps/%s/trace-config?tracing_provider=langfuse" % APP_ID))
d.close()

http://192.168.110.207:3000/project/default

{"enabled": true, "tracing_provider": "langfuse"}

#!/usr/bin/env bash
set -euo pipefail

docker exec docker-db_postgres-1 psql -U postgres -d dify -c \
  "SELECT id, name, tracing FROM apps;"

docker exec docker-db_postgres-1 psql -U postgres -d dify -c \
  "SELECT app_id, tracing_provider, is_active FROM trace_app_config;"

GET    /console/api/apps/<app_id>/trace
POST   /console/api/apps/<app_id>/trace
GET    /console/api/apps/<app_id>/trace-config?tracing_provider=langfuse
POST   /console/api/apps/<app_id>/trace-config
PATCH  /console/api/apps/<app_id>/trace-config

LANGFUSE_HOST=http://192.168.110.207:3000
LANGFUSE_PUBLIC_KEY=pk-lf-84a76d89-b6b4-4943-b4ea-2dee48d4cda7
LANGFUSE_SECRET_KEY=sk-lf-24aa6d73-ed36-4447-9eb6-3ebd446fb298

/app/api/.venv/lib/python3.12/site-packages/dify_trace_langfuse/langfuse_trace.py

#!/usr/bin/env bash
set -euo pipefail

# 1. 两个容器里文件都存在
docker exec docker-worker-1 ls -l /app/api/.venv/lib/python3.12/site-packages/dify_trace_langfuse/langfuse_trace.py
docker exec docker-api-1    ls -l /app/api/.venv/lib/python3.12/site-packages/dify_trace_langfuse/langfuse_trace.py

# 2. 看有没有已经打过补丁（有这行就不用再改）
docker exec docker-worker-1 grep -n 'auto-assign if missing' \
  /app/api/.venv/lib/python3.12/site-packages/dify_trace_langfuse/langfuse_trace.py
docker exec docker-api-1 grep -n 'auto-assign if missing' \
  /app/api/.venv/lib/python3.12/site-packages/dify_trace_langfuse/langfuse_trace.py

# 3. 定位两个函数（记下行号）
docker exec docker-worker-1 grep -n 'def add_span\|def add_generation\|body = CreateSpanBody\|body = CreateGenerationBody' \
  /app/api/.venv/lib/python3.12/site-packages/dify_trace_langfuse/langfuse_trace.py

    def add_span(self, langfuse_span_data: LangfuseSpan | None = None):
        data = filter_none_values(langfuse_span_data.model_dump()) if langfuse_span_data else {}
        try:
            body = CreateSpanBody(          # ← 不要动这一行，在它上面插入
                id=data.get("id"),

            if usage_data:
                usage = Usage(
                    input=usage_data.get("input", 0) or 0,
                    output=usage_data.get("output", 0) or 0,
                    total=usage_data.get("total", 0) or 0,
                    unit=usage_data.get("unit"),
                    input_cost=usage_data.get("inputCost"),
                    output_cost=usage_data.get("outputCost"),
                    total_cost=usage_data.get("totalCost"),
                )

            body = CreateGenerationBody(    # ← 不要动这一行，在它上面插入
                id=data.get("id"),

            # Langfuse v4 requires observation body.id; auto-assign if missing
            if not data.get("id"):
                data["id"] = str(uuid.uuid4())

    def add_span(self, langfuse_span_data: LangfuseSpan | None = None):
        data = filter_none_values(langfuse_span_data.model_dump()) if langfuse_span_data else {}
        try:
            body = CreateSpanBody(
                id=data.get("id"),
                trace_id=data.get("trace_id"),
                name=data.get("name"),
                start_time=data.get("start_time"),
                end_time=data.get("end_time"),
                input=data.get("input"),
                output=data.get("output"),
                metadata=data.get("metadata"),
                level=data.get("level"),
                status_message=data.get("status_message"),
                parent_observation_id=data.get("parent_observation_id"),
                version=data.get("version"),
            )
            event = IngestionEvent_SpanCreate(
                body=body,
                id=self._make_event_id(),
                timestamp=self._now_iso(),
            )
            self.langfuse_client.api.ingestion.batch(batch=[event])
            logger.debug("LangFuse Span created successfully")
        except Exception as e:
            raise ValueError(f"LangFuse Failed to create span: {str(e)}")

    def add_span(self, langfuse_span_data: LangfuseSpan | None = None):
        data = filter_none_values(langfuse_span_data.model_dump()) if langfuse_span_data else {}
        try:
            # Langfuse v4 requires observation body.id; auto-assign if missing
            if not data.get("id"):
                data["id"] = str(uuid.uuid4())
            body = CreateSpanBody(
                id=data.get("id"),
                trace_id=data.get("trace_id"),
                name=data.get("name"),
                start_time=data.get("start_time"),
                end_time=data.get("end_time"),
                input=data.get("input"),
                output=data.get("output"),
                metadata=data.get("metadata"),
                level=data.get("level"),
                status_message=data.get("status_message"),
                parent_observation_id=data.get("parent_observation_id"),
                version=data.get("version"),
            )
            event = IngestionEvent_SpanCreate(
                body=body,
                id=self._make_event_id(),
                timestamp=self._now_iso(),
            )
            self.langfuse_client.api.ingestion.batch(batch=[event])
            logger.debug("LangFuse Span created successfully")
        except Exception as e:
            raise ValueError(f"LangFuse Failed to create span: {str(e)}")

            # Langfuse v4 requires observation body.id; auto-assign if missing
            if not data.get("id"):
                data["id"] = str(uuid.uuid4())

    def add_generation(self, langfuse_generation_data: LangfuseGeneration | None = None):
        data = filter_none_values(langfuse_generation_data.model_dump()) if langfuse_generation_data else {}
        try:
            usage_data = data.pop("usage", None)
            usage = None
            if usage_data:
                usage = Usage(
                    input=usage_data.get("input", 0) or 0,
                    output=usage_data.get("output", 0) or 0,
                    total=usage_data.get("total", 0) or 0,
                    unit=usage_data.get("unit"),
                    input_cost=usage_data.get("inputCost"),
                    output_cost=usage_data.get("outputCost"),
                    total_cost=usage_data.get("totalCost"),
                )

            body = CreateGenerationBody(
                id=data.get("id"),
                trace_id=data.get("trace_id"),
                name=data.get("name"),
                start_time=data.get("start_time"),
                end_time=data.get("end_time"),
                model=data.get("model"),
                model_parameters=data.get("model_parameters"),
                input=data.get("input"),
                output=data.get("output"),
                usage=usage,
                metadata=data.get("metadata"),
                level=data.get("level"),
                status_message=data.get("status_message"),
                parent_observation_id=data.get("parent_observation_id"),
                version=data.get("version"),
                completion_start_time=data.get("completion_start_time"),
            )
            event = IngestionEvent_GenerationCreate(
                body=body,
                id=self._make_event_id(),
                timestamp=self._now_iso(),
            )
            self.langfuse_client.api.ingestion.batch(batch=[event])
            logger.debug("LangFuse Generation created successfully")
        except Exception as e:
            raise ValueError(f"LangFuse Failed to create generation: {str(e)}")

    def add_generation(self, langfuse_generation_data: LangfuseGeneration | None = None):
        data = filter_none_values(langfuse_generation_data.model_dump()) if langfuse_generation_data else {}
        try:
            usage_data = data.pop("usage", None)
            usage = None
            if usage_data:
                usage = Usage(
                    input=usage_data.get("input", 0) or 0,
                    output=usage_data.get("output", 0) or 0,
                    total=usage_data.get("total", 0) or 0,
                    unit=usage_data.get("unit"),
                    input_cost=usage_data.get("inputCost"),
                    output_cost=usage_data.get("outputCost"),
                    total_cost=usage_data.get("totalCost"),
                )

            # Langfuse v4 requires observation body.id; auto-assign if missing
            if not data.get("id"):
                data["id"] = str(uuid.uuid4())
            body = CreateGenerationBody(
                id=data.get("id"),
                trace_id=data.get("trace_id"),
                name=data.get("name"),
                start_time=data.get("start_time"),
                end_time=data.get("end_time"),
                model=data.get("model"),
                model_parameters=data.get("model_parameters"),
                input=data.get("input"),
                output=data.get("output"),
                usage=usage,
                metadata=data.get("metadata"),
                level=data.get("level"),
                status_message=data.get("status_message"),
                parent_observation_id=data.get("parent_observation_id"),
                version=data.get("version"),
                completion_start_time=data.get("completion_start_time"),
            )
            event = IngestionEvent_GenerationCreate(
                body=body,
                id=self._make_event_id(),
                timestamp=self._now_iso(),
            )
            self.langfuse_client.api.ingestion.batch(batch=[event])
            logger.debug("LangFuse Generation created successfully")
        except Exception as e:
            raise ValueError(f"LangFuse Failed to create generation: {str(e)}")

#!/usr/bin/env bash
set -euo pipefail

# 没有 vi 时用 sed 先确认锚点附近的原文
docker exec docker-worker-1 sed -n '/def add_span/,/def update_span/p' \
  /app/api/.venv/lib/python3.12/site-packages/dify_trace_langfuse/langfuse_trace.py | head -40

# 有编辑器就进容器改；没有就用下面的 docker cp 把文件拷出来改
docker cp docker-worker-1:/app/api/.venv/lib/python3.12/site-packages/dify_trace_langfuse/langfuse_trace.py \
  /tmp/langfuse_trace.py

# 在 /tmp/langfuse_trace.py 里按 5.5 改两处，保存后再拷回去
docker cp /tmp/langfuse_trace.py \
  docker-worker-1:/app/api/.venv/lib/python3.12/site-packages/dify_trace_langfuse/langfuse_trace.py

docker cp /tmp/langfuse_trace.py \
  docker-api-1:/app/api/.venv/lib/python3.12/site-packages/dify_trace_langfuse/langfuse_trace.py

from pathlib import Path

p = Path("/app/api/.venv/lib/python3.12/site-packages/dify_trace_langfuse/langfuse_trace.py")
t = p.read_text()
marker = 'Langfuse v4 requires observation body.id; auto-assign if missing'
if marker in t:
    print("already patched")
    raise SystemExit(0)

old_span = """        data = filter_none_values(langfuse_span_data.model_dump()) if langfuse_span_data else {}
        try:
            body = CreateSpanBody(
                id=data.get("id"),
"""
new_span = """        data = filter_none_values(langfuse_span_data.model_dump()) if langfuse_span_data else {}
        try:
            # Langfuse v4 requires observation body.id; auto-assign if missing
            if not data.get("id"):
                data["id"] = str(uuid.uuid4())
            body = CreateSpanBody(
                id=data.get("id"),
"""
old_gen = """            usage_data = data.pop("usage", None)
            usage = None
            if usage_data:
                usage = Usage(
                    input=usage_data.get("input", 0) or 0,
                    output=usage_data.get("output", 0) or 0,
                    total=usage_data.get("total", 0) or 0,
                    unit=usage_data.get("unit"),
                    input_cost=usage_data.get("inputCost"),
                    output_cost=usage_data.get("outputCost"),
                    total_cost=usage_data.get("totalCost"),
                )

            body = CreateGenerationBody(
                id=data.get("id"),
"""
new_gen = """            usage_data = data.pop("usage", None)
            usage = None
            if usage_data:
                usage = Usage(
                    input=usage_data.get("input", 0) or 0,
                    output=usage_data.get("output", 0) or 0,
                    total=usage_data.get("total", 0) or 0,
                    unit=usage_data.get("unit"),
                    input_cost=usage_data.get("inputCost"),
                    output_cost=usage_data.get("outputCost"),
                    total_cost=usage_data.get("totalCost"),
                )

            # Langfuse v4 requires observation body.id; auto-assign if missing
            if not data.get("id"):
                data["id"] = str(uuid.uuid4())
            body = CreateGenerationBody(
                id=data.get("id"),
"""
if old_span not in t:
    raise SystemExit("span block not found")
if old_gen not in t:
    raise SystemExit("gen block not found")
t = t.replace(old_span, new_span, 1).replace(old_gen, new_gen, 1)
p.write_text(t)
print("patched ok")

#!/usr/bin/env bash
set -euo pipefail

docker cp /tmp/patch_lf_id.py docker-worker-1:/tmp/patch_lf_id.py
docker exec docker-worker-1 python /tmp/patch_lf_id.py
# 期望: patched ok

docker cp /tmp/patch_lf_id.py docker-api-1:/tmp/patch_lf_id.py
docker exec docker-api-1 python /tmp/patch_lf_id.py
# 期望: patched ok

docker restart docker-worker-1

docker logs -f docker-worker-1

celery@6a47a2527234 ready.

#!/usr/bin/env bash
set -euo pipefail

# 文件里有补丁
docker exec docker-worker-1 grep -n 'auto-assign if missing' \
  /app/api/.venv/lib/python3.12/site-packages/dify_trace_langfuse/langfuse_trace.py
docker exec docker-api-1 grep -n 'auto-assign if missing' \
  /app/api/.venv/lib/python3.12/site-packages/dify_trace_langfuse/langfuse_trace.py

# worker 是补丁之后才 ready 的（看重启时间）
docker inspect docker-worker-1 --format '{{.State.StartedAt}} {{.State.Status}}'

请用一两句话说明公司请假流程，这是 Langfuse 完整链路验收。

http://192.168.110.207:3000/project/default/traces/1f339a2f-bbda-4042-8e99-1d66bbc6faba

#!/usr/bin/env bash
set -euo pipefail

curl -s -u 'pk-lf-84a76d89-b6b4-4943-b4ea-2dee48d4cda7:sk-lf-24aa6d73-ed36-4447-9eb6-3ebd446fb298' \
  'http://127.0.0.1:3000/api/public/traces?limit=5'

#!/usr/bin/env bash
set -euo pipefail

curl -s -u 'pk-lf-...:sk-lf-...' \
  'http://127.0.0.1:3000/api/public/observations?traceId=<trace_id>'

#!/usr/bin/env bash
set -euo pipefail

docker exec langfuse-clickhouse-1 clickhouse-client \
  --user clickhouse --password <CLICKHOUSE_PASSWORD> \
  --query "SELECT id, name, timestamp FROM traces ORDER BY timestamp DESC LIMIT 5"

docker exec langfuse-clickhouse-1 clickhouse-client \
  --user clickhouse --password <CLICKHOUSE_PASSWORD> \
  --query "SELECT id, type, name, trace_id FROM observations ORDER BY start_time DESC LIMIT 10"

#!/usr/bin/env bash
set -euo pipefail

docker logs --since 10m docker-worker-1 2>&1 | grep ops_trace

Task tasks.ops_trace_task.process_trace_tasks[...] received
new tracing_instance for app_id: 40e95fd3-e7f5-44f2-a03b-dc3e60413540
Processing trace tasks success, app_id: 40e95fd3-e7f5-44f2-a03b-dc3e60413540
Task ... succeeded

#!/usr/bin/env bash
set -euo pipefail

docker logs --since 10m langfuse-langfuse-web-1 2>&1 | grep -E 'Rejected|Error processing|events_only'

-- apps.tracing
{"enabled": true, "tracing_provider": "langfuse"}

-- 查询
SELECT id, name, tracing FROM apps;
SELECT app_id, tracing_provider, is_active FROM trace_app_config;

{
  "tracing_provider": "langfuse",
  "tracing_config": {
    "public_key": "pk-lf-84a76d89-b6b4-4943-b4ea-2dee48d4cda7",
    "secret_key": "sk-lf-24aa6d73-ed36-4447-9eb6-3ebd446fb298",
    "host": "http://192.168.110.207:3000"
  }
}

{
  "enabled": true,
  "tracing_provider": "langfuse"
}

SELECT id, public_key FROM api_keys;
-- 0 rows

LANGFUSE_INIT_PROJECT_PUBLIC_KEY=pk-lf-84a76d89-b6b4-4943-b4ea-2dee48d4cda7
LANGFUSE_INIT_PROJECT_SECRET_KEY=sk-lf-24aa6d73-ed36-4447-9eb6-3ebd446fb298

#!/usr/bin/env bash
set -euo pipefail

cd /Langfuse/langfuse
docker compose -p langfuse up -d langfuse-web

#!/usr/bin/env bash
set -euo pipefail

docker exec langfuse-postgres-1 psql -U postgres -d postgres -c \
  "SELECT id, public_key, note, project_id, display_secret_key FROM api_keys;"

Task tasks.ops_trace_task.process_trace_tasks[...] received
new tracing_instance for app_id: 40e95fd3-e7f5-44f2-a03b-dc3e60413540
Processing trace tasks success, app_id: 40e95fd3-e7f5-44f2-a03b-dc3e60413540
Task ... succeeded in 0.48s

This endpoint is not available on deployments running in Langfuse v4 events_only mode.

Rejected 1 event(s) from the legacy /api/public/ingestion endpoint for project default
because this Langfuse v4 deployment runs in events_only mode. These events were not stored.
Upgrade the client or integration to a v4-compatible SDK or OTLP ingestion path.
As a temporary migration bridge, set LANGFUSE_MIGRATION_V4_WRITE_MODE=dual
on both the web and worker services and redeploy.

LANGFUSE_MIGRATION_V4_WRITE_MODE=dual
LANGFUSE_MIGRATION_V4_ALLOW_PREVIEW_OPT_IN=false

    environment: &langfuse-worker-env
      NEXTAUTH_URL: ${NEXTAUTH_URL:-http://localhost:3000}
      DATABASE_URL: ${DATABASE_URL:-postgresql://postgres:postgres@postgres:5432/postgres} # CHANGEME
      SALT: ${SALT:-mysalt} # CHANGEME
      ENCRYPTION_KEY: ${ENCRYPTION_KEY:-0000000000000000000000000000000000000000000000000000000000000000} # CHANGEME: generate via `openssl rand -hex 32`
      TELEMETRY_ENABLED: ${TELEMETRY_ENABLED:-true}
      LANGFUSE_MIGRATION_V4_WRITE_MODE: ${LANGFUSE_MIGRATION_V4_WRITE_MODE:-dual}
      LANGFUSE_MIGRATION_V4_ALLOW_PREVIEW_OPT_IN: ${LANGFUSE_MIGRATION_V4_ALLOW_PREVIEW_OPT_IN:-false}
      LANGFUSE_ENABLE_EXPERIMENTAL_FEATURES: ${LANGFUSE_ENABLE_EXPERIMENTAL_FEATURES:-false}

    environment:
      <<: *langfuse-worker-env

#!/usr/bin/env bash
set -euo pipefail

cd /Langfuse/langfuse
docker compose -p langfuse up -d langfuse-web langfuse-worker

#!/usr/bin/env bash
set -euo pipefail

docker exec langfuse-langfuse-web-1 printenv LANGFUSE_MIGRATION_V4_WRITE_MODE
docker exec langfuse-langfuse-worker-1 printenv LANGFUSE_MIGRATION_V4_WRITE_MODE
# 都必须输出 dual

#!/usr/bin/env bash
set -euo pipefail

docker exec langfuse-langfuse-web-1 printenv LANGFUSE_MIGRATION_V4_WRITE_MODE

/app/api/.venv/lib/python3.12/site-packages/dify_trace_langfuse/langfuse_trace.py

    def add_span(self, langfuse_span_data: LangfuseSpan | None = None):
        data = filter_none_values(langfuse_span_data.model_dump()) if langfuse_span_data else {}
        try:
            # Langfuse v4 requires observation body.id; auto-assign if missing
            if not data.get("id"):
                data["id"] = str(uuid.uuid4())
            body = CreateSpanBody(
                id=data.get("id"),
                trace_id=data.get("trace_id"),
                name=data.get("name"),
                start_time=data.get("start_time"),
                end_time=data.get("end_time"),
                input=data.get("input"),
                output=data.get("output"),
                metadata=data.get("metadata"),
                level=data.get("level"),
                status_message=data.get("status_message"),
                parent_observation_id=data.get("parent_observation_id"),
                version=data.get("version"),
            )
            event = IngestionEvent_SpanCreate(
                body=body,
                id=self._make_event_id(),
                timestamp=self._now_iso(),
            )
            self.langfuse_client.api.ingestion.batch(batch=[event])
            logger.debug("LangFuse Span created successfully")
        except Exception as e:
            raise ValueError(f"LangFuse Failed to create span: {str(e)}")

    def add_generation(self, langfuse_generation_data: LangfuseGeneration | None = None):
        data = filter_none_values(langfuse_generation_data.model_dump()) if langfuse_generation_data else {}
        try:
            usage_data = data.pop("usage", None)
            usage = None
            if usage_data:
                usage = Usage(
                    input=usage_data.get("input", 0) or 0,
                    output=usage_data.get("output", 0) or 0,
                    total=usage_data.get("total", 0) or 0,
                    unit=usage_data.get("unit"),
                    input_cost=usage_data.get("inputCost"),
                    output_cost=usage_data.get("outputCost"),
                    total_cost=usage_data.get("totalCost"),
                )

            # Langfuse v4 requires observation body.id; auto-assign if missing
            if not data.get("id"):
                data["id"] = str(uuid.uuid4())
            body = CreateGenerationBody(
                id=data.get("id"),
                trace_id=data.get("trace_id"),
                name=data.get("name"),
                start_time=data.get("start_time"),
                end_time=data.get("end_time"),
                model=data.get("model"),
                model_parameters=data.get("model_parameters"),
                input=data.get("input"),
                output=data.get("output"),
                usage=usage,
                metadata=data.get("metadata"),
                level=data.get("level"),
                status_message=data.get("status_message"),
                parent_observation_id=data.get("parent_observation_id"),
                version=data.get("version"),
                completion_start_time=data.get("completion_start_time"),
            )
            event = IngestionEvent_GenerationCreate(
                body=body,
                id=self._make_event_id(),
                timestamp=self._now_iso(),
            )
            self.langfuse_client.api.ingestion.batch(batch=[event])
            logger.debug("LangFuse Generation created successfully")
        except Exception as e:
            raise ValueError(f"LangFuse Failed to create generation: {str(e)}")

from pathlib import Path

p = Path("/app/api/.venv/lib/python3.12/site-packages/dify_trace_langfuse/langfuse_trace.py")
t = p.read_text()
marker = 'Langfuse v4 requires observation body.id; auto-assign if missing'
if marker in t:
    print("already patched")
    raise SystemExit(0)

old_span = """        data = filter_none_values(langfuse_span_data.model_dump()) if langfuse_span_data else {}
        try:
            body = CreateSpanBody(
                id=data.get("id"),
"""
new_span = """        data = filter_none_values(langfuse_span_data.model_dump()) if langfuse_span_data else {}
        try:
            # Langfuse v4 requires observation body.id; auto-assign if missing
            if not data.get("id"):
                data["id"] = str(uuid.uuid4())
            body = CreateSpanBody(
                id=data.get("id"),
"""
old_gen = """            usage_data = data.pop("usage", None)
            usage = None
            if usage_data:
                usage = Usage(
                    input=usage_data.get("input", 0) or 0,
                    output=usage_data.get("output", 0) or 0,
                    total=usage_data.get("total", 0) or 0,
                    unit=usage_data.get("unit"),
                    input_cost=usage_data.get("inputCost"),
                    output_cost=usage_data.get("outputCost"),
                    total_cost=usage_data.get("totalCost"),
                )

            body = CreateGenerationBody(
                id=data.get("id"),
"""
new_gen = """            usage_data = data.pop("usage", None)
            usage = None
            if usage_data:
                usage = Usage(
                    input=usage_data.get("input", 0) or 0,
                    output=usage_data.get("output", 0) or 0,
                    total=usage_data.get("total", 0) or 0,
                    unit=usage_data.get("unit"),
                    input_cost=usage_data.get("inputCost"),
                    output_cost=usage_data.get("outputCost"),
                    total_cost=usage_data.get("totalCost"),
                )

            # Langfuse v4 requires observation body.id; auto-assign if missing
            if not data.get("id"):
                data["id"] = str(uuid.uuid4())
            body = CreateGenerationBody(
                id=data.get("id"),
"""
if old_span not in t:
    raise SystemExit("span block not found")
if old_gen not in t:
    raise SystemExit("gen block not found")
t = t.replace(old_span, new_span, 1).replace(old_gen, new_gen, 1)
p.write_text(t)
print("patched ok")

#!/usr/bin/env bash
set -euo pipefail

# 假设脚本在宿主机 /tmp/patch_lf_id.py
docker cp /tmp/patch_lf_id.py docker-worker-1:/tmp/patch_lf_id.py
docker exec docker-worker-1 python /tmp/patch_lf_id.py
docker cp /tmp/patch_lf_id.py docker-api-1:/tmp/patch_lf_id.py
docker exec docker-api-1 python /tmp/patch_lf_id.py
docker restart docker-worker-1
# 等到日志出现: celery@... ready.

#!/usr/bin/env bash
set -euo pipefail

docker exec docker-worker-1 grep -n 'auto-assign if missing' \
  /app/api/.venv/lib/python3.12/site-packages/dify_trace_langfuse/langfuse_trace.py

#!/usr/bin/env bash
set -euo pipefail

curl -u 'pk:sk' 'http://127.0.0.1:3000/api/public/traces?limit=5'

#!/usr/bin/env bash
set -euo pipefail

curl -s -u 'pk-lf-84a76d89-b6b4-4943-b4ea-2dee48d4cda7:sk-lf-24aa6d73-ed36-4447-9eb6-3ebd446fb298' \
  'http://127.0.0.1:3000/api/public/traces?limit=5'

curl -s -u 'pk-lf-84a76d89-b6b4-4943-b4ea-2dee48d4cda7:sk-lf-24aa6d73-ed36-4447-9eb6-3ebd446fb298' \
  'http://127.0.0.1:3000/api/public/observations?traceId=<trace_id>'

#!/usr/bin/env bash
set -euo pipefail

docker exec langfuse-clickhouse-1 clickhouse-client \
  --user clickhouse --password <CLICKHOUSE_PASSWORD> \
  --query "SELECT id, name, timestamp FROM traces ORDER BY timestamp DESC LIMIT 5"

http://192.168.110.207:3000

#!/usr/bin/env bash
set -euo pipefail

docker exec docker-api-1 curl -s -m 8 http://192.168.110.207:3000/api/public/health
docker exec docker-worker-1 curl -s -m 8 http://192.168.110.207:3000/api/public/health
# 期望: {"status":"OK","version":"4.12.0"}

#!/usr/bin/env bash
set -euo pipefail

# 1. Langfuse 活着
curl -s http://127.0.0.1:3000/api/public/health

# 2. 写入模式
docker exec langfuse-langfuse-web-1 printenv LANGFUSE_MIGRATION_V4_WRITE_MODE
docker exec langfuse-langfuse-worker-1 printenv LANGFUSE_MIGRATION_V4_WRITE_MODE

# 3. Dify 开关
docker exec docker-db_postgres-1 psql -U postgres -d dify -c "SELECT name, tracing FROM apps;"

# 4. 密钥还在
docker exec langfuse-postgres-1 psql -U postgres -d postgres -c "SELECT public_key, project_id FROM api_keys;"

# 5. Dify 容器出网
docker exec docker-worker-1 curl -s -m 8 http://192.168.110.207:3000/api/public/health

LANGFUSE_INIT_ORG_ID=dckj
LANGFUSE_INIT_PROJECT_ID=default
LANGFUSE_INIT_PROJECT_PUBLIC_KEY=pk-lf-84a76d89-b6b4-4943-b4ea-2dee48d4cda7
LANGFUSE_INIT_PROJECT_SECRET_KEY=sk-lf-24aa6d73-ed36-4447-9eb6-3ebd446fb298
LANGFUSE_MIGRATION_V4_WRITE_MODE=dual
LANGFUSE_MIGRATION_V4_ALLOW_PREVIEW_OPT_IN=false

Secret Key = sk-lf-24aa6d73-ed36-4447-9eb6-3ebd446fb298
Public Key = pk-lf-84a76d89-b6b4-4943-b4ea-2dee48d4cda7
Host       = http://192.168.110.207:3000
开关       = 启用，provider = langfuse

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Langfuse 专题 · 步骤 0：确认 Langfuse / Dify 容器与健康检查。"""
from __future__ import annotations

import sys

import _bootstrap  # noqa: F401
import _config as C
from dify_api import Dify


def main() -> int:
    d = Dify(C.DIFY_EMAIL, C.DIFY_PASSWORD)
    try:
        print("=== 容器 ===")
        out, err = d.sh(
            "docker ps --format 'table {{.Names}}\t{{.Status}}' | grep -E 'langfuse|docker-api-1|docker-worker-1'"
        )
        print(out or err)

        print("\n=== Langfuse health ===")
        out, err = d.sh("curl -s %s" % C.LANGFUSE_HEALTH)
        print(out or err)

        for ctr in (C.DIFY_API_CTR, C.DIFY_WORKER):
            print("\n=== %s -> Langfuse ===" % ctr)
            out, err = d.sh(
                "docker exec %s curl -s -m 8 %s" % (ctr, C.LANGFUSE_HEALTH)
            )
            print(out or err)
        return 0
    finally:
        d.close()


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Langfuse 专题 · 步骤 5 · 修改后：给 langfuse_trace.py 补 Observation UUID，重启 worker。"""
from __future__ import annotations

import sys
import time

import _bootstrap  # noqa: F401
import _config as C
import _plugin_util as PU
from dify_api import Dify

LF_TRACE_PATH = (
    "/app/api/.venv/lib/python3.12/site-packages/dify_trace_langfuse/langfuse_trace.py"
)


def patch_container(d: Dify, container: str) -> None:
    content = PU.container_read(d, container, LF_TRACE_PATH)
    if PU.LF_TRACE_MARKER in content:
        print(container, "already patched")
        return
    new_content = PU.apply_langfuse_trace_patch(content)
    PU.container_write(d, container, LF_TRACE_PATH, new_content)
    out, err = d.sh(
        "docker exec %s python -m py_compile %s" % (container, LF_TRACE_PATH)
    )
    print(container, "py_compile:", out or err or "ok")


def main() -> int:
    d = Dify(C.DIFY_EMAIL, C.DIFY_PASSWORD)
    try:
        d.login()
        for ctr in (C.DIFY_WORKER, C.DIFY_API_CTR):
            patch_container(d, ctr)

        print("\n=== restart docker-worker-1 ===")
        out, err = d.sh("docker restart %s" % C.DIFY_WORKER, t=120)
        print(out or err)

        for i in range(15):
            out, _ = d.sh("docker logs --since 30s %s 2>&1 | tail -5" % C.DIFY_WORKER)
            if "ready" in (out or "").lower():
                print("worker ready")
                break
            time.sleep(2)
        else:
            print("[WARN] 未在日志中看到 celery ready，请手动 docker logs -f", C.DIFY_WORKER)

        for ctr in (C.DIFY_WORKER, C.DIFY_API_CTR):
            print(PU.grep_in_container(d, ctr, LF_TRACE_PATH, "auto-assign if missing"))
        return 0
    finally:
        d.close()


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Langfuse 专题 · 步骤 5 · 修改前：确认 langfuse_trace.py 是否已打 Observation ID 补丁。"""
from __future__ import annotations

import sys

import _bootstrap  # noqa: F401
import _config as C
import _plugin_util as PU
from dify_api import Dify

LF_TRACE_PATH = (
    "/app/api/.venv/lib/python3.12/site-packages/dify_trace_langfuse/langfuse_trace.py"
)


def main() -> int:
    d = Dify(C.DIFY_EMAIL, C.DIFY_PASSWORD)
    try:
        for ctr in (C.DIFY_WORKER, C.DIFY_API_CTR):
            print("\n=== %s ===" % ctr)
            out, err = d.sh("docker exec %s ls -l %s" % (ctr, LF_TRACE_PATH))
            print(out or err)
            grep = PU.grep_in_container(d, ctr, LF_TRACE_PATH, "auto-assign if missing")
            if grep.strip():
                print("[OK] 已打补丁:\n", grep)
            else:
                print("[WARN] 未打补丁 — 运行 Langfuse_01_打langfuse_trace补丁_修改后.py")
        return 0
    finally:
        d.close()


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Langfuse 专题 · 步骤 3：POST trace-config + /trace 打开应用 Langfuse Tracing。"""
from __future__ import annotations

import sys

import _bootstrap  # noqa: F401
import _config as C
from dify_api import Dify, pp


def main() -> int:
    d = Dify(C.DIFY_EMAIL, C.DIFY_PASSWORD)
    try:
        ok, _, _ = d.login()
        print("login", ok)
        if not ok:
            return 1

        pp("current trace", d.get("/apps/%s/trace" % C.APP_ID))
        cfg = d.get("/apps/%s/trace-config?tracing_provider=langfuse" % C.APP_ID)
        pp("current config", cfg)

        body = {
            "tracing_provider": "langfuse",
            "tracing_config": {
                "public_key": C.LANGFUSE_PK,
                "secret_key": C.LANGFUSE_SK,
                "host": C.LANGFUSE_URL,
            },
        }
        if (cfg.get("json") or {}).get("has_not_configured"):
            r = d.post("/apps/%s/trace-config" % C.APP_ID, body=body)
            pp("create trace-config", r)
        else:
            r = d.req("PATCH", "/apps/%s/trace-config" % C.APP_ID, body=body)
            pp("update trace-config", r)
            if r.get("code") not in ("200", "201"):
                r = d.post("/apps/%s/trace-config" % C.APP_ID, body=body)
                pp("create after failed patch", r)

        r = d.post(
            "/apps/%s/trace" % C.APP_ID,
            body={"enabled": True, "tracing_provider": "langfuse"},
        )
        pp("enable tracing", r)

        pp("trace after", d.get("/apps/%s/trace" % C.APP_ID))
        pp("config after", d.get("/apps/%s/trace-config?tracing_provider=langfuse" % C.APP_ID))

        out, err = d.sh(
            "docker exec %s psql -U postgres -d dify -c "
            "\"SELECT id, name, tracing FROM apps WHERE id='%s';\""
            % (C.PG_CONTAINER, C.APP_ID)
        )
        print(out or err)
        return 0 if r.get("code") in ("200", "201") else 1
    finally:
        d.close()


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Langfuse 专题 · 步骤 6：发验收对话 + 查 Celery ops_trace + Langfuse traces API。"""
from __future__ import annotations

import sys
import time

import _bootstrap  # noqa: F401
import _config as C
from dify_api import Dify


def main() -> int:
    d = Dify(C.DIFY_EMAIL, C.DIFY_PASSWORD)
    try:
        d.login()
        print("=== 1/3 发验收对话 ===")
        payload = {
            "inputs": {},
            "query": C.PROBE_QUERY_LANGFUSE,
            "response_mode": "streaming",
            "files": [],
            "conversation_id": "",
        }
        r = d.post(
            "/agent/%s/chat-messages" % C.AGENT_ID,
            body=payload,
            extra="-m 420 --max-time 420",
            t=450,
        )
        print("HTTP", r.get("code"))
        print((r.get("text") or "")[-800:])

        print("\n=== 2/3 worker ops_trace 日志 ===")
        time.sleep(5)
        out, err = d.sh(
            "docker logs --since 5m %s 2>&1 | grep ops_trace | tail -10" % C.DIFY_WORKER
        )
        print(out or err or "(无 ops_trace 行)")

        print("\n=== 3/3 Langfuse GET /api/public/traces ===")
        out2, err2 = d.sh(
            "curl -s -u '%s:%s' '%s/api/public/traces?limit=5'"
            % (C.LANGFUSE_PK, C.LANGFUSE_SK, C.LANGFUSE_URL)
        )
        print(out2 or err2)
        return 0 if r.get("code") == "200" else 1
    finally:
        d.close()


if __name__ == "__main__":
    sys.exit(main())