fix(tracing): attach Langfuse trace metadata to the goal evaluator (#4202)

The goal evaluator (runtime/goal.py) runs from runtime/runs/worker.py after
the main graph run has already completed, so there is no graph root for it
to inherit tracing from. create_goal_evaluator_model was built with
attach_tracing=False, and evaluate_goal_completion invoked the model with a
bare config={"run_name": "goal_evaluator"} — no tracing callbacks, no
Langfuse session/user attribution. Every goal-evaluator LLM call went
untraced.

Same class of gap fixed by #2944 for the main agent graph and by #3902 for
memory_agent/suggest_agent: a standalone call site that invokes a model
directly instead of through a traced graph root must attach its own tracing
callbacks and inject Langfuse trace-attribute metadata itself.

- create_goal_evaluator_model: attach_tracing=False -> True, matching the
  other standalone non-graph callers (oneshot_llm.run_oneshot_llm,
  MemoryUpdater).
- evaluate_goal_completion: accept optional thread_id/user_id/
  deerflow_trace_id and inject Langfuse trace metadata onto the ainvoke
  config via the shared inject_langfuse_metadata() helper, mirroring
  oneshot_llm.py's pattern.
- worker.py: thread user_id (resolve_runtime_user_id(runtime)) and
  deerflow_trace_id through _prepare_goal_continuation_input into
  evaluate_goal_completion so the evaluator's trace groups under the
  triggering run's thread/session.

Updates the existing test that pinned attach_tracing=False as expected
behavior, and adds a regression test asserting the ainvoke config carries
Langfuse trace metadata when enabled.
This commit is contained in:
Daoyuan Li 2026-07-15 07:30:34 -07:00 committed by GitHub
parent 6ef0aa2aea
commit 45865e9f3f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 114 additions and 6 deletions

View File

@ -347,7 +347,7 @@ metadata only.
- `POST /wait` (both thread-scoped and `/api/runs/wait`) drains the stream bridge via `wait_for_run_completion()` instead of bare `await record.task`, so it honours the run's `on_disconnect` setting and cancels the background run on real client disconnect rather than returning a stale checkpoint (issue #3265). - `POST /wait` (both thread-scoped and `/api/runs/wait`) drains the stream bridge via `wait_for_run_completion()` instead of bare `await record.task`, so it honours the run's `on_disconnect` setting and cancels the background run on real client disconnect rather than returning a stale checkpoint (issue #3265).
- Redis `StreamBridge` keys use a rolling retained-buffer TTL (`stream_bridge.stream_ttl_seconds`, refreshed on `publish()` / `publish_end()`) as a leak safety net, not as a run timeout. Startup orphan recovery publishes `END_SENTINEL` and schedules stream cleanup for recovered runs; malformed `Last-Event-ID` reconnect values live-tail new Redis events rather than replaying the retained buffer. Do not broaden this into a shared-database multi-pod reaper without adding worker ownership/liveness first. - Redis `StreamBridge` keys use a rolling retained-buffer TTL (`stream_bridge.stream_ttl_seconds`, refreshed on `publish()` / `publish_end()`) as a leak safety net, not as a run timeout. Startup orphan recovery publishes `END_SENTINEL` and schedules stream cleanup for recovered runs; malformed `Last-Event-ID` reconnect values live-tail new Redis events rather than replaying the retained buffer. Do not broaden this into a shared-database multi-pod reaper without adding worker ownership/liveness first.
- Thread-scoped run creation accepts `checkpoint` / `checkpoint_id`; Gateway validates the checkpoint belongs to the request thread before writing `checkpoint_id` / `checkpoint_ns` into `config.configurable` for LangGraph branching. - Thread-scoped run creation accepts `checkpoint` / `checkpoint_id`; Gateway validates the checkpoint belongs to the request thread before writing `checkpoint_id` / `checkpoint_ns` into `config.configurable` for LangGraph branching.
- Thread-scoped Gateway runs evaluate an active `ThreadState.goal` after the visible turn completes. `runtime/goal.py` asks a non-thinking evaluator model to judge only visible conversation evidence and return a typed blocker; the evaluator model is created once per run and reused across hidden continuation checks. Satisfied goals are cleared; every non-satisfied evaluation — continuable or stand-down — is persisted with `last_evaluation` (the blocker, reason, and evidence summary; outcomes that stop the loop additionally record a `stand_down_reason` for observability), but only `goal_not_met_yet` evaluations are streamed as hidden `HumanMessage` continuations, and only when a durable assistant end-of-turn checkpoint exists, the run has not been aborted, the thread did not change during evaluation, and the no-progress breaker has not fired. The continuation cap is 8 — a hard maximum in the `0``8` range; callers requesting more are clamped (`set_goal`/TUI) or rejected with 422 (`PUT /goal`). The no-progress breaker keys on the latest visible assistant evidence (not the evaluator's free-text reason, which an LLM rewords every turn), so two consecutive continuations that add no new visible assistant output stop the loop after 2 attempts. Model-response cleanup helpers such as think-block stripping and code-fence stripping live in `deerflow.utils.llm_text` so `runtime/goal.py` and Gateway suggestion parsing share the same JSON-prep behavior. - Thread-scoped Gateway runs evaluate an active `ThreadState.goal` after the visible turn completes. `runtime/goal.py` asks a non-thinking evaluator model to judge only visible conversation evidence and return a typed blocker; the evaluator model is created once per run and reused across hidden continuation checks. The evaluator runs after the graph root's tracing scope has already closed, so `create_goal_evaluator_model`/`evaluate_goal_completion` attach their own model-level tracing callbacks (`attach_tracing=True`) and inject Langfuse trace metadata (`thread_id`/`user_id`/`deerflow_trace_id`) directly onto the `ainvoke` call — the same standalone-caller pattern as `oneshot_llm.run_oneshot_llm` and `MemoryUpdater` (see Tracing System below). Satisfied goals are cleared; every non-satisfied evaluation — continuable or stand-down — is persisted with `last_evaluation` (the blocker, reason, and evidence summary; outcomes that stop the loop additionally record a `stand_down_reason` for observability), but only `goal_not_met_yet` evaluations are streamed as hidden `HumanMessage` continuations, and only when a durable assistant end-of-turn checkpoint exists, the run has not been aborted, the thread did not change during evaluation, and the no-progress breaker has not fired. The continuation cap is 8 — a hard maximum in the `0``8` range; callers requesting more are clamped (`set_goal`/TUI) or rejected with 422 (`PUT /goal`). The no-progress breaker keys on the latest visible assistant evidence (not the evaluator's free-text reason, which an LLM rewords every turn), so two consecutive continuations that add no new visible assistant output stop the loop after 2 attempts. Model-response cleanup helpers such as think-block stripping and code-fence stripping live in `deerflow.utils.llm_text` so `runtime/goal.py` and Gateway suggestion parsing share the same JSON-prep behavior.
Proxied through nginx: `/api/langgraph/*` → Gateway LangGraph-compatible runtime, all other `/api/*` → Gateway REST APIs. Proxied through nginx: `/api/langgraph/*` → Gateway LangGraph-compatible runtime, all other `/api/*` → Gateway REST APIs.

View File

@ -13,6 +13,7 @@ import hashlib
import inspect import inspect
import json import json
import logging import logging
import os
import threading import threading
import weakref import weakref
from collections.abc import AsyncIterator from collections.abc import AsyncIterator
@ -25,6 +26,7 @@ from langgraph.checkpoint.base import empty_checkpoint, uuid6
import deerflow.utils.llm_text as llm_text import deerflow.utils.llm_text as llm_text
from deerflow.agents.goal_state import GoalBlocker, GoalEvaluation, GoalState from deerflow.agents.goal_state import GoalBlocker, GoalEvaluation, GoalState
from deerflow.models import create_chat_model from deerflow.models import create_chat_model
from deerflow.tracing import inject_langfuse_metadata
from deerflow.utils.messages import message_to_text from deerflow.utils.messages import message_to_text
from deerflow.utils.time import now_iso from deerflow.utils.time import now_iso
@ -242,15 +244,29 @@ def create_goal_evaluator_model(
model_name: str | None = None, model_name: str | None = None,
app_config: Any | None = None, app_config: Any | None = None,
) -> Any: ) -> Any:
"""Create the non-thinking chat model used by the goal evaluator.""" """Create the non-thinking chat model used by the goal evaluator.
The evaluator runs from ``runtime/runs/worker.py`` after the main graph
run has already completed, so unlike ``make_lead_agent``/
``DeerFlowClient.stream``, which attach ``build_tracing_callbacks()`` at
the graph root and correctly pass ``attach_tracing=False`` to avoid
double-attaching there is no graph root here for the evaluator's model
call to inherit tracing from. It must attach its own model-level tracing
callbacks, same as the other standalone, non-graph callers
(``oneshot_llm.run_oneshot_llm``, ``MemoryUpdater``).
"""
return create_chat_model( return create_chat_model(
name=model_name, name=model_name,
thinking_enabled=False, thinking_enabled=False,
app_config=app_config, app_config=app_config,
attach_tracing=False, attach_tracing=True,
) )
def _resolve_environment() -> str | None:
return os.environ.get("DEER_FLOW_ENV") or os.environ.get("ENVIRONMENT")
async def evaluate_goal_completion( async def evaluate_goal_completion(
goal: GoalState, goal: GoalState,
messages: list[Any], messages: list[Any],
@ -258,8 +274,19 @@ async def evaluate_goal_completion(
model: Any | None = None, model: Any | None = None,
model_name: str | None = None, model_name: str | None = None,
app_config: Any | None = None, app_config: Any | None = None,
thread_id: str | None = None,
user_id: str | None = None,
deerflow_trace_id: str | None = None,
) -> GoalEvaluation: ) -> GoalEvaluation:
"""Ask a small non-thinking model whether the active goal is satisfied.""" """Ask a small non-thinking model whether the active goal is satisfied.
``thread_id``/``user_id``/``deerflow_trace_id`` are forwarded to Langfuse
trace metadata only (mirrors ``oneshot_llm.run_oneshot_llm``): this is a
standalone model call outside the main graph, so it must inject its own
Langfuse session/user attribution instead of relying on graph-root
callbacks to lift it same fix as PR #2944 (main graph) and PR #3902
(memory_agent/suggest_agent).
"""
conversation = format_visible_conversation(messages) conversation = format_visible_conversation(messages)
if not conversation or not has_visible_assistant_evidence(messages): if not conversation or not has_visible_assistant_evidence(messages):
return GoalEvaluation( return GoalEvaluation(
@ -283,9 +310,19 @@ async def evaluate_goal_completion(
if model is None: if model is None:
model = create_goal_evaluator_model(model_name=model_name, app_config=app_config) model = create_goal_evaluator_model(model_name=model_name, app_config=app_config)
invoke_config: dict[str, Any] = {"run_name": "goal_evaluator"}
inject_langfuse_metadata(
invoke_config,
thread_id=thread_id,
user_id=user_id,
assistant_id="goal_evaluator",
model_name=model_name,
environment=_resolve_environment(),
deerflow_trace_id=deerflow_trace_id,
)
response = await model.ainvoke( response = await model.ainvoke(
[SystemMessage(content=system_instruction), HumanMessage(content=user_content)], [SystemMessage(content=system_instruction), HumanMessage(content=user_content)],
config={"run_name": "goal_evaluator"}, config=invoke_config,
) )
return parse_goal_evaluation_response(_extract_response_text(response.content)) return parse_goal_evaluation_response(_extract_response_text(response.content))

View File

@ -528,6 +528,8 @@ async def run_agent(
app_config=ctx.app_config, app_config=ctx.app_config,
evaluator_model_factory=_get_goal_evaluator_model, evaluator_model_factory=_get_goal_evaluator_model,
abort_event=record.abort_event, abort_event=record.abort_event,
user_id=resolve_runtime_user_id(runtime),
deerflow_trace_id=deerflow_trace_id,
) )
if continuation_input is None or record.abort_event.is_set(): if continuation_input is None or record.abort_event.is_set():
break break
@ -842,6 +844,8 @@ async def _prepare_goal_continuation_input(
app_config: AppConfig | None, app_config: AppConfig | None,
evaluator_model_factory: Any | None = None, evaluator_model_factory: Any | None = None,
abort_event: asyncio.Event | None = None, abort_event: asyncio.Event | None = None,
user_id: str | None = None,
deerflow_trace_id: str | None = None,
) -> dict[str, Any] | None: ) -> dict[str, Any] | None:
"""Evaluate the active goal and return a hidden continuation input if needed. """Evaluate the active goal and return a hidden continuation input if needed.
@ -919,6 +923,9 @@ async def _prepare_goal_continuation_input(
model=evaluator_model, model=evaluator_model,
model_name=model_name, model_name=model_name,
app_config=app_config, app_config=app_config,
thread_id=thread_id,
user_id=user_id,
deerflow_trace_id=deerflow_trace_id,
) )
if abort_event is not None and abort_event.is_set(): if abort_event is not None and abort_event.is_set():
return None return None

View File

@ -116,11 +116,75 @@ def test_evaluate_goal_completion_uses_non_thinking_model(monkeypatch):
assert result["satisfied"] is True assert result["satisfied"] is True
assert result["blocker"] == "none" assert result["blocker"] == "none"
assert captured["thinking_enabled"] is False assert captured["thinking_enabled"] is False
assert captured["attach_tracing"] is False # The goal evaluator runs from runtime/runs/worker.py after the main graph
# run has already finished, so there is no graph root for it to inherit
# tracing callbacks from (unlike make_lead_agent/DeerFlowClient.stream,
# which attach build_tracing_callbacks() at the graph root and correctly
# pass attach_tracing=False to avoid double-attaching). It must attach its
# own model-level tracing callbacks, same as the other standalone,
# non-graph callers (oneshot_llm.run_oneshot_llm, MemoryUpdater).
assert captured["attach_tracing"] is True
fake_model.ainvoke.assert_awaited_once() fake_model.ainvoke.assert_awaited_once()
# No thread_id/user_id supplied here, and Langfuse is not enabled in the
# ambient test env, so inject_langfuse_metadata() is a no-op and the
# config is unchanged from the plain run_name — see
# test_evaluate_goal_completion_injects_langfuse_metadata below for the
# Langfuse-enabled case.
assert fake_model.ainvoke.await_args.kwargs["config"] == {"run_name": "goal_evaluator"} assert fake_model.ainvoke.await_args.kwargs["config"] == {"run_name": "goal_evaluator"}
def test_evaluate_goal_completion_injects_langfuse_metadata(monkeypatch):
"""Regression test for the goal evaluator's Langfuse tracing gap.
Mirrors PR #2944 (main graph) and PR #3902 (memory_agent/suggest_agent):
a standalone, non-graph model call must inject Langfuse trace-attribute
metadata itself since there is no graph root to lift it from.
"""
from deerflow.config.tracing_config import reset_tracing_config
for name in ("LANGFUSE_TRACING", "LANGFUSE_PUBLIC_KEY", "LANGFUSE_SECRET_KEY", "LANGFUSE_BASE_URL"):
monkeypatch.delenv(name, raising=False)
monkeypatch.setenv("LANGFUSE_TRACING", "true")
monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-lf-test")
monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-lf-test")
reset_tracing_config()
fake_model = MagicMock()
fake_model.ainvoke = AsyncMock(return_value=SimpleNamespace(content='{"satisfied": true, "reason": "Done", "evidence_summary": "Done"}'))
state = goal.build_goal_state("Finish")
try:
result = asyncio.run(
goal.evaluate_goal_completion(
state,
[
HumanMessage(content="Please finish this."),
AIMessage(content="Done."),
],
model=fake_model,
model_name="gpt-4o",
app_config=object(),
thread_id="thread-xyz",
user_id="alice",
deerflow_trace_id="gateway-trace-1",
)
)
finally:
reset_tracing_config()
assert result["satisfied"] is True
fake_model.ainvoke.assert_awaited_once()
config = fake_model.ainvoke.await_args.kwargs["config"]
assert config["run_name"] == "goal_evaluator"
metadata = config.get("metadata") or {}
assert metadata.get("langfuse_session_id") == "thread-xyz", "goal evaluator trace must group under the thread's session"
assert metadata.get("langfuse_user_id") == "alice"
assert metadata.get("langfuse_trace_name") == "goal_evaluator"
assert metadata.get("deerflow_trace_id") == "gateway-trace-1"
tags = metadata.get("langfuse_tags") or []
assert "model:gpt-4o" in tags
def test_evaluate_goal_completion_uses_injected_model(monkeypatch): def test_evaluate_goal_completion_uses_injected_model(monkeypatch):
fake_model = MagicMock() fake_model = MagicMock()
fake_model.ainvoke = AsyncMock(return_value=SimpleNamespace(content='{"satisfied": true, "reason": "Done", "evidence_summary": "Done"}')) fake_model.ainvoke = AsyncMock(return_value=SimpleNamespace(content='{"satisfied": true, "reason": "Done", "evidence_summary": "Done"}'))