gitea_admin/pragent
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
feat/review-budget-governor
pragent/pilot/observability/langfuse.py
T

486 lines
17 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python3
"""pragent pilot — Langfuse trace emission.
Ships one trace per PR review to a self-hosted Langfuse (v3) so the reviewer's
token spend, latency and per-model behaviour are queryable outside the review
body. The review body already renders a usage table; that table is per-PR and
disappears into Gitea. This is the same numbers, aggregated.
Why hand-rolled instead of the `langfuse` SDK: the pilot image is stdlib-only
(see pilot/Dockerfile — no requirements.txt anywhere in the repo), and the
ingestion API is a single authenticated POST of a JSON batch. Pulling an SDK
plus its otel dependency tree into a fail-open telemetry side-path is a bad
trade.
Provider split
--------------
`environment` on every trace is either `ollama` or `claude`, derived from the
resolved display model (`resolve_environment`). That is what keeps the two
spend stories separate in Langfuse: every view, filter and cost breakdown
takes an environment selector, so "what did the local/self-hosted path cost"
and "what did the Claude path cost" are two views of one project rather than
two projects with two key pairs to rotate. Tags carry the finer split
(`provider:headroom`, `model:...`, `engine:opencode`).
Cost
----
The pilot's own path bills $0 (headroom proxy, no per-token charge), so the
`cost` reported to Langfuse is the *equivalent* cost from `cost_model` — what
the same tokens would bill on the comparison model. That is the number worth
trending; a chart of $0.00 is not.
A model is "free" when `cost_model.PRICES` has no entry for it (MiniMax-M2.7,
glm-5.2:cloud) or when its entry is all zeros (the self-hosted vLLM qwen). In
both cases the reported cost is priced against the comparison target instead —
same precedence the review body uses: `.pr-review.json:cost_target` >
`PRAGENT_PRICE_TARGET` > `claude-sonnet-5`. A paid model is priced as itself.
Because a hypothetical and a real charge must never be read as the same
number, every trace is tagged `cost:actual` or `cost:equivalent:<target>`, and
the generation's metadata carries `cost_basis`.
Fail-open: every entry point swallows its own exceptions. Telemetry must never
cost a review.
Env:
LANGFUSE_HOST e.g. http://langfuse-web.langfuse.svc.cluster.local:3000
LANGFUSE_PUBLIC_KEY pk-lf-...
LANGFUSE_SECRET_KEY sk-lf-...
LANGFUSE_TIMEOUT seconds, default 5
LANGFUSE_DEBUG 1 to log ingestion failures to stderr
Disabled (silently) when host or either key is unset.
"""
from __future__ import annotations
import base64
import json
import os
import sys
import time
import urllib.error
import urllib.request
import uuid
from datetime import datetime, timezone
INGESTION_PATH = "/api/public/ingestion"
# Model-key prefixes that mean "this review ran against Anthropic-shaped
# billing". Everything else (glm, MiniMax, qwen, local vLLM) is the ollama /
# self-hosted side of the split.
_CLAUDE_PREFIXES = ("claude-", "anthropic/")
def _now_iso() -> str:
return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
def _enabled() -> tuple[str, str, str] | None:
host = (os.environ.get("LANGFUSE_HOST") or "").strip().rstrip("/")
pk = (os.environ.get("LANGFUSE_PUBLIC_KEY") or "").strip()
sk = (os.environ.get("LANGFUSE_SECRET_KEY") or "").strip()
if not host or not pk or not sk:
return None
return host, pk, sk
def _debug(msg: str) -> None:
if os.environ.get("LANGFUSE_DEBUG"):
print(f"pragent/langfuse: {msg}", file=sys.stderr, flush=True)
def strip_provider(model: str) -> str:
"""`headroom/claude-sonnet-5` -> `claude-sonnet-5`. Bare names pass through."""
return model.split("/", 1)[1] if "/" in model else model
def provider_of(model: str) -> str:
"""The opencode provider block a display model routes through."""
return model.split("/", 1)[0] if "/" in model else "headroom"
def resolve_environment(model: str) -> str:
"""Which spend story this review belongs to: `claude` or `ollama`.
Keyed off the bare model name, not the provider, because both paths route
through the same `headroom` proxy — `headroom/claude-sonnet-5` is Claude
spend, `headroom/glm-5.2:cloud` is not.
"""
bare = strip_provider(model).lower()
return "claude" if bare.startswith(_CLAUDE_PREFIXES) else "ollama"
def _usage_details(usage: dict) -> dict:
"""opencode's usage dict -> Langfuse `usageDetails`.
Langfuse sums every key except the ones it knows are derived, so `input`
here is the *uncached* portion: reporting both `input` (which opencode
reports as the full input, cache included) and `cache_read_input_tokens`
would double-count.
"""
inp = int(usage.get("input") or 0)
cache_read = int(usage.get("cache_read") or 0)
cache_write = int(usage.get("cache_write") or 0)
details = {
"input": max(0, inp - cache_read),
"output": int(usage.get("output") or 0),
}
if cache_read:
details["cache_read_input_tokens"] = cache_read
if cache_write:
details["cache_write_input_tokens"] = cache_write
reasoning = int(usage.get("reasoning") or 0)
if reasoning:
details["reasoning"] = reasoning
return details
DEFAULT_PRICE_TARGET = "claude-sonnet-5"
def resolve_price_target(price_target: str | None = None) -> str:
"""The model to price free/unknown runs against.
Mirrors `ai_review._resolve_price_target`: an explicit target (which the
caller reads from `.pr-review.json:cost_target`) wins, then
`PRAGENT_PRICE_TARGET`, then Sonnet.
"""
if price_target and price_target.strip():
return price_target.strip()
env = os.environ.get("PRAGENT_PRICE_TARGET", "").strip()
return env or DEFAULT_PRICE_TARGET
def _is_free(price) -> bool:
"""A price entry that charges nothing — self-hosted or proxied at no cost."""
return price.input == 0 and price.output == 0
def _cost_details(usage: dict, model: str, price_target: str | None = None) -> tuple[dict, str]:
"""USD for this usage plus the basis it was computed on.
Returns `({"total": …}, basis)` where basis is `actual` for a model that
genuinely bills, or `equivalent:<target>` for one that does not. `({}, "")`
when nothing can be priced at all — better no number than a wrong one.
Local import + broad except: `cost_model` is only present on the opencode
path, and an unknown model key must not break telemetry.
"""
try:
from cost_model import PRICES, Usage, cost
bare = strip_provider(model)
price = PRICES.get(bare)
basis = "actual"
if price is None or _is_free(price):
# MiniMax / glm / self-hosted qwen: $0 through the proxy, so the
# useful number is what these tokens would have billed elsewhere.
target = resolve_price_target(price_target)
price = PRICES.get(target)
if price is None:
_debug(f"comparison target {target!r} not in PRICES")
return {}, ""
basis = f"equivalent:{target}"
u = Usage(
uncached_input=max(0, int(usage.get("input") or 0) - int(usage.get("cache_read") or 0)),
cached_input=int(usage.get("cache_read") or 0),
cache_writes=int(usage.get("cache_write") or 0),
output=int(usage.get("output") or 0),
)
return {"total": round(cost(u, price), 6)}, basis
except Exception as e: # pragma: no cover - defensive
_debug(f"cost lookup failed for {model!r}: {e}")
return {}, ""
def _severity_counts(findings: list[dict] | None) -> dict:
counts: dict[str, int] = {}
for f in findings or []:
sev = str(f.get("severity") or "unknown").lower()
counts[sev] = counts.get(sev, 0) + 1
return counts
def build_batch(
*,
repo: str,
index: str,
sha: str,
title: str,
model: str,
usage: dict | None,
findings: list[dict] | None = None,
summary: str = "",
engine: str = "opencode",
tier: str = "",
lenses: list[str] | None = None,
trace_id: str | None = None,
release: str = "",
price_target: str | None = None,
dropped_count: float | None = None,
) -> list[dict]:
"""The ingestion batch for one review: a trace, a generation, and scores.
Split out from `emit_review_trace` so the shape is testable without a
Langfuse to POST to.
`dropped_count` is how many findings the parser rejected for an unusable
`path`/`line`, measured where the model output was parsed. Passing it turns
on the `dropped_findings` score; leaving it `None` omits that score rather
than reporting a zero the caller never measured.
"""
usage = usage or {}
tid = trace_id or str(uuid.uuid4())
ts = _now_iso()
env = resolve_environment(model)
duration = float(usage.get("duration_s") or 0.0)
started = datetime.fromtimestamp(
time.time() - duration, tz=timezone.utc
).isoformat().replace("+00:00", "Z")
tags = [
f"provider:{provider_of(model)}",
f"model:{strip_provider(model)}",
f"engine:{engine}",
f"repo:{repo}",
]
if tier:
tags.append(f"tier:{tier}")
for lens in lenses or []:
tags.append(f"lens:{lens}")
if usage.get("budget_cap_hit"):
tags.extend([
"budget:capped",
f"budget:{usage.get('budget_cap_reason', 'unknown')}",
])
costs, cost_basis = _cost_details(usage, model, price_target) if usage else ({}, "")
if cost_basis:
# Filterable in Langfuse, so an equivalent-cost chart can never be
# mistaken for money actually spent.
tags.append(f"cost:{cost_basis}")
metadata = {
"repo": repo,
"pr": index,
"sha": sha,
"engine": engine,
"steps": usage.get("steps"),
"duration_s": duration or None,
"findings": len(findings or []),
"severities": _severity_counts(findings),
"provider_cost_usd": usage.get("cost"),
"cost_basis": cost_basis or None,
"iterations": usage.get("steps"),
"tool_calls": usage.get("tool_calls"),
"cap_hit": usage.get("budget_cap_hit"),
"cap_reason": usage.get("budget_cap_reason"),
"tokens_per_finding": round(
float(usage.get("total") or 0) / max(1, len(findings or [])), 2
),
"steps_per_finding": round(
float(usage.get("steps") or 0) / max(1, len(findings or [])), 2
),
}
if usage.get("iterations"):
metadata["iteration_usage"] = usage["iterations"][:50]
if lenses:
metadata["lenses"] = lenses
if tier:
metadata["tier"] = tier
metadata = {k: v for k, v in metadata.items() if v not in (None, {}, [])}
trace_body = {
"id": tid,
"name": "pr-review",
"timestamp": ts,
"environment": env,
"sessionId": f"{repo}#{index}",
"input": _review_input(repo, index, sha, title),
"output": _review_output(summary, findings),
"metadata": metadata,
"tags": tags,
}
if release:
trace_body["release"] = release
events = [
{
"id": str(uuid.uuid4()),
"type": "trace-create",
"timestamp": ts,
"body": trace_body,
}
]
if usage:
gen_body = {
"id": str(uuid.uuid4()),
"traceId": tid,
"type": "GENERATION",
"name": f"{engine}-review",
"environment": env,
"startTime": started,
"endTime": ts,
"model": strip_provider(model),
"usageDetails": _usage_details(usage),
"metadata": metadata,
"level": "DEFAULT",
# Repeated from the trace on purpose: an evaluator's variable
# mapping reads the *observation's* input/output, so a generation
# left blank cannot be judged at all.
"input": _review_input(repo, index, sha, title),
"output": _review_output(summary, findings),
}
if costs:
gen_body["costDetails"] = costs
events.append(
{
"id": str(uuid.uuid4()),
"type": "generation-create",
"timestamp": ts,
"body": gen_body,
}
)
events.extend(
_score_events(
trace_id=tid,
findings=findings,
environment=env,
cost_usd=costs.get("total"),
dropped_count=dropped_count,
timestamp=ts,
cost_basis=cost_basis,
)
)
return events
MAX_JUDGED_FINDINGS = 25
_FIELD_CAP = 600
def _review_input(repo: str, index, sha: str, title: str) -> dict:
return {"repo": repo, "pr": index, "sha": sha, "title": title}
def _review_output(summary: str, findings) -> dict:
"""What the reviewer actually said, in a shape an evaluator can read.
The findings themselves are included, not just their count. A judge given
only `{"summary": ..., "findings": 3}` can say nothing about whether those
three findings are specific, actionable, or consistent with the summary —
which is the whole question worth asking of a reviewer that has no ground
truth to check against.
Capped rather than complete: this rides in every ingestion batch, and a
review with 80 findings would push the payload past what is reasonable to
store per trace. `finding_count` stays exact so nothing reading the count
is misled by the cap.
"""
items = list(findings or [])
return {
"summary": summary[:2000],
"finding_count": len(items),
"findings_truncated": len(items) > MAX_JUDGED_FINDINGS,
"findings": [
{
"path": f.get("path"),
"line": f.get("line"),
"severity": f.get("severity"),
"problem": str(f.get("problem") or "")[:_FIELD_CAP],
"fix": str(f.get("fix") or "")[:_FIELD_CAP],
}
for f in items[:MAX_JUDGED_FINDINGS]
],
}
def _score_events(*, cost_basis: str, **kwargs) -> list[dict]:
"""Deterministic scores for this review, or [] if the scorer is missing.
Local import + blanket except for the same reason the rest of this module
swallows: `eval_scores` is optional, and a scoring bug must not cost the
trace it was supposed to annotate.
"""
try:
import eval_scores
# The cost score is only meaningful next to its basis — a $/finding
# figure computed from an equivalent price is not money that was spent.
comment = f"cost basis: {cost_basis}" if cost_basis else ""
return eval_scores.build_scores(comment=comment, **kwargs)
except Exception as e: # pragma: no cover - defensive
_debug(f"scoring failed: {e}")
return []
def _post(host: str, pk: str, sk: str, batch: list[dict], timeout: float) -> int:
payload = json.dumps({"batch": batch}).encode("utf-8")
auth = base64.b64encode(f"{pk}:{sk}".encode("utf-8")).decode("ascii")
req = urllib.request.Request(
host + INGESTION_PATH,
data=payload,
headers={
"Content-Type": "application/json",
"Authorization": f"Basic {auth}",
"User-Agent": "pragent-pilot/1.0",
},
method="POST",
)
with urllib.request.urlopen(req, timeout=timeout) as resp:
_warn_on_rejected_events(resp.read())
return resp.status
def _warn_on_rejected_events(raw: bytes) -> None:
"""Surface per-event rejections hiding inside a 207.
The ingestion endpoint answers 207 Multi-Status when *some* events failed,
so a caller that only checks the status code reads a batch where every
single event was rejected as a success. That failure mode is invisible
exactly when it matters — the traces simply never appear.
"""
try:
body = json.loads(raw or b"{}")
errors = body.get("errors") or []
if errors:
first = errors[0]
_debug(
f"{len(errors)} event(s) rejected by ingestion; "
f"first: status={first.get('status')} {first.get('error')}"
)
except Exception: # pragma: no cover - never let logging break emission
pass
def emit_review_trace(**kwargs) -> bool:
"""Ship one review's trace. Returns True if Langfuse accepted it.
No-op (False) when Langfuse is unconfigured. Never raises — a telemetry
outage must not turn into a failed review.
"""
conf = _enabled()
if conf is None:
return False
host, pk, sk = conf
try:
timeout = float(os.environ.get("LANGFUSE_TIMEOUT", "5"))
except ValueError:
timeout = 5.0
try:
batch = build_batch(**kwargs)
status = _post(host, pk, sk, batch, timeout)
if status not in (200, 201, 207):
_debug(f"ingestion returned HTTP {status}")
return False
return True
except urllib.error.HTTPError as e:
_debug(f"ingestion HTTP {e.code}: {e.read()[:300]!r}")
except Exception as e:
_debug(f"ingestion failed: {e}")
return False
Reference in New Issue View Git Blame Copy Permalink