镜像站点 · 本页由第三方 GitHub 只读镜像提供,非 GitHub 官方站点,不接受任何登录或凭据输入。前往 github.com
Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
286 changes: 286 additions & 0 deletions .fullsend/scripts/langfuse-post-run.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,286 @@
# Generated by uie-fullsend langfuse-setup. Uses trusted default-branch code.
"""Add generation observations to native Fullsend traces; Python standard library.

Derived from fullsend-component/scripts/lib/langfuse.sh's generation bridge.
Only model/usage and CI metadata are exported; artifacts never supply code,
credentials, an endpoint, or prompt/response content.
"""
import argparse
import hashlib
import json
import math
import os
from pathlib import Path
import re
import secrets
import sys
import time
import urllib.error
import urllib.parse
import urllib.request


def attributes(entries):
if isinstance(entries, dict):
return dict(entries)
result = {}
for entry in entries or []:
value = entry.get("value", {})
for kind in ("stringValue", "intValue", "doubleValue", "boolValue"):
if kind in value:
result[entry["key"]] = value[kind]
break
return result


def otlp_attributes(values):
result = []
for key, value in values.items():
if value is None:
continue
if isinstance(value, bool):
encoded = {"boolValue": value}
elif isinstance(value, int):
encoded = {"intValue": str(value)}
elif isinstance(value, float):
encoded = {"doubleValue": value}
elif isinstance(value, list):
encoded = {"arrayValue": {"values": [{"stringValue": str(item)} for item in value]}}
else:
encoded = {"stringValue": str(value)}
result.append({"key": key, "value": encoded})
return result


def add_filterable_context(values):
for key in ("project_key", "cost_center", "team", "github.repository", "github.run_id"):
if values.get(key):
name = key.replace(".", "_")
for level in ("trace", "observation"):
values[f"langfuse.{level}.metadata.{name}"] = values[key]
tags = [f"{label}:{values[key]}" for key, label in (
("gen_ai.agent.name", "stage"),
("project_key", "project_key"), ("cost_center", "cost_center"),
("team", "team"),
("github.repository", "repo"),
) if values.get(key)]
if tags:
values["langfuse.trace.tags"] = tags


def number(value, integer=False):
if value is None:
return None
parsed = float(value)
if not math.isfinite(parsed) or parsed < 0:
raise ValueError("Invalid usage value")
return int(parsed) if integer else parsed


def read_spans(path):
# The exporter writes one OTLP resourceSpans envelope per JSONL line.
if path.stat().st_size > 32 * 1024 * 1024:
raise ValueError("Telemetry exceeds size limit")
spans = {}
with path.open(encoding="utf-8") as stream:
for line in stream:
if not line.strip():
continue
record = json.loads(line)
for resource in record.get("resourceSpans", []):
for scope in resource.get("scopeSpans", []):
for span in scope.get("spans", []):
spans[(span.get("traceId"), span.get("spanId"))] = span
return list(spans.values())


