Skip to content

Commit c41dc73

Browse files
realarkrealark
committed
tag vendor ids on openai and anthropic llm spans
1 parent e49d40a commit c41dc73

19 files changed

Lines changed: 1178 additions & 209 deletions

File tree

‎AGENTS.md‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -160,11 +160,12 @@ VCR_MODE=record ./gradlew :braintrust-sdk:test --tests 'dev.braintrust.devserver
160160
- when running btx, use the spec filter to target what is specifically under development: `VCR_MODE=off ./gradlew :btx:test -Pbtx.spec.filter=openai/prompt_cach --rerun`
161161
- don't reformat the whole repo, but do run `./gradlew spotlessApply` on files you changed before committing. the pre-commit hook and `./gradlew check` both run `spotlessCheck`, which fails on unformatted code.
162162

163-
## Gotchas
163+
## Misc Tips and Best Practices
164164

165165
- **don't hand-edit cassettes.** they're content-hashed and guarded against committed secrets. a failing VCR test means the recorded interaction changed — re-record it (see the VCR section), don't patch the json.
166166
- **`braintrust-api` is generated code.** don't edit sources under it by hand; it's regenerated from the braintrust openapi spec pinned as `braintrustOpenApiRef` in gradle.properties.
167167
- **there are no version constants to bump.** the sdk version is derived from git tags at build time (`generateVersion()` in build.gradle) and written into braintrust.properties. "bump the version" is not a source change.
168+
- When adding test cases, favor adding to the test file of the module being changed rather than making a new file. For example, if you fix a bug in the `Foo` module, add the test case to `FooTest.java` instead of making a new file, `FooTestMyBuggyCase.java`
168169

169170
## Releasing
170171

‎braintrust-sdk/instrumentation/anthropic_2_2_0/src/main/java/dev/braintrust/instrumentation/anthropic/v2_2_0/ResponseReassembler.java‎

