ssh root@192.168.110.207
rm -f /tmp/dify_cj.txt
python3 - <<'PY'
import json, base64
open("/tmp/dify_login.json", "w").write(json.dumps({
    "email": "38205036@qq.com",
    "password": base64.b64encode(b"Difyai123456").decode(),
    "language": "zh-Hans",
    "remember_me": True,
}))
PY
curl -s -c /tmp/dify_cj.txt -X POST 'http://127.0.0.1/console/api/login' \
  -H 'Content-Type: application/json' \
  -d @/tmp/dify_login.json
CSRF=$(python3 - <<'PY'
jar=open("/tmp/dify_cj.txt").read()
for line in jar.splitlines():
    if "csrf_token" in line:
        print(line.split("\t")[-1].strip())
PY
)
echo "CSRF=$CSRF"

# 207 服务器
/usr/bin/python3 /opt/workbuddy/scripts_cursor/作业3_追踪Agent工具调用验收/作业3_验收Langfuse工具调用.py

curl -s -b /tmp/dify_cj.txt -H "X-CSRF-Token: $CSRF" \
  'http://127.0.0.1/console/api/apps/d00d0382-ca3f-4cf2-9cc9-0cb5aed9404d/trace-config?tracing_provider=langfuse'

curl -s -b /tmp/dify_cj.txt -c /tmp/dify_cj.txt \
  -H "X-CSRF-Token: $CSRF" -H 'Content-Type: application/json' \
  -X POST 'http://127.0.0.1/console/api/apps/d00d0382-ca3f-4cf2-9cc9-0cb5aed9404d/trace-config' \
  -d '{
    "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"
    }
  }'

curl -s -b /tmp/dify_cj.txt -c /tmp/dify_cj.txt \
  -H "X-CSRF-Token: $CSRF" -H 'Content-Type: application/json' \
  -X POST 'http://127.0.0.1/console/api/apps/d00d0382-ca3f-4cf2-9cc9-0cb5aed9404d/trace' \
  -d '{"enabled": true, "tracing_provider": "langfuse"}'

curl -s -b /tmp/dify_cj.txt -H "X-CSRF-Token: $CSRF" \
  'http://127.0.0.1/console/api/apps/d00d0382-ca3f-4cf2-9cc9-0cb5aed9404d/trace'

/usr/bin/python3 /opt/workbuddy/scripts_cursor/作业3_追踪Agent工具调用验收/Langfuse_01_打langfuse_trace补丁_修改后.py

ssh root@192.168.110.207 'docker restart docker-worker-1'
ssh root@192.168.110.207 'docker logs --since 60s docker-worker-1 | tail -20'

/usr/bin/python3 /opt/workbuddy/scripts_cursor/作业3_追踪Agent工具调用验收/作业3_验收Langfuse工具调用.py

curl -s -b /tmp/dify_cj.txt -c /tmp/dify_cj.txt \
  -H "X-CSRF-Token: $CSRF" -H 'Content-Type: application/json' \
  --max-time 420 \
  -X POST 'http://127.0.0.1/console/api/agent/01a0189c-7d7e-742f-83d1-74c5c16878da/chat-messages' \
  -d '{
    "inputs": {},
    "query": "帮我看看 gpu-server-01 的状态",
    "response_mode": "streaming",
    "files": [],
    "conversation_id": ""
  }'

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

TRACE_ID='这里换成上一步的 message_id'
curl -s -u 'pk-lf-84a76d89-b6b4-4943-b4ea-2dee48d4cda7:sk-lf-24aa6d73-ed36-4447-9eb6-3ebd446fb298' \
  "http://192.168.110.207:3000/api/public/observations?traceId=${TRACE_ID}"

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""作业 3 · 验收：Langfuse 中能否看到 Agent 的每次工具调用详情。