def build_payload(path, context):
spans = read_spans(path)
agents = [span for span in spans if attributes(span.get("attributes")).get("gen_ai.operation.name") == "invoke_agent"]
# Fullsend emits a root "run" span and a child "agent" span, both tagged invoke_agent.
# Only the innermost belongs to a generation; counting both also suppresses the metrics
# fallback below, costing cost precision (FRFLY-755). Keyed on (traceId, spanId) so a
# child in one trace cannot suppress an agent in another.
parents = {(span.get("traceId"), span.get("parentSpanId")) for span in agents if span.get("parentSpanId")}
model_bearing = [span for span in agents if attributes(span.get("attributes")).get("gen_ai.request.model")]
agents = [span for span in agents if (span.get("traceId"), span.get("spanId")) not in parents]
metrics = {}
metrics_path = path.with_name("metrics.json")
if metrics_path.is_file() and metrics_path.stat().st_size < 1024 * 1024:
metrics = json.loads(metrics_path.read_text(encoding="utf-8"))
if not isinstance(metrics, dict):
metrics = {}
# Use metrics only for a single agent; don't multiply whole-run totals across agents.
# A genuine multi-agent run can collapse to one nested target, so also refuse when more
# than one agent span carries its own model (FRFLY-755).
if len(agents) != 1 or len(model_bearing) > 1:
metrics = {}
output = []
for agent in agents:
source = attributes(agent.get("attributes"))
if any(span.get("parentSpanId") == agent.get("spanId") and
attributes(span.get("attributes")).get("langfuse.observation.type") == "generation" for span in spans):
continue
model = source.get("gen_ai.request.model") or metrics.get("model")
if not model:
continue # Never invent a model (the GitLab reference has a Gemini fallback).
trace_id, parent_id = agent.get("traceId") or "", agent.get("spanId") or ""
if not re.fullmatch(r"[0-9a-fA-F]{32}", trace_id) or not re.fullmatch(r"[0-9a-fA-F]{16}", parent_id):
raise ValueError("Invalid trace identity")
start, end = int(agent["startTimeUnixNano"]), int(agent["endTimeUnixNano"])
if start < 0 or end < start:
raise ValueError("Invalid span timestamps")
model = str(model).removeprefix("google-vertex/")
values = dict(context)
values.update({"langfuse.observation.type": "generation", "gen_ai.operation.name": "generate_content",
"gen_ai.request.model": model, "langfuse.observation.model.name": model,
"gen_ai.system": source.get("gen_ai.system")})
usage = metrics.get("token_usage") or {}
for attr, fallback in (("gen_ai.usage.input_tokens", "input"), ("gen_ai.usage.output_tokens", "output"),
("gen_ai.usage.cache_read.input_tokens", "cache_read"),
("gen_ai.usage.reasoning.output_tokens", "reasoning")):
value = source.get(attr)
if value is None and attr.endswith("reasoning.output_tokens"):
value = source.get("gen_ai.usage.reasoning_tokens")
if value is None:
value = usage.get(fallback)
if value is not None:
values[attr] = number(value, integer=True)
# fullsend.cost_usd on the span is rounded to 2dp; metrics.json beside the telemetry
# carries full precision for the same run, so prefer it (FRFLY-755).
cost = metrics.get("total_cost_usd")
if cost is None:
cost = source.get("fullsend.cost_usd")
if cost is not None:
values["gen_ai.usage.cost"] = number(cost)
values["langfuse.observation.cost_details"] = json.dumps({"total": number(cost)})
for attr, fallback in (("fullsend.num_turns", "iterations"), ("fullsend.tool_calls", None)):
value = source.get(attr)
if value is None and fallback is not None:
value = metrics.get(fallback)
if value is not None:
values[attr] = number(value, integer=True)
stage = source.get("gen_ai.agent.name") or next(
(parent.name.removeprefix("fullsend-") for parent in path.parents
if parent.name.startswith("fullsend-")), "run")
values["gen_ai.agent.name"] = stage
trace_name = f"{stage}: {context.get('github.repository', 'fullsend')}"
if context.get("github.run_id"):
trace_name += f"#{context['github.run_id']}"
values["langfuse.trace.name"] = trace_name
span_id = hashlib.sha256(f"fullsend-langfuse-generation-v1:{trace_id}:{parent_id}".encode()).hexdigest()[:16]
add_filterable_context(values)
output.append({"traceId": trace_id, "spanId": span_id, "parentSpanId": parent_id,
"name": f"{stage.capitalize()} Generation: {model}", "kind": 1,
"startTimeUnixNano": str(start), "endTimeUnixNano": str(end),
"attributes": otlp_attributes(values), "status": {"code": (agent.get("status") or {}).get("code", 1)}})
return {"resourceSpans": [{"resource": {"attributes": otlp_attributes(context)},
"scopeSpans": [{"scope": {"name": "fullsend-generation-bridge", "version": "1.0.0"}, "spans": output}]}]}