Lines changed: 117 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,117 @@
1+
package dev.braintrust.instrumentation.anthropic.v2_2_0;
2+
3+
import com.anthropic.helpers.MessageAccumulator;
4+
import com.anthropic.models.messages.RawMessageStreamEvent;
5+
import dev.braintrust.json.BraintrustJsonMapper;
6+
import java.io.BufferedReader;
7+
import java.io.ByteArrayInputStream;
8+
import java.io.InputStreamReader;
9+
import java.nio.charset.StandardCharsets;
10+
import javax.annotation.Nullable;
11+
import lombok.extern.slf4j.Slf4j;
12+
13+
/**
14+
* Turns the raw bytes of an Anthropic response into a single JSON document the semconv layer can
15+
* tag.
16+
*
17+
* <p>All of the wire-format bookkeeping lives here — SSE-vs-plain-JSON detection and chunk
18+
* reassembly — so that {@code TracingHttpClient} is left holding only the span lifecycle and one
19+
* flat call into {@code InstrumentationSemConv}.
20+
*/
21+
@Slf4j
22+
class ResponseReassembler {
23+
24+
private ResponseReassembler() {}
25+
26+
/**
27+
* A reassembled response body plus the timing that belongs with it.
28+
*
29+
* <p>{@code body} is null when there was nothing usable to reassemble — an empty response, or
30+
* one we couldn't parse. Callers should still tag the response in that case; the headers remain
31+
* worth recording.
32+
*
33+
* <p>{@code timeToFirstTokenNanos} is only populated for a stream, since a non-streaming
34+
* response has no first token to time.
35+
*/
36+
record Result(@Nullable String body, @Nullable Long timeToFirstTokenNanos) {
37+
static final Result EMPTY = new Result(null, null);
38+
}
39+
40+
/** Detects the wire format and reassembles accordingly. Never throws. */
41+
static Result reassemble(byte[] bytes, long timeToFirstTokenNanos) {
42+
if (bytes.length == 0) {
43+
return Result.EMPTY;
44+
}
45+
try {
46+
String firstLine = firstNonEmptyLine(bytes);
47+
// Anthropic SSE starts with "event: message_start\ndata: ..." so we detect either
48+
// prefix. OpenAI SSE starts directly with "data:".
49+
boolean isSse =
50+
firstLine != null
51+
&& (firstLine.startsWith("data:") || firstLine.startsWith("event:"));
52+
if (isSse) {
53+
return new Result(reassembleSse(bytes), timeToFirstTokenNanos);
54+
}
55+
// Non-streaming: plain Message JSON — pass it whole, no time_to_first_token
56+
return new Result(new String(bytes, StandardCharsets.UTF_8), null);
57+
} catch (Exception e) {
58+
log.error("Could not reassemble Anthropic response buffer", e);
59+
return Result.EMPTY;
60+
}
61+
}
62+
63+
@Nullable
64+
private static String firstNonEmptyLine(byte[] bytes) {
65+
int start = 0;
66+
for (int i = 0; i <= bytes.length; i++) {
67+
if (i == bytes.length || bytes[i] == '\n') {
68+
String line = new String(bytes, start, i - start, StandardCharsets.UTF_8).strip();
69+
if (!line.isEmpty()) return line;
70+
start = i + 1;
71+
}
72+
}
73+
return null;
74+
}
75+
76+
/**
77+
* Anthropic SSE wire format has named events:
78+
*
79+
* <pre>
80+
* event: message_start
81+
* data: {"type":"message_start","message":{...}}
82+
*
83+
* event: content_block_delta
84+
* data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hi"}}
85+
* </pre>
86+
*
87+
* We only need the {@code data:} lines — the event name is redundant with the {@code type}
88+
* field inside the JSON. Feed each data payload to {@link MessageAccumulator} and serialize the
89+
* assembled {@link com.anthropic.models.messages.Message}.
90+
*/
91+
@Nullable
92+
private static String reassembleSse(byte[] sseBytes) {
93+
try {
94+
var mapper = BraintrustJsonMapper.get();
95+
var reader =
96+
new BufferedReader(
97+
new InputStreamReader(
98+
new ByteArrayInputStream(sseBytes), StandardCharsets.UTF_8));
99+
var accumulator = MessageAccumulator.create();
100+
String line;
101+
while ((line = reader.readLine()) != null) {
102+
if (!line.startsWith("data:")) continue;
103+
String data = line.substring("data:".length()).strip();
104+
if (data.isEmpty()) continue;
105+
try {
106+
accumulator.accumulate(mapper.readValue(data, RawMessageStreamEvent.class));
107+
} catch (Exception ignored) {
108+
// skip unrecognized event types (e.g. ping)
109+
}
110+
}
111+
return BraintrustJsonMapper.toJson(accumulator.message());
112+
} catch (Exception e) {
113+
log.error("Could not parse Anthropic SSE buffer to tag streaming span output", e);
114+
return null;
115+
}
116+
}
117+
}

‎braintrust-sdk/instrumentation/anthropic_2_2_0/src/main/java/dev/braintrust/instrumentation/anthropic/v2_2_0/TracingHttpClient.java‎

Lines changed: 49 additions & 87 deletions
Original file line numberDiff line numberDiff line change
@@ -5,25 +5,22 @@
55
import com.anthropic.core.http.HttpRequest;
66
import com.anthropic.core.http.HttpRequestBody;
77
import com.anthropic.core.http.HttpResponse;
8-
import com.anthropic.helpers.MessageAccumulator;
9-
import com.anthropic.models.messages.RawMessageStreamEvent;
108
import dev.braintrust.bootstrap.BraintrustBridge;
119
import dev.braintrust.instrumentation.InstrumentationSemConv;
12-
import dev.braintrust.json.BraintrustJsonMapper;
1310
import io.opentelemetry.api.OpenTelemetry;
1411
import io.opentelemetry.api.trace.Span;
1512
import io.opentelemetry.api.trace.SpanContext;
1613
import io.opentelemetry.api.trace.TraceFlags;
1714
import io.opentelemetry.api.trace.TraceState;
1815
import io.opentelemetry.api.trace.Tracer;
1916
import io.opentelemetry.context.Context;
20-
import java.io.BufferedReader;
21-
import java.io.ByteArrayInputStream;
2217
import java.io.ByteArrayOutputStream;
2318
import java.io.InputStream;
24-
import java.io.InputStreamReader;
2519
import java.io.OutputStream;
2620
import java.nio.charset.StandardCharsets;
21+
import java.util.HashMap;
22+
import java.util.List;
23+
import java.util.Map;
2724
import java.util.concurrent.CompletableFuture;
2825
import java.util.concurrent.atomic.AtomicBoolean;
2926
import java.util.concurrent.atomic.AtomicLong;
@@ -125,7 +122,9 @@ public void close() {
125122
bufferedRequest.baseUrl() != null ? bufferedRequest.baseUrl() : "",
126123
bufferedRequest.pathSegments(),
127124
bufferedRequest.method().name(),
128-
inputJson);
125+
inputJson,
126+
null,
127+
headersAsMap(bufferedRequest.headers()));
129128

