Skip to content

Add LLMObs.submitFeedback for end-user feedback submission #12129

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ddog-thibault-nadin wants to merge 5 commits into
base: master
Choose a base branch
from
+2,054 −52
Open

Add LLMObs.submitFeedback for end-user feedback submission #12129

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
fcd1138
Add LLMObs.submitFeedback for end-user feedback submission
ddog-thibault-nadin Aug 3, 2026
8f65747
Align submitFeedback with the dd-trace-py implementation
ddog-thibault-nadin Aug 4, 2026
2407617
Align the dotted label error message with dd-trace-js
ddog-thibault-nadin Aug 5, 2026
4b96f54
Keep the intake worker alive when a batch cannot be serialized
ddog-thibault-nadin Aug 5, 2026
894040e
Reject non-finite feedback scores and snapshot json values
ddog-thibault-nadin Aug 5, 2026
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
125 changes: 75 additions & 50 deletions ...og/trace/llmobs/EvalProcessingWorker.java → ...adog/trace/llmobs/LLMObsIntakeWorker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,19 +1,16 @@
package datadog.trace.llmobs;

import static datadog.trace.util.AgentThreadFactory.AgentThread.LLMOBS_EVALS_PROCESSOR;
import static datadog.trace.util.AgentThreadFactory.THREAD_JOIN_TIMOUT_MS;
import static datadog.trace.util.AgentThreadFactory.newAgentThread;