def verification_payload(trace_id, context):
if not re.fullmatch(r"[0-9a-fA-F]{32}", trace_id):
raise ValueError("Invalid verification trace ID")
if not context.get("cost_center") or context["cost_center"].lower() == "unassigned":
raise ValueError("Missing configured cost_center resource attribute")
now = time.time_ns()
values = dict(context)
values["langfuse.trace.name"] = "Fullsend Langfuse verification"
add_filterable_context(values)
span = {"traceId": trace_id.lower(), "spanId": secrets.token_hex(8),
"name": "Fullsend Langfuse verification", "kind": 1,
"startTimeUnixNano": str(now), "endTimeUnixNano": str(now + 1_000_000),
"attributes": otlp_attributes(values), "status": {"code": 1}}
return {"resourceSpans": [{"resource": {"attributes": otlp_attributes(context)},
"scopeSpans": [{"scope": {"name": "fullsend-generation-bridge", "version": "1.0.0"},
"spans": [span]}]}]}


def export(payload, env):
endpoint = env.get("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "")
url = urllib.parse.urlsplit(endpoint)
is_loopback_http = url.scheme == "http" and url.hostname in ("localhost", "127.0.0.1", "::1")
if (url.scheme != "https" and not is_loopback_http) or not url.hostname or url.username or url.password or url.query or url.fragment:
raise ValueError("Invalid configured export endpoint")
headers = {"Content-Type": "application/json", "x-langfuse-ingestion-version": "4"}
for item in env.get("OTEL_EXPORTER_OTLP_TRACES_HEADERS", "").split(","):
if "=" in item:
key, value = item.split("=", 1)
headers[key.strip()] = urllib.parse.unquote(value.strip())
if not any(key.lower() == "authorization" for key in headers):
raise ValueError("Missing configured export authorization")
request = urllib.request.Request(endpoint, data=json.dumps(payload).encode(), headers=headers, method="POST")
# Don't forward authentication to an unexpected redirect destination.
class NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, *args, **kwargs):
return None
with urllib.request.build_opener(NoRedirect).open(request, timeout=30) as response:
result = json.loads(response.read() or b"{}")
if int((result.get("partialSuccess") or {}).get("rejectedSpans", 0)):
raise ValueError("Backend rejected generation spans")


def main(argv=None):
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("artifacts", nargs="?", type=Path)
parser.add_argument("--dry-run", action="store_true", help="Build and count spans without contacting Langfuse")
parser.add_argument("--verify-trace-id", help="Export one diagnostic trace with this 32-character hex ID")
args = parser.parse_args(argv)
if args.artifacts is None and args.verify_trace_id is None:
parser.error("artifacts or --verify-trace-id is required")
if args.artifacts is not None and args.verify_trace_id is not None:
parser.error("artifacts and --verify-trace-id cannot be combined")
context = {}
for item in os.environ.get("OTEL_RESOURCE_ATTRIBUTES", "").split(","):
if "=" in item:
key, value = item.split("=", 1)
context[key.strip()] = urllib.parse.unquote(value.strip())
context["github.repository"] = os.environ.get("GITHUB_REPOSITORY", context.get("github.repository", "fullsend"))
for env_key, attr in (("SOURCE_RUN_ID", "github.run_id"), ("SOURCE_RUN_ATTEMPT", "github.run_attempt"),
("SOURCE_ACTOR", "user.id")):
if os.environ.get(env_key):
context[attr] = os.environ[env_key]
if not context.get("team"):
env_team = os.environ.get("FULLSEND_TEAM") or os.environ.get("JIRA_TEAM")
if env_team and env_team.strip().lower() not in ("unassigned", "none", "null", ""):
context["team"] = env_team.strip()
elif context.get("team") and context["team"].strip().lower() in ("unassigned", "none", "null", ""):
context.pop("team", None)
if context.get("github.run_id"):
context["session.id"] = context["langfuse.session.id"] = context["github.run_id"]
if context.get("user.id"):
context["langfuse.user.id"] = context["user.id"]
if args.verify_trace_id is not None:
try:
payload = verification_payload(args.verify_trace_id, context)
if not args.dry_run:
export(payload, os.environ)
print(f"{'Prepared' if args.dry_run else 'Exported'} verification trace {args.verify_trace_id.lower()}.")
return 0
except Exception as error:
status = f" (HTTP {error.code})" if isinstance(error, urllib.error.HTTPError) else ""
print(f"Langfuse verification failed: {type(error).__name__}{status}", file=sys.stderr)
return 1
files = sorted(args.artifacts.rglob("run-telemetry.jsonl"))
if not files:
print("No Fullsend telemetry artifacts found; nothing to export.")
failed = False
for path in files:
try:
payload = build_payload(path, context)
count = len(payload["resourceSpans"][0]["scopeSpans"][0]["spans"])
if count and not args.dry_run:
export(payload, os.environ)
print(f"{'Prepared' if args.dry_run else 'Exported'} {count} generation spans.")
if not count:
print("No agent spans with known models need generation export.")
except Exception as error:
# Never print URLs, headers, artifact content, or upstream error bodies.
status = f" (HTTP {error.code})" if isinstance(error, urllib.error.HTTPError) else ""
print(f"Langfuse generation export failed: {type(error).__name__}{status}", file=sys.stderr)
failed = True
return int(failed)


