xiaoxue_admin
/
EverOS
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(examples): zero-install Langfuse replay + a demo memory worth searching (#374) 

* feat(examples): add zero-install Langfuse trace replay

Native OTel moved span emission into the server, so the Langfuse example
lost its try-before-install path: seeing anything now required a
configured EverOS. Restore one without fabricating spans.

replay.py pushes a recording of a real EverOS run into the reader's own
Langfuse project. Names, attributes, token usage, structure and durations
are replayed verbatim; only ids, timestamps and a `replay` tag are
rewritten, so nothing in the trace is invented. It needs the OTel SDK and
Langfuse keys, nothing else.

record_trace.py is the maintainer tool that produced the recording. It
stands in for Langfuse's OTLP and scores endpoints on localhost, which
works because EverOS derives both from langfuse_host, so one sink captures
both signals straight from a real server run.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UyKinsWs1MgoARPoB9R4NW

* feat(examples): give the Langfuse demo a memory worth searching

The demo ingested one conversation and searched it, so recall had nothing to
choose between and the traces showed plumbing rather than behaviour.

Eleven short conversations now span ten weeks, each on its own topic, so a
question has to find the right memory in a populated store. Two revisit the
same subject five days apart, close enough for geometry clustering to group
them, which finally gives reflection something to consolidate: the demo nudges
reflect_episodes (a `0 2 * * 1` cron otherwise), waits for the merge to land,
and the superseded memory is gone from search by the time the questions are
asked. One question asks about something never discussed, so a miss looks like
a miss.

KEYWORD is no longer a demonstrated method. Its top score is raw BM25, on a
different scale from the calibrated ones, so showing the three side by side
invited a comparison that means nothing.

Readiness is polled per session rather than slept through, since a fixed sleep
searched a half-built index and reported scores lower than the memory deserved.
Polling is deliberately slack: every probe is itself a traced search, and a
tight loop buried the real questions under a wall of readiness checks.

recorded_trace.json is that run against 1.2.1: 237 spans over 60 traces, no
errors, no secrets, synthetic content throughout.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UyKinsWs1MgoARPoB9R4NW

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Dani 2026-07-29 19:25:26 -04:00 committed by GitHub
parent 4256419595
commit e723a4eb1c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 7491 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

77
examples/langfuse/README.md
View File

@ -6,7 +6,50 @@ reflection — and exports them over OTLP to any backend, including
[Langfuse](https://langfuse.com). There is **no wrapper and no extra [Langfuse](https://langfuse.com). There is **no wrapper and no extra
instrumentation code**: enable it in config and the traces appear. instrumentation code**: enable it in config and the traces appear.
## Enable Two ways to look at it:
| | What it is | What you need |
| --- | --- | --- |
| [Replay a recording](#replay-a-recording-no-everos-needed) | A trace a real EverOS server produced, pushed into your Langfuse project | Langfuse keys only |
| [Trace your own server](#trace-your-own-server) | Your EverOS, your data, live | An EverOS server |
## Replay a recording (no EverOS needed)
`recorded_trace.json` is a capture of one real `demo.py` run against EverOS
1.2.1: 237 spans over 60 traces. Eleven conversations are ingested and flushed,
each with its LLM extraction and OME strategies nested underneath; reflection
then consolidates two of them and deprecates what they superseded; and five
questions are asked of the resulting memory, with their recall scores.
`replay.py` pushes it into your own Langfuse project, so you can see what the
integration looks like before deploying anything.
```bash
pip install opentelemetry-sdk opentelemetry-exporter-otlp-proto-http
export LANGFUSE_PUBLIC_KEY="pk-lf-..."
export LANGFUSE_SECRET_KEY="sk-lf-..."
export LANGFUSE_HOST="https://cloud.langfuse.com" # US: https://us.cloud.langfuse.com
python replay.py
```
Then open Langfuse → **Tracing** and filter on the `replay` tag.
Span names, attributes, token usage, parent/child structure and durations are
EverOS's own output, replayed verbatim. Three things are rewritten: trace and
span ids are minted fresh so repeated runs do not collide, timestamps are
shifted so the trace lands at the current time, and root spans carry a `replay`
tag so a recording is never mistaken for live traffic.
Two things in the trace list are not self-explanatory. The short keyword
searches beyond the five questions are `demo.py` waiting for each conversation
to become searchable. And the OME spans outlast the `flush` span they hang
under, because reflection continues after the request returns and re-attaches
to the originating trace through its `traceparent`.
Recall scores are per-method scales: read HYBRID against HYBRID, not against
AGENTIC. Agent cases and skills are not in this recording; the span and score
contract is the same when they appear.
## Trace your own server
1. Install the optional OpenTelemetry extra: 1. Install the optional OpenTelemetry extra:
@ -29,15 +72,26 @@ instrumentation code**: enable it in config and the traces appear.
Container/CI equivalent via env vars: `EVEROS_OBSERVABILITY__ENABLED=true`, Container/CI equivalent via env vars: `EVEROS_OBSERVABILITY__ENABLED=true`,
`EVEROS_OBSERVABILITY__LANGFUSE_PUBLIC_KEY=...`, and so on. `EVEROS_OBSERVABILITY__LANGFUSE_PUBLIC_KEY=...`, and so on.
3. Run EverOS normally: 3. Run EverOS normally, then drive one memory lifecycle through it:
```bash ```bash
everos server start everos server start
python demo.py # add -> flush -> search against 127.0.0.1:8000
``` ```
`demo.py` uses only the standard library and contains no instrumentation
code; the spans come from the server. It ingests eleven conversations, nudges
reflection (a weekly cron otherwise), then asks five questions, so the
traces show recall choosing between memories rather than returning the only
one there is.
Off by default — with `enabled = false` (or the `otel` extra absent) there is Off by default — with `enabled = false` (or the `otel` extra absent) there is
zero tracing overhead. zero tracing overhead.
The signal is plain OTLP/HTTP and vendor-neutral, so the same config exports to
an OpenTelemetry Collector or any other OTLP backend. The `langfuse_*` keys are
just a shortcut that fills in the endpoint and auth header for you.
## What you get ## What you get
| EverOS operation | Langfuse observation | | EverOS operation | Langfuse observation |
@ -48,7 +102,8 @@ zero tracing overhead.
| markdown persistence | span `everos.persist.markdown` | | markdown persistence | span `everos.persist.markdown` |
| `POST /api/v2/memory/search` | retriever `everos.memory.search` → `recall` / `rank` | | `POST /api/v2/memory/search` | retriever `everos.memory.search` → `recall` / `rank` |
| query / recall embedding | embedding `everos.embedding` | | query / recall embedding | embedding `everos.embedding` |
| OME reflection strategies | agent `everos.ome.<strategy>` (linked to the triggering request's trace) | | OME extraction strategies | agent `everos.ome.<strategy>` (linked to the triggering request's trace) |
| reflection consolidating a cluster | span `everos.reflect.consolidate` under `everos.ome.reflect_episodes` |
`langfuse.session.id` / `langfuse.user.id` group the traces. Recall quality is `langfuse.session.id` / `langfuse.user.id` group the traces. Recall quality is
pushed as Langfuse scores, split by whether the method's score is calibrated: pushed as Langfuse scores, split by whether the method's score is calibrated:
@ -58,17 +113,21 @@ cosine values are on a different scale and must not be averaged in with the
calibrated ones. Query and memory text are captured only when calibrated ones. Query and memory text are captured only when
`capture_content = true`. `capture_content = true`.
## Try it ## Re-recording the fixture
With a server running and `[observability]` enabled: `record_trace.py` is the maintainer-side tool that produced
`recorded_trace.json`. It stands in for Langfuse's two ingestion endpoints on
localhost, so a real EverOS server exports its spans *and* its recall scores
there instead of to Langfuse. Nothing about the recording is synthesized.
```bash ```bash
python demo.py python record_trace.py # sink on :4318; writes the fixture on Ctrl-C
``` ```
It drives one add → flush → search (keyword / hybrid / agentic) cycle against Point `[observability].langfuse_host` at `http://127.0.0.1:4318`, start the
`http://127.0.0.1:8000` using only the standard library, then tells you to open server, run `demo.py`, then stop the sink. Only worth redoing when the span
Langfuse → **Tracing** filtered to `session.id = langfuse_demo`. contract changes (a span added, renamed, or given new attributes); ordinary
releases do not invalidate a recording.
## Learn more ## Learn more

534
examples/langfuse/demo.py
View File

@ -1,9 +1,15 @@
"""Minimal EverOS x Langfuse demo — native OpenTelemetry tracing. """EverOS x Langfuse demo — native OpenTelemetry tracing.
EverOS emits OTel spans for its own memory operations when ``[observability]`` EverOS emits OTel spans for its own memory operations when ``[observability]``
is enabled; this script contains **no instrumentation code**. It just drives a is enabled; this script contains **no instrumentation code**. It drives a
running server (add -> flush -> search) so the traces the server produces show running server through a memory lifecycle worth looking at in Langfuse.
up in your Langfuse project.
Eleven short conversations spread over ten weeks, each on its own topic, so
recall has to pick the right memory out of a populated store rather than
returning the only thing in it. Two revisit the same subject days apart (an
October trip moves from Lisbon to Porto), close enough that geometry
clustering groups them, which gives reflection something to consolidate. One
query asks about something never discussed, so a miss looks like a miss.
Prereqs (see README.md): Prereqs (see README.md):
1. pip install "everos[otel]" 1. pip install "everos[otel]"
@ -17,11 +23,300 @@ from __future__ import annotations
import json import json
import time import time
import urllib.error
import urllib.request import urllib.request
BASE = "http://127.0.0.1:8000" BASE = "http://127.0.0.1:8000"
SESSION = "langfuse_demo"
USER = "alice" USER = "alice"
DAY_MS = 86_400_000
# Extraction and the SQLite -> LanceDB index sync run asynchronously, so how
# long a memory takes to become searchable depends on the LLM behind it.
INDEX_TIMEOUT_SECONDS = 300.0
# Deliberately slack: every probe is itself a traced search, and polling hard
# would bury the five real questions under a wall of readiness checks.
INDEX_POLL_SECONDS = 10.0
CONSOLIDATION_TIMEOUT_SECONDS = 180.0
# ── the conversations ────────────────────────────────────────────────────
# ``days_ago`` only spaces the timestamps out; every session is ingested now.
SESSIONS: list[dict] = [
{
"id": "everos-demo-trip-booked",
"days_ago": 70,
"messages": [
{
"role": "user",
"content": (
"We booked the October trip: Lisbon, a week, flying out on the "
"12th."
),
},
{
"role": "assistant",
"content": "Noted, a week in Lisbon in October departing on the 12th.",
},
],
},
{
"id": "everos-demo-dentist",
"days_ago": 63,
"messages": [
{
"role": "user",
"content": (
"The dentist put a crown on my lower left molar today. Check-up in "
"six months."
),
},
{
"role": "assistant",
"content": (
"Recorded the crown on your lower left molar, with a check-up due "
"in six months."
),
},
],
},
{
"id": "everos-demo-trip-changed",
"days_ago": 65,
"messages": [
{
"role": "user",
"content": (
"October changed. We cancelled Lisbon and booked Porto instead, my "
"sister moved the wedding there."
),
},
{
"role": "assistant",
"content": (
"Updated, the October trip is Porto now rather than Lisbon, "
"because the wedding moved."
),
},
],
},
{
"id": "everos-demo-cello",
"days_ago": 49,
"messages": [
{
"role": "user",
"content": (
"I started cello lessons. My teacher is Marta, Thursdays at 19:00."
),
},
{
"role": "assistant",
"content": "Cello with Marta on Thursdays at 19:00, noted.",
},
],
},
{
"id": "everos-demo-cat-allergy",
"days_ago": 42,
"messages": [
{
"role": "user",
"content": (
"The vet says Mochi is allergic to chicken. We switched her to the "
"salmon food."
),
},
{
"role": "assistant",
"content": (
"Mochi has a chicken allergy and is now on salmon food. I will "
"keep chicken out of any suggestions for her."
),
},
],
},
{
"id": "everos-demo-bike",
"days_ago": 35,
"messages": [
{
"role": "user",
"content": (
"Radhaus replaced the rear derailleur on my bike. They said to "
"swap the chain at 3000 km."
),
},
{
"role": "assistant",
"content": (
"New rear derailleur from Radhaus, and the chain is due for a swap "
"at 3000 km."
),
},
],
},
{
"id": "everos-demo-sourdough",
"days_ago": 28,
"messages": [
{
"role": "user",
"content": (
"My sourdough came out flat three bakes running. The starter "
"doubles by noon, so that is not it."
),
},
{
"role": "assistant",
"content": (
"The kitchen drops to about 16C overnight and the dough sat on the "
"counter until morning. Bulk proofing wants 24 to 26C, so the rise "
"stalled and the loaf spread instead of holding shape."
),
},
{
"role": "user",
"content": (
"Proofed it in the oven with just the light on and it worked. Best "
"crumb yet."
),
},
{
"role": "assistant",
"content": (
"The oven light holds about 24C, which fixed it. Worth keeping: "
"judge proofing by the dough, not the clock, because the same "
"recipe needs different hours at different temperatures."
),
},
],
},
{
"id": "everos-demo-spare-keys",
"days_ago": 21,
"messages": [
{
"role": "user",
"content": (
"Tomas next door has our spare keys now. He waters the plants when "
"we travel."
),
},
{
"role": "assistant",
"content": (
"Tomas next door holds the spare keys and waters the plants while "
"you are away."
),
},
],
},
{
"id": "everos-demo-physio",
"days_ago": 14,
"messages": [
{
"role": "user",
"content": (
"Started physio for my right shoulder. Band work twice a day, and "
"no overhead presses until they clear me."
),
},
{
"role": "assistant",
"content": (
"Physio for the right shoulder: band exercises twice daily, and "
"overhead presses are off the table until you are cleared."
),
},
],
},
{
"id": "everos-demo-food-rules",
"days_ago": 8,
"messages": [
{
"role": "user",
"content": (
"When you plan meals for me, remember I am vegetarian and I "
"cannot stand mushrooms. No fish either."
),
},
{
"role": "assistant",
"content": (
"Recorded your food rules for meal planning: vegetarian, no "
"fish, and no mushrooms in anything."
),
},
],
},
{
"id": "everos-demo-morning-routine",
"days_ago": 5,
"messages": [
{
"role": "user",
"content": (
"I run before work every morning, so breakfast ends up late, "
"usually around ten."
),
},
{
"role": "assistant",
"content": (
"Noted: you run before work each morning and eat breakfast "
"late, around ten."
),
},
],
},
]
# ── the queries ──────────────────────────────────────────────────────────
# KEYWORD is deliberately absent: its top score is raw BM25, on a different
# scale from the calibrated methods, so showing the three side by side invites
# a comparison that means nothing. It still runs as the readiness probe.
QUERIES: list[dict] = [
{
"label": "history",
"query": (
"what happened with the October trip we booked, did the destination "
"change after the wedding moved"
),
"note": "the plan, its revision, and whatever reflection made of them",
},
{
"label": "constraint",
"query": (
"what did the vet say about Mochi's allergy and which food did we "
"switch her to"
),
"note": "one specific memory out of eleven conversations",
},
{
"label": "how-to",
"query": (
"why did my sourdough loaves keep coming out flat and what fixed "
"the overnight proofing"
),
"note": "the diagnosis and the fix, not just a stated fact",
},
{
"label": "profile",
"query": (
"which foods should you leave out when you plan my meals, I am vegetarian"
),
"note": "include_profile also returns the distilled profile",
"include_profile": True,
},
{
"label": "miss",
"query": "what did the accountant say about our tax return this year",
"note": "never discussed — candidates come back, the score says no",
},
]
METHODS = ("hybrid", "agentic")
def _post(path: str, body: dict) -> dict: def _post(path: str, body: dict) -> dict:
@ -35,57 +330,204 @@ def _post(path: str, body: dict) -> dict:
return json.load(resp) return json.load(resp)
def main() -> None: def _search(spec: dict, method: str, *, top_k: int = 5) -> dict:
ts = int(time.time() * 1000) """Search one query across everything its owner remembers.
add = _post( No session filter: the point is to make recall choose between memories
"/api/v2/memory/add", from different conversations.
"""
body: dict = {
"user_id": USER,
"query": spec["query"],
"method": method,
"top_k": top_k,
}
if spec.get("include_profile"):
body["include_profile"] = True
return _post("/api/v2/memory/search", body)
def _wire_messages(session: dict, base_ts: int) -> list[dict]:
"""Expand a session's (role, content) pairs into API message items."""
return [
{ {
"session_id": SESSION, "message_id": f"{session['id']}-m{index}",
"messages": [ "role": message["role"],
{ "content": message["content"],
"message_id": "m1", "timestamp": base_ts + index * 60_000,
"role": "user", "sender_id": USER if message["role"] == "user" else "assistant",
"content": "Moved our vector store to LanceDB to fix index bloat.", }
"timestamp": ts, for index, message in enumerate(session["messages"], start=1)
"sender_id": USER, ]
},
{
"message_id": "m2", def _ingest(session: dict, now_ms: int) -> None:
"role": "assistant", base_ts = now_ms - session["days_ago"] * DAY_MS
"content": "Noted — LanceDB with compaction keeps it compact.", messages = _wire_messages(session, base_ts)
"timestamp": ts + 1000, add = _post(
"sender_id": "assistant", "/api/v2/memory/add", {"session_id": session["id"], "messages": messages}
}, )
], flush = _post("/api/v2/memory/flush", {"session_id": session["id"], "messages": []})
}, print(
f" {session['id']:<30} {len(messages):>2} msgs "
f"add={add['data'].get('status')} flush={flush['data'].get('status')}"
) )
print("add ->", add["data"])
flush = _post("/api/v2/memory/flush", {"session_id": SESSION, "messages": []})
print("flush ->", flush["data"])
print("waiting for async index sync ...") def _session_is_searchable(session: dict) -> bool:
time.sleep(10) """Keyword-probe one session with its own opening line.
for method in ("keyword", "hybrid", "agentic"): Querying the session's own words guarantees the lexical overlap BM25
resp = _post( needs, so an empty result means "not indexed yet" rather than "no match".
"""
opening = next(m["content"] for m in session["messages"] if m["role"] == "user")
body = {
"user_id": USER,
"query": opening,
"method": "keyword",
"top_k": 1,
"filters": {"session_id": session["id"]},
}
return bool(_post("/api/v2/memory/search", body)["data"].get("episodes"))
def _wait_for_index() -> list[str]:
"""Poll until every ingested session is searchable; return any laggards.
Waiting on one session is not enough: extraction and the SQLite ->
LanceDB sync run per session and finish out of order, so querying too
early makes recall choose from a partial store and the scores read
lower than the memory deserves.
"""
deadline = time.monotonic() + INDEX_TIMEOUT_SECONDS
laggards: list[str] = []
# One session at a time, in ingest order. Probing every pending session on
# every round would work too, but each probe is itself a traced search, and
# a hundred readiness probes would bury the five real queries in Langfuse.
# Extraction broadly follows ingest order, so by the time session N answers
# its predecessors already have.
for session in SESSIONS:
while not _session_is_searchable(session):
if time.monotonic() > deadline:
laggards.append(session["id"])
break
time.sleep(INDEX_POLL_SECONDS)
return laggards
def _reflect() -> str:
"""Run episode reflection now instead of waiting for its weekly cron.
Consolidation is what merges a cluster of related memories and deprecates
what they superseded, so a demo that never triggers it never shows the
part of EverOS that improves memory over time. ``reflect_episodes`` is
scheduled ``0 2 * * 1``, hence the manual nudge.
"""
body = {"name": "reflect_episodes", "force": True, "timeout": 300.0}
return str(_post("/api/v2/ome/trigger", body)["status"])
def _wait_for_consolidation() -> bool:
"""Poll until the consolidated memory has replaced what it superseded.
``/ome/trigger`` returns once the OME engine is idle, but the merge reaches
LanceDB through the cascade, and deprecating the old episodes is a separate
write from indexing the merged one. Querying in between sees neither, and
scores lower than the memory deserves. So wait for both edges: the first
trip session going unsearchable, then the merged memory answering for it.
"""
superseded, survivor = SESSIONS[0], SESSIONS[2]
opening = next(m["content"] for m in survivor["messages"] if m["role"] == "user")
deadline = time.monotonic() + CONSOLIDATION_TIMEOUT_SECONDS
while time.monotonic() < deadline:
if not _session_is_searchable(superseded):
break
time.sleep(INDEX_POLL_SECONDS)
else:
return False
# The originals are gone; wait for the merged episode to answer in their
# place. No session filter: the merge is its own entry, not either source.
while time.monotonic() < deadline:
found = _post(
"/api/v2/memory/search", "/api/v2/memory/search",
{ {"user_id": USER, "query": opening, "method": "keyword", "top_k": 1},
"user_id": USER, )["data"].get("episodes")
"query": "which vector database did we move to and why", if found:
"method": method, return True
"top_k": 5, time.sleep(INDEX_POLL_SECONDS)
"filters": {"session_id": SESSION}, return False
},
def _describe(data: dict) -> str:
"""What a search returned per memory kind, plus its best score.
The score matters more than the count: recall returns candidates up to
``top_k`` whether or not they are relevant, so a query about something
never discussed still comes back with episodes. The top score is what
says they do not answer it.
"""
parts = [
f"{len(items)} {kind.replace('_', ' ')}"
for kind in ("episodes", "profiles", "agent_cases", "agent_skills")
if (items := data.get(kind) or [])
]
scored = [
item.get("score")
for kind in ("episodes", "agent_cases", "agent_skills")
for item in data.get(kind) or []
if item.get("score") is not None
]
summary = ", ".join(parts) or "nothing"
return f"{summary:<34} top_score={max(scored):.3f}" if scored else summary
def main() -> None:
now_ms = int(time.time() * 1000)
print(f"ingesting {len(SESSIONS)} sessions ...")
for session in SESSIONS:
_ingest(session, now_ms)
print("\nwaiting for async extraction + index sync ...")
if pending := _wait_for_index():
print(
f" still not searchable after {INDEX_TIMEOUT_SECONDS:.0f}s: "
f"{', '.join(pending)}; searching anyway so you can still see "
"the traces"
) )
hits = len(resp["data"].get("episodes", [])) else:
print(f"search[{method}] -> {hits} hit(s)") print(f" all {len(SESSIONS)} sessions searchable")
print("\nrunning reflection (normally a weekly cron) ...")
print(f" reflect_episodes -> {_reflect()}")
if _wait_for_consolidation():
print(f" {SESSIONS[0]['id']} superseded and no longer searchable")
else:
print(" nothing was consolidated; the originals are both still live")
print()
for spec in QUERIES:
print(f"{spec['label']}: {spec['query']}")
print(f" ({spec['note']})")
for method in spec.get("methods", METHODS):
try:
data = _search(spec, method)["data"]
except urllib.error.HTTPError as exc:
# Embedding and rerank are soft dependencies: with neither
# configured a server serves KEYWORD only, and HYBRID /
# AGENTIC answer 422 CAPABILITY_UNAVAILABLE. Report it and
# carry on so the other queries still have something to show.
print(f" {method:<8} HTTP {exc.code}: {exc.reason}")
continue
print(f" {method:<8} {_describe(data)}")
print()
print( print(
f"\nOpen Langfuse -> Tracing and filter session.id = {SESSION} " "Open Langfuse -> Tracing. Traces are grouped by session; the search "
"to see the traces (add / flush / search, with token usage and " "traces carry recall-quality scores, and the flush traces carry the "
"recall-quality scores)." "LLM token usage Langfuse turns into cost."
) )

218
examples/langfuse/record_trace.py Normal file
View File

@ -0,0 +1,218 @@
"""Record a real EverOS trace into ``recorded_trace.json`` (maintainer tool).
This stands in for Langfuse's two ingestion endpoints on localhost, so a real
EverOS server exports its spans *and* its recall scores here instead of to
Langfuse. What lands in the fixture is exactly what EverOS emitted: no span is
synthesized, no attribute is invented. ``replay.py`` then pushes that recording
into any reader's own Langfuse project.
Both signals are captured by one sink because EverOS derives both endpoints
from ``langfuse_host``: spans go to ``<host>/api/public/otel/v1/traces`` and
scores to ``<host>/api/public/scores``.
Usage:
1. Point EverOS at this sink in ``everos.toml``. Keep the LLM, embedding and
rerank sections filled in — a recording with real generations is the
point, since that is what gives Langfuse the token usage to cost out.
[observability]
enabled = true
langfuse_public_key = "pk-lf-local" # any value; the sink ignores auth
langfuse_secret_key = "sk-lf-local"
langfuse_host = "http://127.0.0.1:4318"
capture_content = true # demo data is synthetic, so show it
2. ``python record_trace.py`` # starts the sink on :4318
3. ``everos server start`` # in another shell
4. ``python demo.py`` # drives add -> flush -> search
5. Ctrl-C the sink; it writes ``recorded_trace.json``
Requires the OTel protobuf definitions, which ship with the exporter EverOS
already needs::
pip install opentelemetry-exporter-otlp-proto-http
"""
from __future__ import annotations
import argparse
import gzip
import json
import sys
from datetime import UTC, datetime
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Any
from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import (
ExportTraceServiceRequest,
ExportTraceServiceResponse,
)
from opentelemetry.proto.trace.v1.trace_pb2 import Span as PbSpan
from opentelemetry.proto.trace.v1.trace_pb2 import Status as PbStatus
TRACES_PATH = "/api/public/otel/v1/traces"
SCORES_PATH = "/api/public/scores"
# Collected across requests; written out on shutdown.
_spans: list[dict[str, Any]] = []
_scores: list[dict[str, Any]] = []
_resource: dict[str, Any] = {}
def _any_value(value: Any) -> Any:
"""Decode an OTLP ``AnyValue`` into a plain Python value."""
which = value.WhichOneof("value")
if which == "array_value":
return [_any_value(item) for item in value.array_value.values]
if which == "kvlist_value":
return {kv.key: _any_value(kv.value) for kv in value.kvlist_value.values}
if which is None:
return None
return getattr(value, which)
def _attributes(pairs: Any) -> dict[str, Any]:
return {kv.key: _any_value(kv.value) for kv in pairs}
def _ingest_traces(body: bytes) -> int:
"""Decode one OTLP export request, appending its spans to ``_spans``."""
global _resource
request = ExportTraceServiceRequest()
request.ParseFromString(body)
count = 0
for resource_spans in request.resource_spans:
if not _resource:
_resource = _attributes(resource_spans.resource.attributes)
for scope_spans in resource_spans.scope_spans:
for span in scope_spans.spans:
parent = span.parent_span_id.hex()
_spans.append(
{
"trace_id": span.trace_id.hex(),
"span_id": span.span_id.hex(),
"parent_span_id": parent or None,
"name": span.name,
"kind": PbSpan.SpanKind.Name(span.kind),
"start_unix_nano": span.start_time_unix_nano,
"end_unix_nano": span.end_time_unix_nano,
"status": {
"code": PbStatus.StatusCode.Name(span.status.code),
"message": span.status.message,
},
"attributes": _attributes(span.attributes),
}
)
count += 1
return count
class _Handler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
def do_POST(self) -> None:
length = int(self.headers.get("content-length") or 0)
body = self.rfile.read(length)
if self.headers.get("content-encoding") == "gzip":
body = gzip.decompress(body)
path = self.path.split("?", 1)[0]
if path == TRACES_PATH:
try:
added = _ingest_traces(body)
except Exception as exc: # keep the sink alive; the export retries
print(f" ! failed to decode an export: {exc}", file=sys.stderr)
self._respond(400, b"")
return
print(f" spans +{added} (total {len(_spans)})")
self._respond(
200,
ExportTraceServiceResponse().SerializeToString(),
content_type="application/x-protobuf",
)
return
if path == SCORES_PATH:
score = json.loads(body)
_scores.append(score)
print(
f" score {score.get('name')}={score.get('value')} "
f"({score.get('comment')})"
)
self._respond(201, b"{}", content_type="application/json")
return
self._respond(404, b"")
def _respond(
self, status: int, body: bytes, *, content_type: str | None = None
) -> None:
self.send_response(status)
if content_type:
self.send_header("content-type", content_type)
self.send_header("content-length", str(len(body)))
self.end_headers()
if body:
self.wfile.write(body)
def log_message(self, *args: Any) -> None:
"""Silence the default per-request logging; we print our own summary."""
def _write_fixture(path: str, everos_version: str | None) -> None:
if not _spans:
print("\nNothing recorded — no fixture written.", file=sys.stderr)
return
_spans.sort(key=lambda span: span["start_unix_nano"])
fixture = {
"recorded_at": datetime.now(UTC).isoformat(timespec="seconds"),
"everos_version": everos_version or _resource.get("service.version"),
"resource": _resource,
"spans": _spans,
"scores": _scores,
}
with open(path, "w", encoding="utf-8") as handle:
json.dump(fixture, handle, indent=2, ensure_ascii=False)
handle.write("\n")
traces = len({span["trace_id"] for span in _spans})
print(
f"\nWrote {path}: {len(_spans)} span(s) across {traces} trace(s), "
f"{len(_scores)} score(s)."
)
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--port", type=int, default=4318)
parser.add_argument("--out", default="recorded_trace.json")
parser.add_argument(
"--everos-version",
default=None,
help="Stamped into the fixture; defaults to the exporter's "
"service.version resource attribute.",
)
args = parser.parse_args()
server = ThreadingHTTPServer(("127.0.0.1", args.port), _Handler)
print(
f"Recording on http://127.0.0.1:{args.port}\n"
f" spans <- POST {TRACES_PATH}\n"
f" scores <- POST {SCORES_PATH}\n"
"Point everos.toml's [observability].langfuse_host at it, start the "
"server, run demo.py, then Ctrl-C here.\n"
)
# Let KeyboardInterrupt break out of serve_forever, then write the fixture on
# the way out. Calling server.shutdown() from a signal handler instead would
# deadlock: it waits for the serve_forever loop that the handler is blocking.
try:
server.serve_forever()
except KeyboardInterrupt:
print("\nstopping ...")
finally:
server.server_close()
_write_fixture(args.out, args.everos_version)
if __name__ == "__main__":
main()

6451
examples/langfuse/recorded_trace.json Normal file
View File

File diff suppressed because it is too large Load Diff

266
examples/langfuse/replay.py Normal file
View File

@ -0,0 +1,266 @@
"""Replay a recorded EverOS trace into your own Langfuse project.
No EverOS install and no model API keys: this pushes a trace that a real
EverOS server actually produced (``recorded_trace.json``, captured with
``record_trace.py``) into your Langfuse project, so you can see what the
integration looks like in your own UI before deciding to deploy anything.
It is a recording, not a live server. Span names, attributes, token usage,
parent/child structure and durations are EverOS's own output, replayed
verbatim. Three things are necessarily rewritten: trace/span ids are minted
fresh (so repeated runs do not collide), timestamps are shifted so the trace
lands at the current time, and the root spans get a ``replay`` tag so nobody
mistakes it for live traffic.
Usage::
pip install opentelemetry-sdk opentelemetry-exporter-otlp-proto-http
export LANGFUSE_PUBLIC_KEY="pk-lf-..."
export LANGFUSE_SECRET_KEY="sk-lf-..."
export LANGFUSE_HOST="https://cloud.langfuse.com" # US: https://us.cloud.langfuse.com
python replay.py
To trace your own EverOS server instead, see ``README.md`` — that needs no
replay at all, just ``[observability]`` in ``everos.toml``.
"""
from __future__ import annotations
import argparse
import base64
import json
import os
import sys
import time
import urllib.error
import urllib.request
from collections import defaultdict
from typing import Any
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.trace import (
Span,
SpanKind,
Status,
StatusCode,
set_span_in_context,
)
DEFAULT_HOST = "https://cloud.langfuse.com"
REPLAY_TAG = "replay"
SCORE_MAX_ATTEMPTS = 5
# Small gap between scores; cheaper than discovering the limiter one 429 at a time.
SCORE_PACE_SECONDS = 0.15
def _credentials() -> tuple[str, str]:
"""Langfuse OTLP endpoint + Basic auth header, from the standard env vars."""
public_key = os.environ.get("LANGFUSE_PUBLIC_KEY")
secret_key = os.environ.get("LANGFUSE_SECRET_KEY")
host = os.environ.get("LANGFUSE_HOST", DEFAULT_HOST).rstrip("/")
if not (public_key and secret_key):
sys.exit(
"LANGFUSE_PUBLIC_KEY and LANGFUSE_SECRET_KEY must be set "
"(project settings in Langfuse)."
)
token = base64.b64encode(f"{public_key}:{secret_key}".encode()).decode()
return host, f"Basic {token}"
def _check_credentials(host: str, auth: str) -> None:
"""Fail fast, and say why, before pushing a few hundred spans.
Langfuse keys are region-scoped, and the OTLP exporter only reports a
rejected export through the SDK's own logging, so a wrong host otherwise
looks like a successful run into an empty project.
"""
request = urllib.request.Request(
f"{host}/api/public/projects", headers={"Authorization": auth}
)
try:
with urllib.request.urlopen(request, timeout=20) as response:
response.read()
except urllib.error.HTTPError as exc:
if exc.code not in {401, 403}:
# Only auth is under test; any other response is the replay's problem.
return
other = (
"https://cloud.langfuse.com"
if "us." in host
else "https://us.cloud.langfuse.com"
)
sys.exit(
f"{host} rejected these keys ({exc.code}). Langfuse projects are "
f"region-scoped, so if the project lives in the other region set "
f"LANGFUSE_HOST={other} and try again."
)
except OSError:
return # unreachable host surfaces on the real export a moment later
def _span_kind(name: str) -> SpanKind:
bare = name.removeprefix("SPAN_KIND_")
if bare in {"", "UNSPECIFIED"}:
return SpanKind.INTERNAL
return SpanKind[bare]
def _status(record: dict[str, Any]) -> Status | None:
code = record.get("code", "STATUS_CODE_UNSET").removeprefix("STATUS_CODE_")
if code in {"", "UNSET"}:
return None
return Status(StatusCode[code], record.get("message") or None)
def _post_score(host: str, auth: str, payload: dict[str, Any]) -> None:
"""POST one score, backing off when Langfuse rate-limits the endpoint.
Scores go one per request, so replaying a whole recording sends dozens in a
row and reliably trips the limiter without this.
"""
request = urllib.request.Request(
f"{host}/api/public/scores",
data=json.dumps(payload).encode(),
headers={"content-type": "application/json", "Authorization": auth},
method="POST",
)
for attempt in range(SCORE_MAX_ATTEMPTS):
try:
with urllib.request.urlopen(request, timeout=20) as response:
response.read()
return
except urllib.error.HTTPError as exc:
retryable = exc.code == 429 or 500 <= exc.code < 600
if not retryable or attempt == SCORE_MAX_ATTEMPTS - 1:
raise
after = exc.headers.get("retry-after") if exc.headers else None
delay = float(after) if after and after.isdigit() else 2.0**attempt
time.sleep(delay)
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--fixture", default="recorded_trace.json")
args = parser.parse_args()
host, auth = _credentials()
_check_credentials(host, auth)
try:
with open(args.fixture, encoding="utf-8") as handle:
fixture = json.load(handle)
except FileNotFoundError:
sys.exit(
f"{args.fixture} not found. Fetch it next to this script from "
"https://github.com/EverMind-AI/EverOS/tree/main/examples/langfuse"
)
spans: list[dict[str, Any]] = fixture["spans"]
if not spans:
sys.exit(f"{args.fixture} contains no spans.")
provider = TracerProvider(resource=Resource.create(fixture.get("resource") or {}))
provider.add_span_processor(
BatchSpanProcessor(
OTLPSpanExporter(
endpoint=f"{host}/api/public/otel/v1/traces",
headers={"Authorization": auth},
)
)
)
tracer = provider.get_tracer("everos.replay")
# Land the recording at "now", preserving every relative duration.
offset = time.time_ns() - min(span["start_unix_nano"] for span in spans)
by_id = {span["span_id"]: span for span in spans}
children: dict[str, list[dict[str, Any]]] = defaultdict(list)
roots: list[dict[str, Any]] = []
for span in spans:
parent = span["parent_span_id"]
if parent and parent in by_id:
children[parent].append(span)
else:
roots.append(span)
# old span id -> (new trace id hex, new span id hex), for remapping scores.
remapped: dict[str, tuple[str, str]] = {}
trace_remap: dict[str, str] = {}
def emit(record: dict[str, Any], parent: Span | None) -> None:
attributes = dict(record["attributes"])
if parent is None:
tags = attributes.get("langfuse.trace.tags")
tags = list(tags) if isinstance(tags, list) else []
if REPLAY_TAG not in tags:
tags.append(REPLAY_TAG)
attributes["langfuse.trace.tags"] = tags
recorded_at = fixture.get("recorded_at")
if recorded_at:
attributes["langfuse.trace.metadata.replay_of"] = recorded_at
span = tracer.start_span(
record["name"],
context=set_span_in_context(parent) if parent is not None else None,
kind=_span_kind(record["kind"]),
start_time=record["start_unix_nano"] + offset,
attributes=attributes,
)
context = span.get_span_context()
remapped[record["span_id"]] = (
format(context.trace_id, "032x"),
format(context.span_id, "016x"),
)
trace_remap.setdefault(record["trace_id"], format(context.trace_id, "032x"))
for child in children[record["span_id"]]:
emit(child, span)
status = _status(record["status"])
if status is not None:
span.set_status(status)
span.end(end_time=record["end_unix_nano"] + offset)
for root in roots:
emit(root, None)
provider.force_flush()
provider.shutdown()
print(f"Replayed {len(spans)} span(s) in {len(roots)} trace(s) to {host}")
sent = 0
skipped = 0
for score in fixture.get("scores", []):
payload = dict(score)
observation = score.get("observationId")
if observation and observation in remapped:
trace_id, span_id = remapped[observation]
payload["traceId"] = trace_id
payload["observationId"] = span_id
elif score.get("traceId") in trace_remap:
payload["traceId"] = trace_remap[score["traceId"]]
payload.pop("observationId", None)
else:
skipped += 1
continue
try:
_post_score(host, auth, payload)
sent += 1
except urllib.error.HTTPError as exc:
print(f" ! score {score.get('name')} rejected: {exc}", file=sys.stderr)
time.sleep(SCORE_PACE_SECONDS)
if sent or skipped:
note = f", {skipped} unmapped" if skipped else ""
print(f"Pushed {sent} recall score(s){note}")
print(
"\nOpen Langfuse -> Tracing and filter on the 'replay' tag. "
"This is a recorded EverOS run, not a live server: "
"see README.md to trace your own."
)
if __name__ == "__main__":
main()