diff --git a/.github/workflows/maven-ci.yml b/.github/workflows/maven-ci.yml index ad6433264..9d24e910a 100644 --- a/.github/workflows/maven-ci.yml +++ b/.github/workflows/maven-ci.yml @@ -109,8 +109,16 @@ jobs: - name: Build and Test with Coverage [Linux] if: runner.os == 'Linux' + shell: bash run: | export PATH="$HOME/.mvnd/bin:$PATH" + if ! command -v mvnd >/dev/null 2>&1; then + echo "mvnd not found, installing..." + curl -fsSL https://github.com/apache/maven-mvnd/releases/download/${{ env.MVND_VERSION }}/maven-mvnd-${{ env.MVND_VERSION }}-linux-amd64.tar.gz | tar xz + mkdir -p ~/.mvnd + mv maven-mvnd-${{ env.MVND_VERSION }}-linux-amd64/* ~/.mvnd/ + export PATH="$HOME/.mvnd/bin:$PATH" + fi # -T1: single-threaded build so reactor order is respected and agentscope-core # is installed before modules (e.g. subagent) that resolve it from the reactor. mvnd -B -T1 clean verify diff --git a/agentscope-core/pom.xml b/agentscope-core/pom.xml index 54bf500a7..b5bb9a12b 100644 --- a/agentscope-core/pom.xml +++ b/agentscope-core/pom.xml @@ -140,5 +140,11 @@ com.networknt json-schema-validator + + + io.opentelemetry + opentelemetry-api + test + diff --git a/agentscope-core/src/main/java/io/agentscope/core/hook/recorder/JsonlTraceExporter.java b/agentscope-core/src/main/java/io/agentscope/core/hook/recorder/JsonlTraceExporter.java new file mode 100644 index 000000000..9d60df268 --- /dev/null +++ b/agentscope-core/src/main/java/io/agentscope/core/hook/recorder/JsonlTraceExporter.java @@ -0,0 +1,543 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.agentscope.core.hook.recorder; + +import io.agentscope.core.hook.ActingChunkEvent; +import io.agentscope.core.hook.ErrorEvent; +import io.agentscope.core.hook.Hook; +import io.agentscope.core.hook.HookEvent; +import io.agentscope.core.hook.HookEventType; +import io.agentscope.core.hook.PostActingEvent; +import io.agentscope.core.hook.PostCallEvent; +import io.agentscope.core.hook.PostReasoningEvent; +import io.agentscope.core.hook.PostSummaryEvent; +import io.agentscope.core.hook.PreActingEvent; +import io.agentscope.core.hook.PreCallEvent; +import io.agentscope.core.hook.PreReasoningEvent; +import io.agentscope.core.hook.PreSummaryEvent; +import io.agentscope.core.hook.ReasoningChunkEvent; +import io.agentscope.core.hook.ReasoningEvent; +import io.agentscope.core.hook.SummaryChunkEvent; +import io.agentscope.core.util.JsonUtils; +import java.io.BufferedWriter; +import java.io.IOException; +import java.io.PrintWriter; +import java.io.StringWriter; +import java.lang.reflect.Method; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.OpenOption; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.util.EnumSet; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.UUID; +import java.util.WeakHashMap; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Predicate; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import reactor.core.publisher.Mono; + +/** + * A built-in, out-of-the-box JSONL trace exporter based on the Hook event system. + * + *

Each HookEvent is written as a single JSON object per line. This is designed for local + * debugging, offline troubleshooting, and attaching logs to issues. + * + *

Notes: + * + *