if __name__ == "__main__":
sys.exit(main())
63 changes: 63 additions & 0 deletions .github/workflows/fullsend-langfuse.yaml
Original file line number Diff line number Diff line change
@@ -0,0 +1,63 @@
# Generated by uie-fullsend langfuse-setup. Uses trusted default-branch code.
name: Fullsend Langfuse
on:
workflow_run:
workflows: [fullsend]
types: [completed]
workflow_dispatch:
inputs:
trace_id:
description: OTLP trace ID for verification (32 hexadecimal characters)
required: true
type: string
permissions:
actions: read
contents: read
concurrency:
group: langfuse-${{ github.event.workflow_run.id || github.run_id }}-${{ github.event.workflow_run.run_attempt || github.run_attempt }}
cancel-in-progress: false
jobs:
generation:
runs-on: ubuntu-24.04
timeout-minutes: 10
steps:
- uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
with:
ref: ${{ github.event.repository.default_branch }}
persist-credentials: false
- name: Download Fullsend output as data
if: github.event_name == 'workflow_run'
id: artifacts
env:
GH_TOKEN: ${{ github.token }}
GH_REPO: ${{ github.repository }}
RUN_ID: ${{ github.event.workflow_run.id }}
run: |
artifact_names="$(gh api --paginate "repos/$GH_REPO/actions/runs/$RUN_ID/artifacts?per_page=100" --jq '.artifacts[].name | select(startswith("fullsend-"))')"
if [ -z "$artifact_names" ]; then
echo "No Fullsend artifacts were published by run $RUN_ID; skipping trace export."
echo "found=false" >> "$GITHUB_OUTPUT"
exit 0
fi
if ! gh run download "$RUN_ID" --pattern 'fullsend-*' --dir "$RUNNER_TEMP/langfuse-artifacts"; then
echo "::error::Failed to download Fullsend artifacts from run $RUN_ID."
exit 1
fi
echo "found=true" >> "$GITHUB_OUTPUT"
- name: Export Langfuse trace
if: github.event_name == 'workflow_dispatch' || steps.artifacts.outputs.found == 'true'
env:
OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: ${{ vars.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT }}
OTEL_EXPORTER_OTLP_TRACES_HEADERS: ${{ secrets.OTEL_EXPORTER_OTLP_TRACES_HEADERS }}
OTEL_RESOURCE_ATTRIBUTES: ${{ vars.OTEL_RESOURCE_ATTRIBUTES }}
SOURCE_RUN_ID: ${{ github.event.workflow_run.id || github.run_id }}
SOURCE_RUN_ATTEMPT: ${{ github.event.workflow_run.run_attempt || github.run_attempt }}
SOURCE_ACTOR: ${{ github.event.workflow_run.actor.login || github.actor }}
VERIFY_TRACE_ID: ${{ inputs.trace_id }}
EVENT_NAME: ${{ github.event_name }}
run: |
if [ "$EVENT_NAME" = workflow_dispatch ]; then
python3 .fullsend/scripts/langfuse-post-run.py --verify-trace-id "$VERIFY_TRACE_ID"
else
python3 .fullsend/scripts/langfuse-post-run.py "$RUNNER_TEMP/langfuse-artifacts"
fi