#!/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())