+ */ +public final class JsonlTraceExporter implements Hook, AutoCloseable { + + private static final Logger log = LoggerFactory.getLogger(JsonlTraceExporter.class); + private static final long CLOSE_TIMEOUT_SECONDS = 30L; + private static final OpenTelemetryAccess OPEN_TELEMETRY_ACCESS = OpenTelemetryAccess.create(); + + private final Path outputFile; + private final boolean flushEveryLine; + private final boolean failFast; + private final int priority; + private final Predicate eventFilter; + + private final Object lock = new Object(); + private final BufferedWriter writer; + private final ExecutorService exportExecutor; + private final AtomicBoolean closed = new AtomicBoolean(false); + + // WeakHashMap keeps per-agent run state from accumulating indefinitely after agent instances + // become unreachable. Concurrency safety does not rely on WeakHashMap itself: all access to + // this map is serialized through the exporter’s single-threaded queue. + private final Map runStates = new WeakHashMap<>(); + + private JsonlTraceExporter( + Path outputFile, + boolean append, + boolean flushEveryLine, + boolean failFast, + int priority, + Predicate eventFilter) { + this.outputFile = Objects.requireNonNull(outputFile, "outputFile cannot be null"); + this.flushEveryLine = flushEveryLine; + this.failFast = failFast; + this.priority = priority; + this.eventFilter = Objects.requireNonNull(eventFilter, "eventFilter cannot be null"); + this.writer = openWriter(outputFile, append); + this.exportExecutor = createExportExecutor(); + } + + public static Builder builder(Path outputFile) { + return new Builder(outputFile); + } + + @Override + public int priority() { + return priority; + } + + @Override + public Mono onEvent(T event) { + T nonNullEvent = Objects.requireNonNull(event, "event cannot be null"); + if (!eventFilter.test(nonNullEvent)) { + return Mono.just(nonNullEvent); + } + + return Mono.defer(() -> enqueueWrite(nonNullEvent, OPEN_TELEMETRY_ACCESS.captureCurrent())) + .onErrorResume( + error -> { + if (failFast) { + return Mono.error(error); + } + log.warn( + "Failed to export hook event to JSONL: {}", + error.getMessage(), + error); + return Mono.just(nonNullEvent); + }); + } + + private Mono enqueueWrite(T event, OpenTelemetryIds openTelemetryIds) { + if (closed.get()) { + return Mono.error( + new RejectedExecutionException( + "JSONL exporter is closed: " + outputFile.toAbsolutePath())); + } + + CompletableFuture future = new CompletableFuture<>(); + try { + exportExecutor.execute( + () -> { + try { + writeEvent(event, openTelemetryIds); + future.complete(event); + } catch (Throwable error) { + future.completeExceptionally(error); + } + }); + } catch (RejectedExecutionException error) { + future.completeExceptionally(error); + } + return Mono.fromFuture(future); + } + + private void writeEvent(HookEvent event, OpenTelemetryIds openTelemetryIds) throws IOException { + RunState runState = getOrUpdateRunState(event); + + Map record = new LinkedHashMap<>(); + record.put("ts", event.getTimestamp()); + record.put("event_type", event.getType().name()); + record.put("agent_id", event.getAgent().getAgentId()); + record.put("agent_name", event.getAgent().getName()); + + record.put("run_id", runState.currentRunId); + record.put("turn_id", runState.turnId); + record.put("step_id", runState.stepId); + + if (openTelemetryIds != null) { + openTelemetryIds.putIfPresent(record); + } + + if (event instanceof ReasoningEvent reasoningEvent) { + record.put("model_name", reasoningEvent.getModelName()); + record.put("generate_options", reasoningEvent.getGenerateOptions()); + } + + if (event instanceof PreCallEvent e) { + record.put("input_messages", e.getInputMessages()); + } else if (event instanceof PostCallEvent e) { + record.put("final_message", e.getFinalMessage()); + } else if (event instanceof PreReasoningEvent e) { + record.put("input_messages", e.getInputMessages()); + record.put("effective_generate_options", e.getEffectiveGenerateOptions()); + } else if (event instanceof PostReasoningEvent e) { + record.put("reasoning_message", e.getReasoningMessage()); + record.put("stop_requested", e.isStopRequested()); + record.put("goto_reasoning_requested", e.isGotoReasoningRequested()); + if (e.isGotoReasoningRequested()) { + record.put("goto_reasoning_msgs", e.getGotoReasoningMsgs()); + } + } else if (event instanceof ReasoningChunkEvent e) { + record.put("incremental_chunk", e.getIncrementalChunk()); + record.put("accumulated", e.getAccumulated()); + } else if (event instanceof PreActingEvent e) { + record.put("tool_use", e.getToolUse()); + } else if (event instanceof ActingChunkEvent e) { + record.put("tool_use", e.getToolUse()); + record.put("incremental_chunk", e.getChunk()); + record.put("chunk", e.getChunk()); + } else if (event instanceof PostActingEvent e) { + record.put("tool_use", e.getToolUse()); + record.put("tool_result", e.getToolResult()); + } else if (event instanceof PreSummaryEvent e) { + record.put("model_name", e.getModelName()); + record.put("input_messages", e.getInputMessages()); + record.put("max_iterations", e.getMaxIterations()); + record.put("current_iteration", e.getCurrentIteration()); + record.put("effective_generate_options", e.getEffectiveGenerateOptions()); + } else if (event instanceof SummaryChunkEvent e) { + record.put("incremental_chunk", e.getIncrementalChunk()); + record.put("accumulated", e.getAccumulated()); + } else if (event instanceof PostSummaryEvent e) { + record.put("summary_message", e.getSummaryMessage()); + } else if (event instanceof ErrorEvent e) { + record.put("error_class", e.getError().getClass().getName()); + record.put("error_message", e.getError().getMessage()); + record.put("stacktrace", stackTraceToString(e.getError())); + } + + String line = JsonUtils.getJsonCodec().toJson(record); + synchronized (lock) { + writer.write(line); + writer.newLine(); + if (flushEveryLine) { + writer.flush(); + } + } + } + + private RunState getOrUpdateRunState(HookEvent event) { + String agentId = event.getAgent().getAgentId(); + RunState state = runStates.computeIfAbsent(agentId, unused -> new RunState()); + if (event.getType() == HookEventType.PRE_CALL) { + state.currentRunId = UUID.randomUUID().toString(); + state.turnId++; + state.stepId = 0; + } else { + state.stepId++; + } + return state; + } + + private static BufferedWriter openWriter(Path outputFile, boolean append) { + try { + Path parent = outputFile.toAbsolutePath().getParent(); + if (parent != null) { + Files.createDirectories(parent); + } + + OpenOption[] options = + append + ? new OpenOption[] { + StandardOpenOption.CREATE, StandardOpenOption.APPEND + } + : new OpenOption[] { + StandardOpenOption.CREATE, + StandardOpenOption.TRUNCATE_EXISTING, + StandardOpenOption.WRITE + }; + return Files.newBufferedWriter(outputFile, StandardCharsets.UTF_8, options); + } catch (IOException e) { + throw new IllegalStateException( + "Failed to open JSONL output file: " + outputFile.toAbsolutePath(), e); + } + } + + private static String stackTraceToString(Throwable error) { + if (error == null) { + return null; + } + StringWriter sw = new StringWriter(); + try (PrintWriter pw = new PrintWriter(sw)) { + error.printStackTrace(pw); + } + return sw.toString(); + } + + @Override + public void close() throws IOException { + if (!closed.compareAndSet(false, true)) { + return; + } + + boolean drained = false; + try { + Future barrier = exportExecutor.submit(() -> {}); + barrier.get(CLOSE_TIMEOUT_SECONDS, TimeUnit.SECONDS); + drained = true; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException( + "Interrupted while waiting for JSONL exporter to finish pending writes", e); + } catch (TimeoutException e) { + throw new IOException( + "Timed out while waiting for JSONL exporter to finish pending writes", e); + } catch (ExecutionException e) { + throw new IOException( + "Failed while waiting for JSONL exporter to finish pending writes", + e.getCause()); + } finally { + if (drained) { + exportExecutor.shutdown(); + } else { + exportExecutor.shutdownNow(); + } + runStates.clear(); + synchronized (lock) { + writer.flush(); + writer.close(); + } + } + } + + private static ExecutorService createExportExecutor() { + ThreadFactory threadFactory = + runnable -> { + Thread thread = new Thread(runnable, "agentscope-jsonl-trace-exporter"); + thread.setDaemon(true); + return thread; + }; + return Executors.newSingleThreadExecutor(threadFactory); + } + + private static final class RunState { + private String currentRunId = UUID.randomUUID().toString(); + private long turnId = 0; + private long stepId = 0; + } + + private static final class OpenTelemetryAccess { + private final Method currentMethod; + private final Method getSpanContextMethod; + private final Method isValidMethod; + private final Method getTraceIdMethod; + private final Method getSpanIdMethod; + + private OpenTelemetryAccess( + Method currentMethod, + Method getSpanContextMethod, + Method isValidMethod, + Method getTraceIdMethod, + Method getSpanIdMethod) { + this.currentMethod = currentMethod; + this.getSpanContextMethod = getSpanContextMethod; + this.isValidMethod = isValidMethod; + this.getTraceIdMethod = getTraceIdMethod; + this.getSpanIdMethod = getSpanIdMethod; + } + + private static OpenTelemetryAccess create() { + try { + ClassLoader classLoader = JsonlTraceExporter.class.getClassLoader(); + Class spanClass = + Class.forName("io.opentelemetry.api.trace.Span", false, classLoader); + Class spanContextClass = + Class.forName("io.opentelemetry.api.trace.SpanContext", false, classLoader); + return new OpenTelemetryAccess( + spanClass.getMethod("current"), + spanClass.getMethod("getSpanContext"), + spanContextClass.getMethod("isValid"), + spanContextClass.getMethod("getTraceId"), + spanContextClass.getMethod("getSpanId")); + } catch (Throwable ignored) { + return new OpenTelemetryAccess(null, null, null, null, null); + } + } + + private void putIfPresent(Map record) { + if (currentMethod == null) { + return; + } + OpenTelemetryIds openTelemetryIds = captureCurrent(); + if (openTelemetryIds != null) { + openTelemetryIds.putIfPresent(record); + } + } + + private OpenTelemetryIds captureCurrent() { + if (currentMethod == null) { + return null; + } + try { + Object span = currentMethod.invoke(null); + if (span == null) { + return null; + } + Object spanContext = getSpanContextMethod.invoke(span); + if (spanContext == null || !(boolean) isValidMethod.invoke(spanContext)) { + return null; + } + return new OpenTelemetryIds( + (String) getTraceIdMethod.invoke(spanContext), + (String) getSpanIdMethod.invoke(spanContext)); + } catch (Throwable ignored) { + return null; + } + } + } + + private static final class OpenTelemetryIds { + private final String traceId; + private final String spanId; + + private OpenTelemetryIds(String traceId, String spanId) { + this.traceId = traceId; + this.spanId = spanId; + } + + private void putIfPresent(Map record) { + if (traceId != null && spanId != null) { + record.put("trace_id", traceId); + record.put("span_id", spanId); + } + } + } + + public static final class Builder { + private final Path outputFile; + + private boolean append = true; + private boolean flushEveryLine = true; + private boolean failFast = false; + private int priority = 900; + + private Set enabledEvents = + EnumSet.of( + HookEventType.PRE_CALL, + HookEventType.POST_CALL, + HookEventType.PRE_REASONING, + HookEventType.POST_REASONING, + HookEventType.PRE_ACTING, + HookEventType.POST_ACTING, + HookEventType.ERROR); + + private boolean includeReasoningChunks = false; + private boolean includeActingChunks = false; + private boolean includeSummary = false; + private boolean includeSummaryChunks = false; + + private Builder(Path outputFile) { + this.outputFile = Objects.requireNonNull(outputFile, "outputFile cannot be null"); + } + + /** Appends to existing file if present (default: true). */ + public Builder append(boolean append) { + this.append = append; + return this; + } + + /** Flushes after each JSONL line (default: true). */ + public Builder flushEveryLine(boolean flushEveryLine) { + this.flushEveryLine = flushEveryLine; + return this; + } + + /** + * If enabled, exporter errors fail the agent execution (default: false, best-effort). + */ + public Builder failFast(boolean failFast) { + this.failFast = failFast; + return this; + } + + /** Hook priority (default: 900, low priority for logging/export). */ + public Builder priority(int priority) { + this.priority = priority; + return this; + } + + /** Enables exactly these event types (chunk and summary helpers still apply). */ + public Builder enabledEvents(Set enabledEvents) { + this.enabledEvents = + enabledEvents == null || enabledEvents.isEmpty() + ? EnumSet.noneOf(HookEventType.class) + : EnumSet.copyOf(enabledEvents); + return this; + } + + /** Includes reasoning streaming events (ReasoningChunkEvent). */ + public Builder includeReasoningChunks(boolean includeReasoningChunks) { + this.includeReasoningChunks = includeReasoningChunks; + return this; + } + + /** Includes tool streaming events (ActingChunkEvent). */ + public Builder includeActingChunks(boolean includeActingChunks) { + this.includeActingChunks = includeActingChunks; + return this; + } + + /** Includes summary events (PreSummaryEvent/PostSummaryEvent). */ + public Builder includeSummary(boolean includeSummary) { + this.includeSummary = includeSummary; + return this; + } + + /** Includes summary streaming events (SummaryChunkEvent). */ + public Builder includeSummaryChunks(boolean includeSummaryChunks) { + this.includeSummaryChunks = includeSummaryChunks; + return this; + } + + public JsonlTraceExporter build() { + EnumSet types = EnumSet.copyOf(enabledEvents); + if (includeReasoningChunks) { + types.add(HookEventType.REASONING_CHUNK); + } + if (includeActingChunks) { + types.add(HookEventType.ACTING_CHUNK); + } + if (includeSummary) { + types.add(HookEventType.PRE_SUMMARY); + types.add(HookEventType.POST_SUMMARY); + } + if (includeSummaryChunks) { + types.add(HookEventType.SUMMARY_CHUNK); + } + + Predicate filter = e -> e != null && types.contains(e.getType()); + return new JsonlTraceExporter( + outputFile, append, flushEveryLine, failFast, priority, filter); + } + } +} diff --git a/agentscope-core/src/test/java/io/agentscope/core/hook/recorder/JsonlTraceExporterTest.java b/agentscope-core/src/test/java/io/agentscope/core/hook/recorder/JsonlTraceExporterTest.java new file mode 100644 index 000000000..68b805621 --- /dev/null +++ b/agentscope-core/src/test/java/io/agentscope/core/hook/recorder/JsonlTraceExporterTest.java @@ -0,0 +1,476 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.agentscope.core.hook.recorder; + +import static io.agentscope.core.hook.HookEventType.PRE_CALL; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.JsonNode; +import io.agentscope.core.agent.Agent; +import io.agentscope.core.agent.Event; +import io.agentscope.core.agent.StreamOptions; +import io.agentscope.core.hook.ActingChunkEvent; +import io.agentscope.core.hook.ErrorEvent; +import io.agentscope.core.hook.PostActingEvent; +import io.agentscope.core.hook.PostCallEvent; +import io.agentscope.core.hook.PostReasoningEvent; +import io.agentscope.core.hook.PostSummaryEvent; +import io.agentscope.core.hook.PreActingEvent; +import io.agentscope.core.hook.PreCallEvent; +import io.agentscope.core.hook.PreReasoningEvent; +import io.agentscope.core.hook.PreSummaryEvent; +import io.agentscope.core.hook.ReasoningChunkEvent; +import io.agentscope.core.hook.SummaryChunkEvent; +import io.agentscope.core.message.Msg; +import io.agentscope.core.message.MsgRole; +import io.agentscope.core.message.TextBlock; +import io.agentscope.core.message.ToolResultBlock; +import io.agentscope.core.message.ToolUseBlock; +import io.agentscope.core.model.GenerateOptions; +import io.agentscope.core.tool.Toolkit; +import io.agentscope.core.util.JsonUtils; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanContext; +import io.opentelemetry.api.trace.TraceFlags; +import io.opentelemetry.api.trace.TraceState; +import io.opentelemetry.context.Scope; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +class JsonlTraceExporterTest { + + @TempDir Path tempDir; + + @Test + void writesJsonlLinesForEnabledEvents() throws Exception { + Path output = tempDir.resolve("trace.jsonl"); + + TestAgent agent = new TestAgent("agent-1", "TestAgent"); + Toolkit toolkit = new Toolkit(); + + ToolUseBlock toolUse = + ToolUseBlock.builder() + .id("tool-1") + .name("search") + .input(Map.of("q", "hello")) + .build(); + ToolResultBlock toolChunk = ToolResultBlock.text("chunk"); + ToolResultBlock toolResult = ToolResultBlock.text("done"); + + GenerateOptions options = GenerateOptions.builder().build(); + + Msg userMsg = textMsg(MsgRole.USER, "hello"); + Msg assistantMsg = textMsg(MsgRole.ASSISTANT, "world"); + + Msg reasoningChunk = textMsg(MsgRole.ASSISTANT, "a"); + Msg reasoningAccum = textMsg(MsgRole.ASSISTANT, "ab"); + + Msg summaryChunk = textMsg(MsgRole.ASSISTANT, "s1"); + Msg summaryAccum = textMsg(MsgRole.ASSISTANT, "s12"); + + try (JsonlTraceExporter exporter = + JsonlTraceExporter.builder(output) + .append(false) + .flushEveryLine(true) + .includeReasoningChunks(true) + .includeActingChunks(true) + .includeSummary(true) + .includeSummaryChunks(true) + .build()) { + exporter.onEvent(new PreCallEvent(agent, List.of(userMsg))).block(); + exporter.onEvent(new PreReasoningEvent(agent, "mock-model", options, List.of(userMsg))) + .block(); + exporter.onEvent( + new ReasoningChunkEvent( + agent, "mock-model", options, reasoningChunk, reasoningAccum)) + .block(); + exporter.onEvent(new PostReasoningEvent(agent, "mock-model", options, assistantMsg)) + .block(); + exporter.onEvent(new PreActingEvent(agent, toolkit, toolUse)).block(); + exporter.onEvent(new ActingChunkEvent(agent, toolkit, toolUse, toolChunk)).block(); + exporter.onEvent(new PostActingEvent(agent, toolkit, toolUse, toolResult)).block(); + exporter.onEvent( + new PreSummaryEvent( + agent, + "mock-model", + options, + List.of(userMsg, assistantMsg), + 10, + 10)) + .block(); + exporter.onEvent( + new SummaryChunkEvent( + agent, "mock-model", options, summaryChunk, summaryAccum)) + .block(); + exporter.onEvent(new PostSummaryEvent(agent, "mock-model", options, assistantMsg)) + .block(); + exporter.onEvent(new ErrorEvent(agent, new IllegalStateException("boom"))).block(); + exporter.onEvent(new PostCallEvent(agent, assistantMsg)).block(); + } + + List> records = readAll(output); + assertTrue(records.size() >= 11); + + Map preCall = records.get(0); + assertEquals("PRE_CALL", preCall.get("event_type")); + assertEquals("agent-1", preCall.get("agent_id")); + assertEquals("TestAgent", preCall.get("agent_name")); + assertNotNull(preCall.get("run_id")); + assertEquals(0, ((Number) preCall.get("step_id")).intValue()); + + assertTrue(containsKeyValue(records, "event_type", "PRE_REASONING")); + assertTrue(containsKeyValue(records, "event_type", "POST_REASONING")); + assertTrue(containsKeyValue(records, "event_type", "REASONING_CHUNK")); + assertTrue(containsKeyValue(records, "event_type", "PRE_ACTING")); + assertTrue(containsKeyValue(records, "event_type", "ACTING_CHUNK")); + assertTrue(containsKeyValue(records, "event_type", "POST_ACTING")); + assertTrue(containsKeyValue(records, "event_type", "PRE_SUMMARY")); + assertTrue(containsKeyValue(records, "event_type", "SUMMARY_CHUNK")); + assertTrue(containsKeyValue(records, "event_type", "POST_SUMMARY")); + assertTrue(containsKeyValue(records, "event_type", "ERROR")); + assertTrue(containsKeyValue(records, "event_type", "POST_CALL")); + + Map actingChunkRecord = findByEventType(records, "ACTING_CHUNK"); + assertNotNull(actingChunkRecord.get("incremental_chunk")); + assertNotNull(actingChunkRecord.get("chunk")); + assertEquals( + JsonUtils.getJsonCodec().toJson(actingChunkRecord.get("incremental_chunk")), + JsonUtils.getJsonCodec().toJson(actingChunkRecord.get("chunk"))); + } + + @Test + void filtersDisabledEvents() throws Exception { + Path output = tempDir.resolve("filtered.jsonl"); + TestAgent agent = new TestAgent("agent-1", "TestAgent"); + + try (JsonlTraceExporter exporter = + JsonlTraceExporter.builder(output) + .append(false) + .flushEveryLine(true) + .enabledEvents(Set.of(PRE_CALL)) + .build()) { + exporter.onEvent(new PreCallEvent(agent, List.of(textMsg(MsgRole.USER, "hi")))).block(); + exporter.onEvent(new PostCallEvent(agent, textMsg(MsgRole.ASSISTANT, "bye"))).block(); + } + + List> records = readAll(output); + assertEquals(1, records.size()); + assertEquals("PRE_CALL", records.get(0).get("event_type")); + } + + @Test + void runIdChangesAcrossTurns() throws Exception { + Path output = tempDir.resolve("runs.jsonl"); + TestAgent agent = new TestAgent("agent-1", "TestAgent"); + + try (JsonlTraceExporter exporter = + JsonlTraceExporter.builder(output).append(false).flushEveryLine(true).build()) { + exporter.onEvent(new PreCallEvent(agent, List.of(textMsg(MsgRole.USER, "t1")))).block(); + exporter.onEvent(new PostCallEvent(agent, textMsg(MsgRole.ASSISTANT, "o1"))).block(); + exporter.onEvent(new PreCallEvent(agent, List.of(textMsg(MsgRole.USER, "t2")))).block(); + } + + List> records = readAll(output); + String run1 = (String) records.get(0).get("run_id"); + String run2 = (String) records.get(2).get("run_id"); + assertNotNull(run1); + assertNotNull(run2); + assertNotEquals(run1, run2); + assertEquals(1, ((Number) records.get(0).get("turn_id")).intValue()); + assertEquals(2, ((Number) records.get(2).get("turn_id")).intValue()); + } + + @Test + void failFastControlsErrorPropagation() throws Exception { + Path output = tempDir.resolve("failfast.jsonl"); + TestAgent agent = new TestAgent("agent-1", "TestAgent"); + + JsonlTraceExporter bestEffort = + JsonlTraceExporter.builder(output) + .append(false) + .flushEveryLine(true) + .failFast(false) + .build(); + bestEffort.close(); + bestEffort.onEvent(new PreCallEvent(agent, List.of(textMsg(MsgRole.USER, "hi")))).block(); + + JsonlTraceExporter failFast = + JsonlTraceExporter.builder(output) + .append(true) + .flushEveryLine(true) + .failFast(true) + .build(); + failFast.close(); + assertThrows( + RuntimeException.class, + () -> + failFast.onEvent( + new PreCallEvent( + agent, List.of(textMsg(MsgRole.USER, "hi")))) + .block()); + } + + @Test + void rejectsNullEvents() throws Exception { + Path output = tempDir.resolve("null.jsonl"); + try (JsonlTraceExporter exporter = + JsonlTraceExporter.builder(output).append(false).flushEveryLine(true).build()) { + assertThrows(NullPointerException.class, () -> exporter.onEvent(null)); + } + } + + @Test + void serializesConcurrentExportsPerAgent() throws Exception { + Path output = tempDir.resolve("concurrent.jsonl"); + TestAgent agent = new TestAgent("agent-1", "TestAgent"); + Msg assistantMsg = textMsg(MsgRole.ASSISTANT, "world"); + GenerateOptions options = GenerateOptions.builder().build(); + int eventCount = 200; + + ExecutorService executor = Executors.newFixedThreadPool(32); + CountDownLatch start = new CountDownLatch(1); + CountDownLatch done = new CountDownLatch(eventCount); + ConcurrentLinkedQueue failures = new ConcurrentLinkedQueue<>(); + + try (JsonlTraceExporter exporter = + JsonlTraceExporter.builder(output).append(false).flushEveryLine(true).build()) { + exporter.onEvent(new PreCallEvent(agent, List.of(textMsg(MsgRole.USER, "hi")))).block(); + + for (int i = 0; i < eventCount; i++) { + executor.submit( + () -> { + try { + assertTrue(start.await(30, TimeUnit.SECONDS)); + exporter.onEvent( + new PostReasoningEvent( + agent, "mock-model", options, assistantMsg)) + .block(); + } catch (Throwable t) { + failures.add(t); + } finally { + done.countDown(); + } + }); + } + + start.countDown(); + assertTrue(done.await(30, TimeUnit.SECONDS)); + assertTrue(failures.isEmpty(), () -> "Unexpected failures: " + failures); + } finally { + executor.shutdownNow(); + } + + List> records = readAll(output); + assertEquals(eventCount + 1, records.size()); + + Set stepIds = new HashSet<>(); + Set runIds = new HashSet<>(); + for (Map record : records) { + runIds.add((String) record.get("run_id")); + stepIds.add(((Number) record.get("step_id")).longValue()); + } + + assertEquals(1, runIds.size()); + assertEquals(eventCount + 1, stepIds.size()); + for (long expected = 0; expected <= eventCount; expected++) { + long expectedStepId = expected; + assertTrue(stepIds.contains(expectedStepId), () -> "Missing step_id " + expectedStepId); + } + } + + @Test + void closeWaitsForSubscribedWrites() throws Exception { + Path output = tempDir.resolve("close-drain.jsonl"); + int eventCount = 100; + List> subscriptions = new ArrayList<>(); + + JsonlTraceExporter exporter = + JsonlTraceExporter.builder(output).append(false).flushEveryLine(false).build(); + for (int i = 0; i < eventCount; i++) { + TestAgent agent = new TestAgent("agent-" + i, "Agent-" + i); + subscriptions.add( + exporter.onEvent( + new PreCallEvent(agent, List.of(textMsg(MsgRole.USER, "hi-" + i))))); + } + + subscriptions.forEach(Mono::subscribe); + exporter.close(); + + List> records = readAll(output); + assertEquals(eventCount, records.size()); + } + + @Test + void exportsOpenTelemetryIdsWhenAvailable() throws Exception { + Path output = tempDir.resolve("otel.jsonl"); + TestAgent agent = new TestAgent("agent-1", "TestAgent"); + + SpanContext spanContext = + SpanContext.create( + "0123456789abcdef0123456789abcdef", + "89abcdef01234567", + TraceFlags.getSampled(), + TraceState.getDefault()); + + try (Scope ignored = Span.wrap(spanContext).makeCurrent(); + JsonlTraceExporter exporter = + JsonlTraceExporter.builder(output) + .append(false) + .flushEveryLine(true) + .build()) { + exporter.onEvent(new PreCallEvent(agent, List.of(textMsg(MsgRole.USER, "hi")))).block(); + } + + Map record = readAll(output).get(0); + assertEquals("0123456789abcdef0123456789abcdef", record.get("trace_id")); + assertEquals("89abcdef01234567", record.get("span_id")); + } + + private static Msg textMsg(MsgRole role, String text) { + return Msg.builder().role(role).content(TextBlock.builder().text(text).build()).build(); + } + + private static List> readAll(Path output) throws IOException { + List lines = Files.readAllLines(output); + return lines.stream() + .filter(l -> l != null && !l.isBlank()) + .map( + l -> + JsonUtils.getJsonCodec() + .fromJson(l, new TypeReference>() {})) + .toList(); + } + + private static boolean containsKeyValue( + List> records, String key, String value) { + for (Map record : records) { + Object actual = record.get(key); + if (value.equals(actual)) { + return true; + } + } + return false; + } + + private static Map findByEventType( + List> records, String eventType) { + return records.stream() + .filter(record -> eventType.equals(record.get("event_type"))) + .findFirst() + .orElseThrow(() -> new AssertionError("Missing event type " + eventType)); + } + + private static final class TestAgent implements Agent { + private final String agentId; + private final String name; + + private TestAgent(String agentId, String name) { + this.agentId = agentId; + this.name = name; + } + + @Override + public String getAgentId() { + return agentId; + } + + @Override + public String getName() { + return name; + } + + @Override + public void interrupt() {} + + @Override + public void interrupt(Msg msg) {} + + @Override + public Mono call(Msg msg) { + return Mono.error(new UnsupportedOperationException("not used")); + } + + @Override + public Mono call(List msgs) { + return Mono.error(new UnsupportedOperationException("not used")); + } + + @Override + public Mono call(List msgs, Class structuredOutputClass) { + return Mono.error(new UnsupportedOperationException("not used")); + } + + @Override + public Mono call(List msgs, JsonNode schema) { + return Mono.error(new UnsupportedOperationException("not used")); + } + + @Override + public Flux stream(Msg msg, StreamOptions options) { + return Flux.error(new UnsupportedOperationException("not used")); + } + + @Override + public Flux stream(List msgs, StreamOptions options) { + return Flux.error(new UnsupportedOperationException("not used")); + } + + @Override + public Flux stream(Msg msg, StreamOptions options, Class structuredModel) { + return Flux.error(new UnsupportedOperationException("not used")); + } + + @Override + public Flux stream(List msgs, StreamOptions options, Class structuredModel) { + return Flux.error(new UnsupportedOperationException("not used")); + } + + @Override + public Flux stream(List msgs, StreamOptions options, JsonNode schema) { + return Flux.error(new UnsupportedOperationException("not used")); + } + + @Override + public Mono observe(Msg msg) { + return Mono.error(new UnsupportedOperationException("not used")); + } + + @Override + public Mono observe(List msgs) { + return Mono.error(new UnsupportedOperationException("not used")); + } + } +} diff --git a/agentscope-examples/quickstart/src/main/java/io/agentscope/examples/quickstart/HookExample.java b/agentscope-examples/quickstart/src/main/java/io/agentscope/examples/quickstart/HookExample.java index ef29286e6..c18548b3c 100644 --- a/agentscope-examples/quickstart/src/main/java/io/agentscope/examples/quickstart/HookExample.java +++ b/agentscope-examples/quickstart/src/main/java/io/agentscope/examples/quickstart/HookExample.java @@ -25,6 +25,7 @@ import io.agentscope.core.hook.PreActingEvent; import io.agentscope.core.hook.PreCallEvent; import io.agentscope.core.hook.ReasoningChunkEvent; +import io.agentscope.core.hook.recorder.JsonlTraceExporter; import io.agentscope.core.memory.InMemoryMemory; import io.agentscope.core.message.Msg; import io.agentscope.core.message.ToolResultBlock; @@ -34,6 +35,7 @@ import io.agentscope.core.tool.ToolParam; import io.agentscope.core.tool.Toolkit; import io.agentscope.examples.quickstart.util.MsgUtils; +import java.nio.file.Path; import java.util.List; import reactor.core.publisher.Mono; @@ -56,6 +58,13 @@ public static void main(String[] args) throws Exception { // Create monitoring hook Hook monitoringHook = new MonitoringHook(); + // Built-in JSONL trace exporter (local file, easy to attach to issues) + JsonlTraceExporter jsonlTrace = + JsonlTraceExporter.builder(Path.of("logs", "agentscope-trace.jsonl")) + .includeReasoningChunks(true) + .includeActingChunks(true) + .build(); + // Create toolkit with a tool that emits progress Toolkit toolkit = new Toolkit(); toolkit.registerTool(new ProgressTools()); @@ -63,30 +72,32 @@ public static void main(String[] args) throws Exception { System.out.println("Registered tools:"); System.out.println(" - process_data: Simulate data processing with progress updates\n"); - // Create Agent with hook - ReActAgent agent = - ReActAgent.builder() - .name("HookAgent") - .sysPrompt( - "You are a helpful assistant. When processing data, use the" - + " process_data tool.") - .model( - DashScopeChatModel.builder() - .apiKey(apiKey) - .modelName("qwen-plus") - .stream(true) - .enableThinking(true) - .formatter(new DashScopeChatFormatter()) - .build()) - .toolkit(toolkit) - .memory(new InMemoryMemory()) - .hooks(List.of(monitoringHook)) - .build(); - - System.out.println("Try asking: 'Process the customer dataset'\n"); - - // Start interactive chat - ExampleUtils.startChat(agent); + try (jsonlTrace) { + // Create Agent with hooks + ReActAgent agent = + ReActAgent.builder() + .name("HookAgent") + .sysPrompt( + "You are a helpful assistant. When processing data, use the" + + " process_data tool.") + .model( + DashScopeChatModel.builder() + .apiKey(apiKey) + .modelName("qwen-plus") + .stream(true) + .enableThinking(true) + .formatter(new DashScopeChatFormatter()) + .build()) + .toolkit(toolkit) + .memory(new InMemoryMemory()) + .hooks(List.of(monitoringHook, jsonlTrace)) + .build(); + + System.out.println("Try asking: 'Process the customer dataset'\n"); + + // Start interactive chat + ExampleUtils.startChat(agent); + } } /** diff --git a/docs/en/task/hook.md b/docs/en/task/hook.md index 45fabbfd8..f0e20b3c1 100644 --- a/docs/en/task/hook.md +++ b/docs/en/task/hook.md @@ -130,6 +130,44 @@ ReActAgent agent = ReActAgent.builder() Hooks are immutable after agent construction. +## Built-in JSONL Trace Exporter + +For local debugging and offline troubleshooting, AgentScope Java provides a built-in JSONL exporter: + +> Warning: the JSONL trace exporter writes full prompts, messages, tool inputs, and error stack +> traces to local files. These records may contain sensitive user data, credentials, or other +> secrets, so only enable it in trusted environments and handle the output file as sensitive data. + +```java +import io.agentscope.core.ReActAgent; +import io.agentscope.core.hook.recorder.JsonlTraceExporter; +import io.agentscope.core.model.Model; +import io.agentscope.core.tool.Toolkit; +import java.io.IOException; +import java.nio.file.Path; +import java.util.List; + +public class JsonlTraceExample { + + public void run(Model model, Toolkit toolkit) throws IOException { + try (JsonlTraceExporter exporter = + JsonlTraceExporter.builder(Path.of("logs", "agentscope-trace.jsonl")) + .includeReasoningChunks(true) // optional + .includeActingChunks(true) // optional + .build()) { + ReActAgent agent = ReActAgent.builder() + .name("Assistant") + .model(model) + .toolkit(toolkit) + .hooks(List.of(exporter)) + .build(); + + // Use the agent while the exporter is still open + } + } +} +``` + ## Hook Examples ### Monitoring Tool Execution diff --git a/docs/zh/task/hook.md b/docs/zh/task/hook.md index b5903295c..e85657c7d 100644 --- a/docs/zh/task/hook.md +++ b/docs/zh/task/hook.md @@ -206,3 +206,42 @@ public class ErrorHandlingHook implements Hook { cd agentscope-examples/quickstart mvn exec:java -Dexec.mainClass="io.agentscope.examples.quickstart.HookExample" ``` + +## 内置 JSONL 跟踪导出器 + +为了便于本地调试和离线排障，AgentScope Java 提供了一个内置的 JSONL 导出器： + +> **警告**：JSONL 跟踪导出器会将完整的 prompt、消息、工具输入以及错误堆栈 +> 写入本地文件。这些记录可能包含敏感用户数据、凭据或其他机密信息， +> 因此只能在受信任的环境中启用，并应将输出文件按敏感数据处理。 + +```java +import io.agentscope.core.ReActAgent; +import io.agentscope.core.hook.recorder.JsonlTraceExporter; +import io.agentscope.core.model.Model; +import io.agentscope.core.tool.Toolkit; +import java.io.IOException; +import java.nio.file.Path; +import java.util.List; + +public class JsonlTraceExample { + + public void run(Model model, Toolkit toolkit) throws IOException { + try (JsonlTraceExporter exporter = + JsonlTraceExporter.builder(Path.of("logs", "agentscope-trace.jsonl")) + .includeReasoningChunks(true) // 可选 + .includeActingChunks(true) // 可选 + .build()) { + ReActAgent agent = ReActAgent.builder() + .name("Assistant") + .model(model) + .toolkit(toolkit) + .hooks(List.of(exporter)) + .build(); + + // 在 exporter 关闭前使用 agent + } + } +} +``` +