验收目标（自动检查版）：
1) Trace 存在（找到输入包含 query 的那条）
2) observations 里至少有一条工具 Observation（name=工具 operationId）
3) 工具 inputs / outputs 不为空
4) 至少能看到一步耗时（start/end 或 duration 字段之一）
5) generation 上能看到 token 统计（totalTokens / usage_details 等）
"""

from __future__ import annotations

import json
import sys
import time
import re
from typing import Any

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


def enable_trace_for_app(d: Dify, app_id: str) -> None:
    # 1) trace-config（langfuse keys + 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={
                "tracing_provider": "langfuse",
                "tracing_config": {
                    "public_key": C.LANGFUSE_PK,
                    "secret_key": C.LANGFUSE_SK,
                    "host": C.LANGFUSE_URL,
                },
            },
        )
        if r.get("code") not in ("200", "201"):
            raise RuntimeError("create trace-config failed: %r" % r)
    else:
        r = d.req(
            "PATCH",
            "/apps/%s/trace-config" % app_id,
            body={
                "tracing_provider": "langfuse",
                "tracing_config": {
                    "public_key": C.LANGFUSE_PK,
                    "secret_key": C.LANGFUSE_SK,
                    "host": C.LANGFUSE_URL,
                },
            },
        )
        if r.get("code") not in ("200", "201"):
            # 有些版本如果 PATCH 不适配，就退回 POST
            r2 = d.post(
                "/apps/%s/trace-config" % app_id,
                body={
                    "tracing_provider": "langfuse",
                    "tracing_config": {
                        "public_key": C.LANGFUSE_PK,
                        "secret_key": C.LANGFUSE_SK,
                        "host": C.LANGFUSE_URL,
                    },
                },
            )
            if r2.get("code") not in ("200", "201"):
                raise RuntimeError("patch trace-config failed: %r / %r" % (r, r2))

    # 2) enable trace
    r2 = d.post(
        "/apps/%s/trace" % app_id,
        body={"enabled": True, "tracing_provider": "langfuse"},
    )
    if r2.get("code") not in ("200", "201"):
        print("[WARN] enable trace returned code:", r2.get("code"))

    # 3) verify (方便判断是否真正启用了 ops_trace / trace_instance)
    print("=== verify trace enabled ===")
    print(d.get("/apps/%s/trace" % app_id))
    print("=== verify trace-config ===")
    print(d.get("/apps/%s/trace-config?tracing_provider=langfuse" % app_id))


def lf_public_get(d: Dify, path: str) -> Any:
    # 使用 Langfuse web 可达的地址（本机 curl）
    url = path if path.startswith("http") else C.LANGFUSE_URL.rstrip("/") + path
    out, err = d.sh(
        "curl -s -m 20 -u '%s:%s' '%s'"
        % (C.LANGFUSE_PK, C.LANGFUSE_SK, url)
    )
    if err and not out:
        raise RuntimeError("curl failed: %s" % err)
    return json.loads(out)


def find_trace_by_query(d: Dify, query: str) -> str | None:
    for _ in range(12):
        data = lf_public_get(d, "/api/public/traces?limit=5")
        items = data.get("data") or []
        for it in items:
            inp = it.get("input")
            inp_s = json.dumps(inp, ensure_ascii=False)
            if query in inp_s:
                return it.get("id")
        time.sleep(3)
    return None


def main() -> int:
    # homework1 的 HTTP 工具 Agent
    app_id = C.HOMEWORK1_APP_ID
    agent_id = C.HOMEWORK1_AGENT_ID
    query = C.HOMEWORK1_PROBE_QUERY
    tool_name = "getServerStatus"  # operationId

    d = Dify(C.DIFY_EMAIL, C.DIFY_PASSWORD)
    try:
        d.login()
        enable_trace_for_app(d, app_id)

        print("=== 1/3 发 Agent 对话（触发工具调用） ===")
        payload = {
            "inputs": {},
            "query": query,
            "response_mode": "streaming",
            "files": [],
            "conversation_id": "",
        }
        r: dict[str, Any] | None = None
        code = None
        text = ""
        for attempt in range(1, 4):
            print("\n--- agent chat attempt %d/3 ---" % attempt)
            r = d.post(
                "/agent/%s/chat-messages" % agent_id,
                body=payload,
                extra="-m 420 --max-time 420",
                t=450,
            )
            code = r.get("code") if isinstance(r, dict) else None
            text = (r.get("text") or "") if isinstance(r, dict) else ""
            print("chat http:", code)
            print("chat tail:", text[-2000:])
            if code in ("200", "201"):
                break
            if attempt < 3:
                print("[WARN] Agent chat failed; retry after 6s.")
                time.sleep(6)
        if code not in ("200", "201"):
            print("[FAIL] Agent chat failed after retries; code=%r" % code)
            return 1

        # 优先从流式返回里提取 message_id，当作 traceId，避免匹配到旧 trace
        tid = ""
        m = re.search(r'"message_id"\s*:\s*"([0-9a-fA-F-]{36})"', text)
        if m:
            tid = m.group(1)
        else:
            m2 = re.search(r'"id"\s*:\s*"([0-9a-fA-F-]{36})"', text)
            if m2:
                tid = m2.group(1)

        print("\n=== 2/3 在 Langfuse 查 trace（优先流式提取 traceId） ===")
        if not tid:
            tid = find_trace_by_query(d, query) or ""
        if not tid:
            print("[FAIL] 没找到对应 trace：query=%r" % query)
            return 1
        print("traceId:", tid)

        print("\n=== 3/3 拉 observations 并检查工具调用 ===")
        items = []
        for i in range(10):
            time.sleep(3)
            obs = lf_public_get(d, "/api/public/observations?traceId=%s" % tid)
            items = obs.get("data") or []
            if items:
                break
        print("observations count:", len(items))

        tool_items = []
        for it in items:
            if it.get("name") == tool_name or tool_name in (it.get("name") or ""):
                tool_items.append(it)

        # 关键检查：工具调用 observation 是否出现
        if not tool_items:
            print("[FAIL] observations 里找不到工具：%s" % tool_name)
            # 打印前几个观察项，方便你手工对照
            for it in items[:10]:
                print("obs:", it.get("type"), it.get("name"), it.get("id"))
            return 1

        print("tool observation(s):", len(tool_items))
        sample = tool_items[0]
        print("tool type:", sample.get("type"), "name:", sample.get("name"))
        print("tool start/end:", sample.get("startTime"), sample.get("endTime"))

        # 工具输入/输出字段在不同版本字段名可能不同，这里做宽松匹配
        input_s = sample.get("input")
        output_s = sample.get("output")
        input_head = json.dumps(input_s, ensure_ascii=False)[:400] if input_s is not None else ""
        output_head = json.dumps(output_s, ensure_ascii=False)[:400] if output_s is not None else ""
        print("tool input head:", input_head)
        print("tool output head:", output_head)

        if not (input_s or output_s):
            print("[FAIL] tool input/output 为空（需要检查 tracing patch / ingestion 丢弃规则）")
            return 1

        # 生成 token：通常在 type=GENERATION / name=llm 的 observation 或 usage 字段里
        gen = [it for it in items if (it.get("type") or "").upper() == "GENERATION" and (it.get("name") or "") == "llm"]
        if not gen:
            # 宽松一点：找任意 llm generation
            gen = [it for it in items if (it.get("name") or "") == "llm" or (it.get("type") or "").upper() == "GENERATION"]
        if not gen:
            print("[WARN] 没找到 llm generation observation；token 可能只在 trace.input 里。")
            return 0

        usage = gen[0].get("usage") or gen[0].get("usage_details") or {}
        # 兼容几个可能的字段名
        total_tokens = usage.get("totalTokens") or usage.get("total_tokens") or usage.get("total") or gen[0].get("totalTokens")
        print("llm usage totalTokens:", total_tokens)

        print("\n[PASS] Langfuse 已显示工具调用 observation：%s" % tool_name)
        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)
    already_id = PU.LF_TRACE_MARKER in content
    already_tool_spans_v2 = PU.LF_AGENT_TOOL_SPANS_MARKER_V2 in content
    has_parent_link = "parent_observation_id=langfuse_generation_data.id" in content
    has_gen_id_assignment = "langfuse_generation_data.id = message_id" in content
    if (
        already_id
        and already_tool_spans_v2
        and has_parent_link
        and has_gen_id_assignment
    ):
        print(container, "already patched (id + tool spans v2 with parent link)")
        return
    new_content = PU.apply_langfuse_trace_patch(content)
    new_content = PU.apply_langfuse_agent_tool_spans_patch(new_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())

def container_read(d, container: str, path: str) -> str:
    out, err = d.sh("docker exec %s cat %s" % (container, path))
    if not out and err:
        raise RuntimeError(err)
    # normalize CRLF -> LF so patch string comparisons work uniformly
    return out.replace("\r\n", "\n")


LF_TRACE_MARKER = "Langfuse v4 requires observation body.id; auto-assign if missing"
LF_AGENT_TOOL_SPANS_MARKER = "# >>> AGENT TOOL SPANS PATCH"
LF_AGENT_TOOL_SPANS_MARKER_V2 = "# >>> AGENT TOOL SPANS PATCH v2"
LF_AGENT_TOOL_SPANS_MARKER_V2 = "# >>> AGENT TOOL SPANS PATCH v2"

def apply_langfuse_trace_patch(content: str) -> str:
    if LF_TRACE_MARKER in content:
        return content
    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 content:
        raise RuntimeError("langfuse_trace.py span 锚点未找到")
    if old_gen not in content:
        raise RuntimeError("langfuse_trace.py generation 锚点未找到")
    content = content.replace(old_span, new_span, 1).replace(old_gen, new_gen, 1)
    return content

def apply_langfuse_agent_tool_spans_patch(content: str) -> str:
    """在 langfuse_trace.py 的 message_trace() 内补写 mode=agent 的工具调用 spans。

    版本 v2：改用与 ops_trace_manager.tool_trace() 相同的字段名
    （agent_thought.tools / .tool_meta / .observation / .created_at），
    以及 base_agent_runner 保存的 tool_invoke_meta 等字段。
    """

    if LF_AGENT_TOOL_SPANS_MARKER_V2 in content:
        # already has v2 block, but might be missing:
        # 1) stable generation id assignment: langfuse_generation_data.id = message_id
        # 2) parent linking on tool span: parent_observation_id=langfuse_generation_data.id
        has_parent = "parent_observation_id=langfuse_generation_data.id" in content
        has_gen_id = "langfuse_generation_data.id = message_id" in content
        if has_parent and has_gen_id:
            return content

        # targeted fix #1: inject generation id assignment before add_generation
        old_start = (
            "        self.add_generation(langfuse_generation_data)\n\n        "
            + LF_AGENT_TOOL_SPANS_MARKER_V2
            + "\n"
        )
        new_start = (
            "        try:\n"
            "            langfuse_generation_data.id = message_id\n"
            "        except Exception:\n"
            "            pass\n\n"
            "        self.add_generation(langfuse_generation_data)\n\n        "
            + LF_AGENT_TOOL_SPANS_MARKER_V2
            + "\n"
        )
        if old_start in content:
            content = content.replace(old_start, new_start, 1)

        # targeted fix #2: add parent_observation_id to tool spans
        old_trace_line = "                        trace_id=trace_id,\n                        start_time=start_t,\n"
        new_trace_line = (
            "                        trace_id=trace_id,\n"
            "                        parent_observation_id=langfuse_generation_data.id,\n"
            "                        start_time=start_t,\n"
        )
        if old_trace_line in content:
            content = content.replace(old_trace_line, new_trace_line, 1)

        return content

    # 判断是否有旧版补丁，若有则把旧块整体换成 v2
    if LF_AGENT_TOOL_SPANS_MARKER in content:
        old_block_start = "\n        " + LF_AGENT_TOOL_SPANS_MARKER + "\n"
        old_block_end_marker = "agent tool spans patch failed"
        if old_block_start not in content or old_block_end_marker not in content:
            raise RuntimeError("找到旧版 PATCH marker 但无法定位完整旧块边界")
        idx_start = content.index(old_block_start)
        idx_mod = content.index("\n    def moderation_trace", idx_start)
        content = content[:idx_start] + content[idx_mod:]
        old_anchor = (
            "        self.add_generation(langfuse_generation_data)\n\n"
            "    def moderation_trace"
        )
        if old_anchor not in content:
            raise RuntimeError("清除旧块后锚点仍未找到，请手动检查容器文件")
    else:
        old_anchor = (
            "        self.add_generation(langfuse_generation_data)\n\n"
            "    def moderation_trace"
        )
        if old_anchor not in content:
            raise RuntimeError("langfuse_trace.py message_trace 锚点未找到")

    # 完整的补丁插入块（以换行形式展开，避免 f-string 混乱）
    patch_code = (
        "        " + LF_AGENT_TOOL_SPANS_MARKER_V2 + "\n"
        "        # ensure generation id stable so tool spans can link as children\n"
        "        try:\n"
        "            langfuse_generation_data.id = message_id\n"
        "        except Exception:\n"
        "            pass\n"
        "\n"
        "        self.add_generation(langfuse_generation_data)\n"
        "\n"
        "        # mode=agent: derive tool spans from message_data.agent_thoughts\n"
        "        # Uses same fields as ops_trace_manager.tool_trace():\n"
        "        # agent_thought.tools / .tool_meta / .observation / .created_at\n"
        "        try:\n"
        "            agent_thoughts_list = getattr(message_data, 'agent_thoughts', None) or []\n"
        "            for at in agent_thoughts_list:\n"
        "                raw_tools = getattr(at, 'tools', None)\n"
        "                if not raw_tools:\n"
        "                    continue\n"
        "                if isinstance(raw_tools, str):\n"
        "                    tool_name_list = [t.strip() for t in raw_tools.split(',') if t.strip()]\n"
        "                elif isinstance(raw_tools, list):\n"
        "                    tool_name_list = [str(t) for t in raw_tools if t]\n"
        "                else:\n"
        "                    tool_name_list = []\n"
        "\n"
        "                raw_obs = getattr(at, 'observation', None) or ''\n"
        "                obs_map = {}\n"
        "                if isinstance(raw_obs, dict):\n"
        "                    obs_map = raw_obs\n"
        "                elif isinstance(raw_obs, str) and raw_obs:\n"
        "                    if tool_name_list:\n"
        "                        obs_map = {tool_name_list[0]: raw_obs}\n"
        "\n"
        "                raw_tool_input = getattr(at, 'tool_input', None) or ''\n"
        "                input_map = {}\n"
        "                if isinstance(raw_tool_input, dict):\n"
        "                    input_map = raw_tool_input\n"
        "                elif isinstance(raw_tool_input, str) and raw_tool_input:\n"
        "                    try:\n"
        "                        parsed = json.loads(raw_tool_input)\n"
        "                        if isinstance(parsed, dict):\n"
        "                            input_map = parsed\n"
        "                        else:\n"
        "                            input_map = {tool_name_list[0]: raw_tool_input} if tool_name_list else {}\n"
        "                    except Exception:\n"
        "                        input_map = {tool_name_list[0]: raw_tool_input} if tool_name_list else {}\n"
        "\n"
        "                tool_meta_raw = getattr(at, 'tool_meta', None) or {}\n"
        "                at_start = getattr(at, 'created_at', None) or trace_info.start_time\n"
        "\n"
        "                for tn in tool_name_list:\n"
        "                    tool_output = obs_map.get(tn, obs_map.get(None) if len(tool_name_list) == 1 else None)\n"
        "                    if tool_output is None and len(obs_map) == 1:\n"
        "                        tool_output = next(iter(obs_map.values()), None)\n"
        "                    tool_inp = input_map.get(tn, input_map.get(None))\n"
        "                    if tool_inp is None and len(input_map) == 1:\n"
        "                        tool_inp = next(iter(input_map.values()), None)\n"
        "\n"
        "                    tmeta = tool_meta_raw.get(tn, {}) if isinstance(tool_meta_raw, dict) else {}\n"
        "                    time_cost_s = tmeta.get('time_cost', 0) or 0\n"
        "                    try:\n"
        "                        time_cost_f = float(time_cost_s)\n"
        "                    except (TypeError, ValueError):\n"
        "                        time_cost_f = 0.0\n"
        "\n"
        "                    start_t = at_start\n"
        "                    end_t = start_t + timedelta(seconds=time_cost_f) if start_t else None\n"
        "                    err = tmeta.get('error')\n"
        "\n"
        "                    tool_span = LangfuseSpan(\n"
        "                        name=tn,\n"
        "                        input=tool_inp,\n"
        "                        output=tool_output,\n"
        "                        trace_id=trace_id,\n"
        "                        parent_observation_id=langfuse_generation_data.id,\n"
        "                        start_time=start_t,\n"
        "                        end_time=end_t,\n"
        "                        metadata={**metadata, 'tool_name': tn, 'time_cost': time_cost_f},\n"
        "                        level=(LevelEnum.DEFAULT if not err else LevelEnum.ERROR),\n"
        "                        status_message=err,\n"
        "                    )\n"
        "                    self.add_span(langfuse_span_data=tool_span)\n"
        "        except Exception:\n"
        "            logger.exception('agent tool spans patch v2 failed')\n"
        "\n"
        "    def moderation_trace"
    )

    return content.replace(old_anchor, patch_code, 1)