import com.squareup.moshi.JsonAdapter;
import com.squareup.moshi.Moshi;
import datadog.common.queue.MessagePassingBlockingQueue;
import datadog.common.queue.Queues;
import datadog.communication.ddagent.DDAgentFeaturesDiscovery;
import datadog.communication.ddagent.SharedCommunicationObjects;
import datadog.communication.http.HttpRetryPolicy;
import datadog.communication.http.OkHttpUtils;
import datadog.trace.api.Config;
import datadog.trace.llmobs.domain.LLMObsEval;
import datadog.trace.util.AgentThreadFactory.AgentThread;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
Expand All @@ -25,64 +22,74 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class EvalProcessingWorker implements AutoCloseable {
/**
* Batches LLM Observability payloads on a dedicated thread and posts them to an intake endpoint.
*
* <p>Evaluations and feedback share this machinery but not an endpoint: evaluations are submitted
* to {@code v1/eval-metric} and feedback to {@code v2/eval-metric}, so each gets its own instance
* with its own queue and its own batch.
*
* @param <T> the payload type this worker submits
*/
public class LLMObsIntakeWorker<T> implements AutoCloseable {

private static final String EVAL_METRIC_API_DOMAIN = "api";
private static final String EVAL_METRIC_API_PATH = "api/intake/llm-obs/v1/eval-metric";
private static final String INTAKE_API_DOMAIN = "api";

private static final String EVP_SUBDOMAIN_HEADER_NAME = "X-Datadog-EVP-Subdomain";
private static final String DD_API_KEY_HEADER_NAME = "DD-API-KEY";

private static final Logger log = LoggerFactory.getLogger(EvalProcessingWorker.class);
private static final Logger log = LoggerFactory.getLogger(LLMObsIntakeWorker.class);

private final MessagePassingBlockingQueue<LLMObsEval> queue;
/** Serializes a whole batch into the request body sent to the intake. */
public interface BatchSerializer<T> {
String toJson(List<T> batch);
}

private final MessagePassingBlockingQueue<T> queue;
private final Thread serializerThread;

public EvalProcessingWorker(
public LLMObsIntakeWorker(
final String payloadDescription,
final String apiPath,
final AgentThread agentThread,
final int capacity,
final long flushInterval,
final TimeUnit timeUnit,
final SharedCommunicationObjects sco,
Config config) {
final Config config,
final BatchSerializer<T> serializer) {
this.queue = Queues.mpscBlockingConsumerArrayQueue(capacity);

boolean isAgentless = config.isLlmObsAgentlessEnabled();
if (isAgentless && (config.getApiKey() == null || config.getApiKey().isEmpty())) {
log.error("Agentless eval metric submission requires an API key");
log.error("Agentless {} submission requires an API key", payloadDescription);
}

Headers headers;
HttpUrl submissionUrl;
if (isAgentless) {
submissionUrl =
HttpUrl.get(
"https://"
+ EVAL_METRIC_API_DOMAIN
+ "."
+ config.getSite()
+ "/"
+ EVAL_METRIC_API_PATH);
HttpUrl.get("https://" + INTAKE_API_DOMAIN + "." + config.getSite() + "/" + apiPath);
headers = Headers.of(DD_API_KEY_HEADER_NAME, config.getApiKey());
} else {
submissionUrl =
HttpUrl.get(
sco.agentUrl.toString()
+ DDAgentFeaturesDiscovery.V2_EVP_PROXY_ENDPOINT
+ EVAL_METRIC_API_PATH);
headers = Headers.of(EVP_SUBDOMAIN_HEADER_NAME, EVAL_METRIC_API_DOMAIN);
sco.agentUrl.toString() + DDAgentFeaturesDiscovery.V2_EVP_PROXY_ENDPOINT + apiPath);
Comment on lines 75 to +77

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 4, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Use the shared agent client for feedback proxy submissions

In non-agentless configurations that use a Unix domain socket or named pipe for the Agent, sco.agentUrl is only a placeholder URL while the real socket transport is configured on sco.agentHttpClient. This worker still builds the feedback proxy URL from that placeholder and sends it with its own raw OkHttpClient, so submitFeedback cannot reach the Agent in those supported setups; use the shared agent client (and the shared intake client for agentless) when posting these batches.

Useful? React with 👍 / 👎.

@ddog-thibault-nadin ddog-thibault-nadin Aug 5, 2026 •
edited
Loading

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think was already there, proposed a fix

@ddog-thibault-nadin ddog-thibault-nadin Aug 6, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

see #12154

headers = Headers.of(EVP_SUBDOMAIN_HEADER_NAME, INTAKE_API_DOMAIN);
}

EvalSerializingHandler serializingHandler =
new EvalSerializingHandler(queue, flushInterval, timeUnit, submissionUrl, headers);
this.serializerThread = newAgentThread(LLMOBS_EVALS_PROCESSOR, serializingHandler);
SerializingHandler<T> serializingHandler =
new SerializingHandler<>(
payloadDescription, queue, flushInterval, timeUnit, submissionUrl, headers, serializer);
this.serializerThread = newAgentThread(agentThread, serializingHandler);
}

public void start() {
this.serializerThread.start();
}

public boolean addToQueue(final LLMObsEval eval) {
return queue.offer(eval);
public boolean addToQueue(final T payload) {
return queue.offer(payload);
}

@Override
Expand All @@ -94,41 +101,42 @@ public void close() {
}
}

public static class EvalSerializingHandler implements Runnable {
public static class SerializingHandler<T> implements Runnable {

private static final Logger log = LoggerFactory.getLogger(EvalSerializingHandler.class);
private static final Logger log = LoggerFactory.getLogger(SerializingHandler.class);
private static final int FLUSH_THRESHOLD = 50;

private final MessagePassingBlockingQueue<LLMObsEval> queue;
private final String payloadDescription;
private final MessagePassingBlockingQueue<T> queue;
private final long ticksRequiredToFlush;
private long lastTicks;

private final Moshi moshi;
private final JsonAdapter<LLMObsEval.Request> evalJsonAdapter;
private final BatchSerializer<T> serializer;
private final OkHttpClient httpClient;
private final HttpUrl submissionUrl;
private final Headers headers;

private final List<LLMObsEval> buffer = new ArrayList<>();
private final List<T> buffer = new ArrayList<>();

public EvalSerializingHandler(
final MessagePassingBlockingQueue<LLMObsEval> queue,
public SerializingHandler(
final String payloadDescription,
final MessagePassingBlockingQueue<T> queue,
final long flushInterval,
final TimeUnit timeUnit,
final HttpUrl submissionUrl,
final Headers headers) {
final Headers headers,
final BatchSerializer<T> serializer) {
this.payloadDescription = payloadDescription;
this.queue = queue;
this.moshi = new Moshi.Builder().add(LLMObsEval.class, new LLMObsEval.Adapter()).build();

this.evalJsonAdapter = moshi.adapter(LLMObsEval.Request.class);
this.serializer = serializer;
this.httpClient = new OkHttpClient();
this.submissionUrl = submissionUrl;
this.headers = headers;

this.lastTicks = System.nanoTime();
this.ticksRequiredToFlush = timeUnit.toNanos(flushInterval);

log.debug("starting eval metric serializer, url={}", submissionUrl);
log.debug("starting {} serializer, url={}", payloadDescription, submissionUrl);
}

@Override
Expand All @@ -139,16 +147,17 @@ public void run() {
Thread.currentThread().interrupt();
}
log.debug(
"eval processor worker exited. submitting evals stopped. unsubmitted evals left: {}",
"{} processor worker exited. submitting stopped. unsubmitted payloads left: {}",
payloadDescription,
!queuesAreEmpty());
}

private void runDutyCycle() throws InterruptedException {
Thread thread = Thread.currentThread();
while (!thread.isInterrupted()) {
LLMObsEval eval = queue.poll(100, TimeUnit.MILLISECONDS);
if (eval != null) {
buffer.add(eval);
T payload = queue.poll(100, TimeUnit.MILLISECONDS);
if (payload != null) {
buffer.add(payload);
consumeBatch();
}
flushIfNecessary();
Expand All @@ -164,10 +173,22 @@ protected void flushIfNecessary() {
return;
}
if (shouldFlush()) {
LLMObsEval.Request llmobsEvalReq = new LLMObsEval.Request(this.buffer);
HttpRetryPolicy.Factory retryPolicyFactory = new HttpRetryPolicy.Factory(5, 100, 2.0, true);

String reqBod = evalJsonAdapter.toJson(llmobsEvalReq);
String reqBod;
try {
reqBod = serializer.toJson(this.buffer);
} catch (Exception e) {
// A batch that cannot be serialized will never serialize, so it is dropped rather than
// retried. Letting this escape would kill the worker and strand every later payload.
log.error(
"Could not serialize {} payloads, dropping {} of them",
payloadDescription,
this.buffer.size(),
e);
this.buffer.clear();
return;
}

RequestBody requestBody =
RequestBody.create(okhttp3.MediaType.parse("application/json"), reqBod);
Expand All @@ -178,16 +199,20 @@ protected void flushIfNecessary() {
OkHttpUtils.sendWithRetries(httpClient, retryPolicyFactory, request)) {

if (response.isSuccessful()) {
log.debug("successfully flushed evaluation request with {} evals", this.buffer.size());
log.debug(
"successfully flushed {} request with {} payloads",
payloadDescription,
this.buffer.size());
this.buffer.clear();
} else {
log.error(
"Could not submit eval metrics (HTTP code {}) {}",
"Could not submit {} (HTTP code {}) {}",
payloadDescription,
response.code(),
response.body() != null ? response.body().string() : "");
}
} catch (Exception e) {
log.error("Could not submit eval metrics", e);
log.error("Could not submit " + payloadDescription, e);
}
}
}
Expand Down
97 changes: 95 additions & 2 deletions dd-java-agent/agent-llmobs/src/main/java/datadog/trace/llmobs/LLMObsSystem.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,10 +6,13 @@
import datadog.trace.api.llmobs.LLMObs;
import datadog.trace.api.llmobs.LLMObsSpan;
import datadog.trace.api.llmobs.LLMObsTags;
import datadog.trace.api.telemetry.LLMObsMetricCollector;
import datadog.trace.bootstrap.instrumentation.api.Tags;
import datadog.trace.llmobs.domain.DDLLMObsSpan;
import datadog.trace.llmobs.domain.LLMObsEval;
import datadog.trace.llmobs.domain.LLMObsFeedbackEvent;
import datadog.trace.llmobs.domain.LLMObsInternal;
import datadog.trace.util.AgentThreadFactory.AgentThread;
import java.lang.instrument.Instrumentation;
import java.util.Map;
import java.util.concurrent.TimeUnit;
Expand All @@ -23,6 +26,15 @@ public class LLMObsSystem {

private static final String CUSTOM_MODEL_VAL = "custom";

private static final String EVAL_METRIC_API_PATH = "api/intake/llm-obs/v1/eval-metric";
// Feedback is a v2 concept: submitter, the feedback-only targets and the non-score value types
// only exist there. Evaluations deliberately stay on v1 with their existing flat payload; both
// now carry event_kind, so the two are told apart the same way as in dd-trace-py and dd-trace-js.
private static final String FEEDBACK_API_PATH = "api/intake/llm-obs/v2/eval-metric";

private static final int QUEUE_CAPACITY = 1024;
private static final long FLUSH_INTERVAL_MS = 100;

public static void start(Instrumentation inst, SharedCommunicationObjects sco) {
Config config = Config.get();
if (!config.isLlmObsEnabled()) {
Expand All @@ -37,18 +49,99 @@ public static void start(Instrumentation inst, SharedCommunicationObjects sco) {
LLMObsInternal.setLLMObsSpanFactory(new LLMObsManualSpanFactory(mlApp, wellKnownTags));

LLMObsInternal.setLLMObsEvalProcessor(new LLMObsCustomEvalProcessor(mlApp, sco, config));

LLMObsInternal.setLLMObsFeedbackProcessor(
new LLMObsCustomFeedbackProcessor(mlApp, sco, config));
}

private static class LLMObsCustomFeedbackProcessor implements LLMObs.LLMObsFeedbackProcessor {
private final String defaultMLApp;
private final LLMObsIntakeWorker<LLMObsFeedbackEvent> feedbackProcessingWorker;

public LLMObsCustomFeedbackProcessor(
String defaultMLApp, SharedCommunicationObjects sco, Config config) {

this.defaultMLApp = defaultMLApp;
this.feedbackProcessingWorker =
new LLMObsIntakeWorker<>(
"feedback",
FEEDBACK_API_PATH,
AgentThread.LLMOBS_FEEDBACK_PROCESSOR,
QUEUE_CAPACITY,
FLUSH_INTERVAL_MS,
TimeUnit.MILLISECONDS,
sco,
config,
LLMObsFeedbackEvent.batchSerializer());
this.feedbackProcessingWorker.start();
}

@Override
public void submitFeedback(LLMObs.Feedback feedback) {
if (feedback == null) {
LLMObsMetricCollector.get().recordFeedbackSubmitted(null, null, "invalid_feedback");
LOGGER.error("null feedback provided, feedback not recorded");
return;
}

// The builder never throws so that instrumented code stays safe when the agent is absent;
// validation happens here instead, only once LLM Observability is actually enabled.
LLMObs.Feedback.ValidationError error = feedback.validate();
if (error != null) {
recordFeedbackTelemetry(feedback, error.getCode());
throw new IllegalArgumentException(error.getMessage());
}

String mlApp = feedback.getMlApp();
if (mlApp == null || mlApp.isEmpty()) {
mlApp = defaultMLApp;
}

if (!this.feedbackProcessingWorker.addToQueue(new LLMObsFeedbackEvent(feedback, mlApp))) {
recordFeedbackTelemetry(feedback, "queue_full");
LOGGER.warn(
"queue full, failed to add feedback, ml_app={}, {}={}, label={}",
mlApp,
feedback.getTargetType().getWireKey(),
feedback.getTargetValue(),
feedback.getLabel());
return;
}

recordFeedbackTelemetry(feedback, null);
}

private static void recordFeedbackTelemetry(
LLMObs.Feedback feedback, @Nullable String errorCode) {
LLMObs.Feedback.MetricType metricType = feedback.getMetricType();
LLMObs.Feedback.TargetType targetType = feedback.getTargetType();
LLMObsMetricCollector.get()
.recordFeedbackSubmitted(
metricType == null ? null : metricType.toString(),
targetType == null ? null : targetType.getWireKey(),
errorCode);
}
}

private static class LLMObsCustomEvalProcessor implements LLMObs.LLMObsEvalProcessor {
private final String defaultMLApp;
private final EvalProcessingWorker evalProcessingWorker;
private final LLMObsIntakeWorker<LLMObsEval> evalProcessingWorker;

public LLMObsCustomEvalProcessor(
String defaultMLApp, SharedCommunicationObjects sco, Config config) {

this.defaultMLApp = defaultMLApp;
this.evalProcessingWorker =
new EvalProcessingWorker(1024, 100, TimeUnit.MILLISECONDS, sco, config);
new LLMObsIntakeWorker<>(
"eval metrics",
EVAL_METRIC_API_PATH,
AgentThread.LLMOBS_EVALS_PROCESSOR,
QUEUE_CAPACITY,
FLUSH_INTERVAL_MS,
TimeUnit.MILLISECONDS,
sco,
config,
LLMObsEval.batchSerializer());
this.evalProcessingWorker.start();
}

Expand Down
Loading