130129
var response = underlying.execute(bufferedRequest, requestOptions);
131130
return new TeeingStreamHttpResponse(response, span, tracer);
@@ -153,7 +152,9 @@ public void close() {
153152
bufferedRequest.baseUrl() != null ? bufferedRequest.baseUrl() : "",
154153
bufferedRequest.pathSegments(),
155154
bufferedRequest.method().name(),
156-
inputJson);
155+
inputJson,
156+
null,
157+
headersAsMap(bufferedRequest.headers()));
157158
return underlying
158159
.executeAsync(bufferedRequest, requestOptions)
159160
.thenApply(
@@ -264,9 +265,34 @@ private void onStreamClosed() {
264265
synchronized (teeBuffer) {
265266
bytes = teeBuffer.toByteArray();
266267
}
268+
269+
// Recorded before tagging: the anthropic sdk raises above this layer, so the
270+
// error status is ours alone to set, and losing it to a body-parsing problem is
271+
// worse than losing the parsed output.
272+
// Anything outside 2xx, not just 4xx/5xx: both vendor SDKs treat success as
273+
// exactly 200..299, so a final 3xx that the http client did not follow (a 304, or
274+
// a redirect with no usable Location) is raised to the caller as an
275+
// UnexpectedStatusCodeException and must mark the span failed too.
276+
int statusCode = delegate.statusCode();
277+
if (statusCode < 200 || statusCode >= 300) {
278+
InstrumentationSemConv.tagLLMSpanHttpError(
279+
span, statusCode, new String(bytes, StandardCharsets.UTF_8));
280+
}
281+
282+
// Wire-format bookkeeping lives in ResponseReassembler; this hands semconv
283+
// everything the response carried in one flat call. A null body (empty or
284+
// unparseable response) still tags the headers.
267285
// tagLLMSpanResponse also emits child spans for any server-side tool calls (web
268286
// search, etc.) nested under the LLM span while it is still live.
269-
tagSpanFromBuffer(tracer, span, bytes, timeToFirstTokenNanos.get());
287+
var reassembled =
288+
ResponseReassembler.reassemble(bytes, timeToFirstTokenNanos.get());
289+
InstrumentationSemConv.tagLLMSpanResponse(
290+
tracer,
291+
span,
292+
InstrumentationSemConv.PROVIDER_NAME_ANTHROPIC,
293+
reassembled.body(),
294+
reassembled.timeToFirstTokenNanos(),
295+
headersAsMap(delegate.headers()));
270296
} finally {
271297
span.end();
272298
}
@@ -360,89 +386,25 @@ private void notifyClosed() {
360386
// Span tagging from buffered bytes
361387
// -------------------------------------------------------------------------
362388

363-
private static void tagSpanFromBuffer(
364-
Tracer tracer, Span span, byte[] bytes, Long timeToFirstTokenNanos) {
365-
if (bytes.length == 0) return;
366-
try {
367-
String firstLine = firstNonEmptyLine(bytes);
368-
// Anthropic SSE starts with "event: message_start\ndata: ..." so we detect
369-
// either prefix. OpenAI SSE starts directly with "data:".
370-
boolean isSse =
371-
firstLine != null
372-
&& (firstLine.startsWith("data:") || firstLine.startsWith("event:"));
373-
if (isSse) {
374-
tagSpanFromSseBytes(tracer, span, bytes, timeToFirstTokenNanos);
375-
} else {
376-
// Non-streaming: plain Message JSON — pass it whole, no time_to_first_token
377-
String responseJson = new String(bytes, StandardCharsets.UTF_8);
378-
InstrumentationSemConv.tagLLMSpanResponse(
379-
tracer,
380-
span,
381-
InstrumentationSemConv.PROVIDER_NAME_ANTHROPIC,
382-
responseJson,
383-
null);
384-
}
385-
} catch (Exception e) {
386-
log.error("Could not tag span from Anthropic response buffer", e);
387-
}
388-
}
389-
390-
private static String firstNonEmptyLine(byte[] bytes) {
391-
int start = 0;
392-
for (int i = 0; i <= bytes.length; i++) {
393-
if (i == bytes.length || bytes[i] == '\n') {
394-
String line = new String(bytes, start, i - start, StandardCharsets.UTF_8).strip();
395-
if (!line.isEmpty()) return line;
396-
start = i + 1;
397-
}
398-
}
399-
return null;
400-
}
401-
402389
/**
403-
* Anthropic SSE wire format has named events:
404-
*
405-
* <pre>
406-
* event: message_start
407-
* data: {"type":"message_start","message":{...}}
408-
*
409-
* event: content_block_delta
410-
* data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hi"}}
411-
* </pre>
412-
*
413-
* We only need the {@code data:} lines — the event name is redundant with the {@code type}
414-
* field inside the JSON. Feed each data payload to {@link MessageAccumulator} and serialize the
415-
* assembled {@link com.anthropic.models.messages.Message} for the span.
390+
* Adapts the anthropic sdk's {@code Headers} to the vendor-neutral shape {@link
391+
* InstrumentationSemConv} consumes. Returns an empty map on failure so a header-shape change
392+
* can never take down the tagging that follows it.
416393
*/
417-
private static void tagSpanFromSseBytes(
418-
Tracer tracer, Span span, byte[] sseBytes, Long timeToFirstTokenNanos) {
394+
private static Map<String, List<String>> headersAsMap(
395+
@Nullable com.anthropic.core.http.Headers headers) {
396+
if (headers == null) {
397+
return Map.of();
398+
}
419399
try {
420-
var mapper = BraintrustJsonMapper.get();
421-
var reader =
422-
new BufferedReader(
423-
new InputStreamReader(
424-
new ByteArrayInputStream(sseBytes), StandardCharsets.UTF_8));
425-
var accumulator = MessageAccumulator.create();
426-
String line;
427-
while ((line = reader.readLine()) != null) {
428-
if (!line.startsWith("data:")) continue;
429-
String data = line.substring("data:".length()).strip();
430-
if (data.isEmpty()) continue;
431-
try {
432-
accumulator.accumulate(mapper.readValue(data, RawMessageStreamEvent.class));
433-
} catch (Exception ignored) {
434-
// skip unrecognized event types (e.g. ping)
435-
}
400+
var map = new HashMap<String, List<String>>();
401+
for (String name : headers.names()) {
402+
map.put(name, headers.values(name));
436403
}
437-
String assembledMessageJson = BraintrustJsonMapper.toJson(accumulator.message());
438-
InstrumentationSemConv.tagLLMSpanResponse(
439-
tracer,
440-
span,
441-
InstrumentationSemConv.PROVIDER_NAME_ANTHROPIC,
442-
assembledMessageJson,
443-
timeToFirstTokenNanos);
404+
return map;
444405
} catch (Exception e) {
445-
log.error("Could not parse Anthropic SSE buffer to tag streaming span output", e);
406+
log.debug("could not read headers", e);
407+
return Map.of();
446408
}
447409
}
448410
}

‎braintrust-sdk/instrumentation/anthropic_2_2_0/src/main/java/dev/braintrust/instrumentation/anthropic/v2_2_0/auto/AnthropicInstrumentationModule.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,8 @@ public List<String> getHelperClassNames() {
3636
MANUAL_INSTRUMENTATION_PACKAGE + "TracingHttpClient$ExtractedRequest",
3737
MANUAL_INSTRUMENTATION_PACKAGE + "BraintrustAnthropic",
3838
MANUAL_INSTRUMENTATION_PACKAGE + "ContextCapturingProxy",
39+
MANUAL_INSTRUMENTATION_PACKAGE + "ResponseReassembler",
40+
MANUAL_INSTRUMENTATION_PACKAGE + "ResponseReassembler$Result",
3941
"dev.braintrust.json.BraintrustJsonMapper",
4042
"dev.braintrust.instrumentation.InstrumentationSemConv");
4143
}

0 commit comments

Comments
 (0)