unique = new LinkedHashMap<>();
+ for (SkillSummary skill : skills) {
+ unique.putIfAbsent(skill.name(), skill);
+ }
+ return List.copyOf(unique.values());
+ }
+
+ private static String componentSearchSource(AgentComponent component) {
+ String kind = component.kind().toLowerCase(Locale.ROOT);
+ if (kind.equals("knowledgebase") || kind.equals("knowledge")) {
+ return "knowledge";
+ }
+ if (kind.equals("long_term_memory") || kind.equals("memory")) {
+ return "memory";
+ }
+ return "";
+ }
+
+ public record ToolSummary(String name, String description, String type) {
+ public ToolSummary {
+ name = Objects.requireNonNullElse(name, "");
+ description = Objects.requireNonNullElse(description, "");
+ type = Objects.requireNonNullElse(type, "");
+ }
+ }
+
+ public record SkillSummary(String name, String description) {
+ public SkillSummary {
+ if (name == null || !name.matches("[A-Za-z0-9][A-Za-z0-9_-]{0,127}")) {
+ throw new IllegalArgumentException("Invalid skill name: " + name);
+ }
+ description = Objects.requireNonNullElse(description, "");
+ }
+ }
+}
diff --git a/core/src/main/java/com/volcengine/veadk/agent/SaveSessionPolicy.java b/core/src/main/java/com/volcengine/veadk/agent/SaveSessionPolicy.java
new file mode 100644
index 0000000..40838a3
--- /dev/null
+++ b/core/src/main/java/com/volcengine/veadk/agent/SaveSessionPolicy.java
@@ -0,0 +1,64 @@
+/**
+ * Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+ *
+ * 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 com.volcengine.veadk.agent;
+
+import java.time.Duration;
+
+/** Controls when completed sessions are copied into long-term memory. */
+public record SaveSessionPolicy(
+ int minNewEvents,
+ Duration minInterval,
+ boolean saveOnSessionSwitch,
+ boolean suppressErrors) {
+
+ public static final int DEFAULT_MIN_NEW_EVENTS = 10;
+ public static final Duration DEFAULT_MIN_INTERVAL = Duration.ofSeconds(60);
+
+ public SaveSessionPolicy {
+ if (minNewEvents < 0) {
+ throw new IllegalArgumentException("minNewEvents must not be negative");
+ }
+ if (minInterval == null || minInterval.isNegative()) {
+ throw new IllegalArgumentException("minInterval must not be negative");
+ }
+ }
+
+ public static SaveSessionPolicy defaults() {
+ return new SaveSessionPolicy(
+ environmentInt("MIN_MESSAGES_THRESHOLD", DEFAULT_MIN_NEW_EVENTS),
+ Duration.ofSeconds(
+ environmentInt(
+ "MIN_TIME_THRESHOLD",
+ Math.toIntExact(DEFAULT_MIN_INTERVAL.toSeconds()))),
+ true,
+ true);
+ }
+
+ /** Policy matching the original Java callback: save every invocation without switch flushing. */
+ public static SaveSessionPolicy legacy() {
+ return new SaveSessionPolicy(0, Duration.ZERO, false, true);
+ }
+
+ private static int environmentInt(String name, int fallback) {
+ String value = System.getenv(name);
+ if (value == null || value.isBlank()) {
+ return fallback;
+ }
+ try {
+ return Integer.parseInt(value.trim());
+ } catch (NumberFormatException exception) {
+ return fallback;
+ }
+ }
+}
diff --git a/core/src/main/java/com/volcengine/veadk/agent/SaveSessionToMemoryCallback.java b/core/src/main/java/com/volcengine/veadk/agent/SaveSessionToMemoryCallback.java
index dd3e7ef..c02f459 100644
--- a/core/src/main/java/com/volcengine/veadk/agent/SaveSessionToMemoryCallback.java
+++ b/core/src/main/java/com/volcengine/veadk/agent/SaveSessionToMemoryCallback.java
@@ -18,9 +18,18 @@
import com.google.adk.agents.CallbackContext;
import com.google.adk.agents.Callbacks;
import com.google.adk.agents.InvocationContext;
+import com.google.adk.memory.BaseMemoryService;
+import com.google.adk.sessions.BaseSessionService;
+import com.google.adk.sessions.Session;
import com.google.genai.types.Content;
import com.volcengine.veadk.utils.ReadonlyContextAccessorUtil;
+import io.reactivex.rxjava3.core.Completable;
import io.reactivex.rxjava3.core.Maybe;
+import io.reactivex.rxjava3.subjects.CompletableSubject;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.function.LongSupplier;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -28,16 +37,191 @@ public class SaveSessionToMemoryCallback implements Callbacks.AfterAgentCallback
private static final Logger log = LoggerFactory.getLogger(SaveSessionToMemoryCallback.class);
+ private final SaveSessionPolicy policy;
+ private final LongSupplier currentTimeMillis;
+ private final BaseMemoryService memoryServiceOverride;
+ private final boolean legacyFireAndForget;
+ private final ConcurrentMap saveStates = new ConcurrentHashMap<>();
+ private final ConcurrentMap activeSessions = new ConcurrentHashMap<>();
+ private final ConcurrentMap saveQueues = new ConcurrentHashMap<>();
+
+ public SaveSessionToMemoryCallback() {
+ this(SaveSessionPolicy.legacy(), System::currentTimeMillis, null, true);
+ }
+
+ public SaveSessionToMemoryCallback(SaveSessionPolicy policy) {
+ this(policy, System::currentTimeMillis, null, false);
+ }
+
+ SaveSessionToMemoryCallback(SaveSessionPolicy policy, LongSupplier currentTimeMillis) {
+ this(policy, currentTimeMillis, null, false);
+ }
+
+ SaveSessionToMemoryCallback(SaveSessionPolicy policy, BaseMemoryService memoryServiceOverride) {
+ this(policy, System::currentTimeMillis, memoryServiceOverride, false);
+ }
+
+ private SaveSessionToMemoryCallback(
+ SaveSessionPolicy policy,
+ LongSupplier currentTimeMillis,
+ BaseMemoryService memoryServiceOverride,
+ boolean legacyFireAndForget) {
+ this.policy = Objects.requireNonNull(policy, "policy");
+ this.currentTimeMillis = Objects.requireNonNull(currentTimeMillis, "currentTimeMillis");
+ this.memoryServiceOverride = memoryServiceOverride;
+ this.legacyFireAndForget = legacyFireAndForget;
+ }
+
+ public SaveSessionPolicy policy() {
+ return policy;
+ }
+
@Override
public Maybe call(CallbackContext callbackContext) {
- InvocationContext invocationContext =
+ if (legacyFireAndForget) {
+ return legacyCall(callbackContext);
+ }
+ Completable save =
+ Completable.defer(
+ () ->
+ save(
+ ReadonlyContextAccessorUtil.getInvocationContext(
+ callbackContext)));
+ if (policy.suppressErrors()) {
+ save =
+ save.onErrorComplete(
+ error -> {
+ log.error("Failed to save session", error);
+ return true;
+ });
+ }
+ return save.andThen(Maybe.empty());
+ }
+
+ private Maybe legacyCall(CallbackContext callbackContext) {
+ InvocationContext context =
ReadonlyContextAccessorUtil.getInvocationContext(callbackContext);
- invocationContext
- .memoryService()
- .addSessionToMemory(invocationContext.session())
+ memoryService(context)
+ .addSessionToMemory(context.session())
.subscribe(
- () -> log.info("Saved session {}", invocationContext.session().id()),
- err -> log.error("Failed to save session", err));
+ () -> log.info("Saved session {}", context.session().id()),
+ error -> log.error("Failed to save session", error));
return Maybe.empty();
}
+
+ private Completable save(InvocationContext context) {
+ UserKey userKey = new UserKey(context.appName(), context.userId());
+ return Completable.defer(
+ () -> {
+ CompletableSubject completion = CompletableSubject.create();
+ CompletableSubject previous = saveQueues.put(userKey, completion);
+ Completable turn = previous == null ? Completable.complete() : previous;
+ return turn.andThen(saveOne(context, userKey))
+ .doFinally(
+ () -> {
+ completion.onComplete();
+ saveQueues.remove(userKey, completion);
+ });
+ });
+ }
+
+ private Completable saveOne(InvocationContext context, UserKey userKey) {
+ return Completable.defer(
+ () -> {
+ Session currentSession = context.session();
+ long now = currentTimeMillis.getAsLong();
+ String previousSessionId = activeSessions.get(userKey);
+ Completable previousSave =
+ savePreviousSessionIfSwitched(
+ context, previousSessionId, currentSession.id(), now);
+ return previousSave.andThen(
+ Completable.defer(
+ () -> {
+ activeSessions.put(userKey, currentSession.id());
+ return saveCurrentSession(context, now);
+ }));
+ });
+ }
+
+ private Completable savePreviousSessionIfSwitched(
+ InvocationContext context,
+ String previousSessionId,
+ String currentSessionId,
+ long now) {
+ if (!policy.saveOnSessionSwitch()
+ || previousSessionId == null
+ || previousSessionId.equals(currentSessionId)) {
+ return Completable.complete();
+ }
+ BaseSessionService sessions = context.sessionService();
+ BaseMemoryService memory = memoryService(context);
+ return sessions.getSession(
+ context.appName(),
+ context.userId(),
+ previousSessionId,
+ java.util.Optional.empty())
+ .flatMapCompletable(
+ session ->
+ memory.addSessionToMemory(session)
+ .doOnComplete(
+ () -> {
+ saveStates.put(
+ new SessionKey(
+ context.appName(),
+ context.userId(),
+ previousSessionId),
+ new SaveState(
+ now, session.events().size()));
+ log.info(
+ "Saved previous session {} after"
+ + " session switch",
+ previousSessionId);
+ }));
+ }
+
+ private Completable saveCurrentSession(InvocationContext context, long now) {
+ String sessionId = context.session().id();
+ SessionKey key = new SessionKey(context.appName(), context.userId(), sessionId);
+ return context.sessionService()
+ .getSession(
+ context.appName(), context.userId(), sessionId, java.util.Optional.empty())
+ .switchIfEmpty(
+ Maybe.error(
+ new IllegalStateException(
+ "Session not found in session service: " + sessionId)))
+ .flatMapCompletable(
+ session -> {
+ int eventCount = session.events().size();
+ SaveState previous = saveStates.get(key);
+ if (shouldSkip(previous, eventCount, now)) {
+ return Completable.complete();
+ }
+ return memoryService(context)
+ .addSessionToMemory(session)
+ .doOnComplete(
+ () -> {
+ saveStates.put(key, new SaveState(now, eventCount));
+ log.info("Saved session {}", sessionId);
+ });
+ });
+ }
+
+ private BaseMemoryService memoryService(InvocationContext context) {
+ return memoryServiceOverride == null ? context.memoryService() : memoryServiceOverride;
+ }
+
+ private boolean shouldSkip(SaveState previous, int currentEventCount, long now) {
+ if (previous == null) {
+ return false;
+ }
+ long elapsed = now - previous.savedAtMillis();
+ int newEvents = currentEventCount - previous.eventCount();
+ return elapsed < policy.minInterval().toMillis() && newEvents < policy.minNewEvents();
+ }
+
+ private record UserKey(String appName, String userId) {}
+
+ private record SessionKey(String appName, String userId, String sessionId) {}
+
+ private record SaveState(long savedAtMillis, int eventCount) {}
}
diff --git a/core/src/main/java/com/volcengine/veadk/config/ConfigLoader.java b/core/src/main/java/com/volcengine/veadk/config/ConfigLoader.java
new file mode 100644
index 0000000..5a599fc
--- /dev/null
+++ b/core/src/main/java/com/volcengine/veadk/config/ConfigLoader.java
@@ -0,0 +1,159 @@
+/**
+ * Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+ *
+ * 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 com.volcengine.veadk.config;
+
+import java.io.IOException;
+import java.io.Reader;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import org.yaml.snakeyaml.LoaderOptions;
+import org.yaml.snakeyaml.Yaml;
+import org.yaml.snakeyaml.constructor.SafeConstructor;
+
+final class ConfigLoader {
+
+ private ConfigLoader() {}
+
+ static Map load(Path workingDirectory, Map environment) {
+ Path directory = workingDirectory.toAbsolutePath().normalize();
+ Map values = new LinkedHashMap<>();
+
+ findUpwards(directory, "config.yaml").ifPresent(path -> values.putAll(loadYaml(path)));
+ Path dotenv = directory.resolve(".env");
+ if (Files.isRegularFile(dotenv)) {
+ values.putAll(loadDotenv(dotenv));
+ }
+ environment.forEach((key, value) -> values.put(normalize(key), value));
+ applyProviderAliases(values);
+ return values;
+ }
+
+ private static java.util.Optional findUpwards(Path directory, String filename) {
+ Path current = directory;
+ while (current != null) {
+ Path candidate = current.resolve(filename);
+ if (Files.isRegularFile(candidate)) {
+ return java.util.Optional.of(candidate);
+ }
+ current = current.getParent();
+ }
+ return java.util.Optional.empty();
+ }
+
+ private static Map loadYaml(Path path) {
+ LoaderOptions options = new LoaderOptions();
+ options.setAllowDuplicateKeys(false);
+ Yaml yaml = new Yaml(new SafeConstructor(options));
+ try (Reader reader = Files.newBufferedReader(path, StandardCharsets.UTF_8)) {
+ Object root = yaml.load(reader);
+ Map flattened = new LinkedHashMap<>();
+ if (root instanceof Map map) {
+ flatten(map, "", flattened);
+ }
+ return flattened;
+ } catch (IOException | RuntimeException exception) {
+ throw new IllegalArgumentException("Unable to load VeADK config: " + path, exception);
+ }
+ }
+
+ private static void flatten(Map source, String prefix, Map target) {
+ source.forEach(
+ (rawKey, value) -> {
+ String key = String.valueOf(rawKey);
+ String path = prefix.isEmpty() ? key : prefix + "_" + key;
+ if (value instanceof Map nested) {
+ flatten(nested, path, target);
+ } else {
+ target.put(normalize(path), stringify(value));
+ }
+ });
+ }
+
+ private static String stringify(Object value) {
+ if (value == null) {
+ return "";
+ }
+ if (value instanceof List list) {
+ return list.stream()
+ .map(String::valueOf)
+ .reduce((left, right) -> left + "," + right)
+ .orElse("");
+ }
+ return String.valueOf(value);
+ }
+
+ private static Map loadDotenv(Path path) {
+ Map values = new LinkedHashMap<>();
+ try {
+ for (String rawLine : Files.readAllLines(path, StandardCharsets.UTF_8)) {
+ String line = rawLine.trim();
+ if (line.isEmpty() || line.startsWith("#")) {
+ continue;
+ }
+ if (line.startsWith("export ")) {
+ line = line.substring("export ".length()).trim();
+ }
+ int separator = line.indexOf('=');
+ if (separator <= 0) {
+ continue;
+ }
+ String key = normalize(line.substring(0, separator).trim());
+ String value = unquote(line.substring(separator + 1).trim());
+ values.put(key, value);
+ }
+ return values;
+ } catch (IOException exception) {
+ throw new IllegalArgumentException("Unable to load VeADK dotenv: " + path, exception);
+ }
+ }
+
+ private static String unquote(String value) {
+ if (value.length() >= 2) {
+ char first = value.charAt(0);
+ char last = value.charAt(value.length() - 1);
+ if ((first == '\'' && last == '\'') || (first == '"' && last == '"')) {
+ return value.substring(1, value.length() - 1);
+ }
+ }
+ return value;
+ }
+
+ private static void applyProviderAliases(Map values) {
+ if (!"byteplus".equalsIgnoreCase(values.getOrDefault("CLOUD_PROVIDER", ""))) {
+ return;
+ }
+ copyIfAbsent(values, "BYTEPLUS_ACCESS_KEY", "VOLCENGINE_ACCESS_KEY");
+ copyIfAbsent(values, "BYTEPLUS_SECRET_KEY", "VOLCENGINE_SECRET_KEY");
+ }
+
+ private static void copyIfAbsent(
+ Map values, String source, String destination) {
+ String existing = values.get(destination);
+ String alias = values.get(source);
+ if ((existing == null || existing.isBlank()) && alias != null && !alias.isBlank()) {
+ values.put(destination, alias);
+ }
+ }
+
+ static String normalize(String key) {
+ return key.trim().toUpperCase(Locale.ROOT).replace('.', '_').replace('-', '_');
+ }
+}
diff --git a/core/src/main/java/com/volcengine/veadk/config/VeADKConfig.java b/core/src/main/java/com/volcengine/veadk/config/VeADKConfig.java
new file mode 100644
index 0000000..6ec4ca6
--- /dev/null
+++ b/core/src/main/java/com/volcengine/veadk/config/VeADKConfig.java
@@ -0,0 +1,207 @@
+/**
+ * Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+ *
+ * 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 com.volcengine.veadk.config;
+
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+/** Loads VeADK settings from {@code config.yaml}, {@code .env}, and process environment. */
+public final class VeADKConfig {
+
+ public static final String DEFAULT_MODEL_NAME = "doubao-seed-2-1-pro-260628";
+ public static final String DEFAULT_MODEL_PROVIDER = "openai";
+ public static final String DEFAULT_MODEL_API_BASE = "https://ark.cn-beijing.volces.com/api/v3/";
+ public static final String DEFAULT_EMBEDDING_MODEL = "doubao-embedding-vision-250615";
+ public static final int DEFAULT_EMBEDDING_DIMENSION = 2048;
+
+ private static final Set SECRET_MARKERS =
+ Set.of("API_KEY", "SECRET", "PASSWORD", "TOKEN", "CREDENTIAL");
+
+ private final Map values;
+
+ private VeADKConfig(Map values) {
+ this.values = Collections.unmodifiableMap(new LinkedHashMap<>(values));
+ }
+
+ public static VeADKConfig load() {
+ return load(Path.of(""));
+ }
+
+ public static VeADKConfig load(Path workingDirectory) {
+ return load(workingDirectory, System.getenv());
+ }
+
+ /** Loads with an explicit environment map, useful for embedded runtimes and deterministic tests. */
+ public static VeADKConfig load(Path workingDirectory, Map environment) {
+ return new VeADKConfig(ConfigLoader.load(workingDirectory, environment));
+ }
+
+ public static VeADKConfig from(Map values) {
+ Map normalized = new LinkedHashMap<>();
+ values.forEach((key, value) -> normalized.put(ConfigLoader.normalize(key), value));
+ return new VeADKConfig(normalized);
+ }
+
+ public Optional get(String key) {
+ return Optional.ofNullable(values.get(ConfigLoader.normalize(key)));
+ }
+
+ public String get(String key, String defaultValue) {
+ String value = values.get(ConfigLoader.normalize(key));
+ return value == null || value.isBlank() ? defaultValue : value;
+ }
+
+ public String require(String key) {
+ String normalized = ConfigLoader.normalize(key);
+ String value = values.get(normalized);
+ if (value == null || value.isBlank()) {
+ throw new IllegalStateException(
+ "Missing required configuration: "
+ + normalized
+ + ". Configure it in the environment, .env, or config.yaml.");
+ }
+ return value;
+ }
+
+ public boolean getBoolean(String key, boolean defaultValue) {
+ String value = values.get(ConfigLoader.normalize(key));
+ if (value == null || value.isBlank()) {
+ return defaultValue;
+ }
+ return switch (value.trim().toLowerCase(Locale.ROOT)) {
+ case "true", "1", "yes", "on", "enabled" -> true;
+ case "false", "0", "no", "off", "disabled" -> false;
+ default ->
+ throw new IllegalArgumentException(
+ "Configuration " + ConfigLoader.normalize(key) + " must be a boolean");
+ };
+ }
+
+ public int getInt(String key, int defaultValue) {
+ String value = values.get(ConfigLoader.normalize(key));
+ if (value == null || value.isBlank()) {
+ return defaultValue;
+ }
+ try {
+ return Integer.parseInt(value.trim());
+ } catch (NumberFormatException exception) {
+ throw new IllegalArgumentException(
+ "Configuration " + ConfigLoader.normalize(key) + " must be an integer",
+ exception);
+ }
+ }
+
+ public ModelConfig model() {
+ boolean byteplus = "byteplus".equalsIgnoreCase(get("CLOUD_PROVIDER", ""));
+ String defaultName = byteplus ? "seed-2-0-lite-260228" : DEFAULT_MODEL_NAME;
+ String defaultBase =
+ byteplus
+ ? "https://ark.ap-southeast.bytepluses.com/api/v3"
+ : DEFAULT_MODEL_API_BASE;
+ return new ModelConfig(
+ get("MODEL_AGENT_NAME", defaultName),
+ get("MODEL_AGENT_PROVIDER", DEFAULT_MODEL_PROVIDER),
+ get("MODEL_AGENT_API_BASE", defaultBase),
+ get("MODEL_AGENT_API_KEY", ""),
+ get("MODEL_AGENT_API_KEY_NAME", ""));
+ }
+
+ public EmbeddingConfig embedding() {
+ boolean byteplus = "byteplus".equalsIgnoreCase(get("CLOUD_PROVIDER", ""));
+ return new EmbeddingConfig(
+ get(
+ "MODEL_EMBEDDING_NAME",
+ byteplus ? "skylark-embedding-vision-250615" : DEFAULT_EMBEDDING_MODEL),
+ getInt("MODEL_EMBEDDING_DIM", DEFAULT_EMBEDDING_DIMENSION),
+ get("MODEL_EMBEDDING_API_BASE", model().apiBase()),
+ get("MODEL_EMBEDDING_API_KEY", model().apiKey()));
+ }
+
+ public OpenTelemetryConfig openTelemetry() {
+ return new OpenTelemetryConfig(
+ getBoolean("OBSERVABILITY_OPENTELEMETRY_TRACE_CONTENT", true),
+ get(
+ "OBSERVABILITY_OPENTELEMETRY_TLS_ENDPOINT",
+ "https://tls-cn-beijing.volces.com:4317"),
+ get("OBSERVABILITY_OPENTELEMETRY_TLS_REGION", "cn-beijing"),
+ get("OBSERVABILITY_OPENTELEMETRY_TLS_SERVICE_NAME", ""));
+ }
+
+ /** Returns all resolved values. Prefer {@link #redactedValues()} for diagnostics. */
+ public Map values() {
+ return values;
+ }
+
+ public Map redactedValues() {
+ Map redacted = new LinkedHashMap<>();
+ values.forEach(
+ (key, value) ->
+ redacted.put(
+ key, isSecret(key) && !value.isBlank() ? "" : value));
+ return Collections.unmodifiableMap(redacted);
+ }
+
+ private static boolean isSecret(String key) {
+ String upper = key.toUpperCase(Locale.ROOT);
+ return SECRET_MARKERS.stream().anyMatch(upper::contains);
+ }
+
+ @Override
+ public String toString() {
+ return "VeADKConfig" + redactedValues();
+ }
+
+ public record ModelConfig(
+ String name, String provider, String apiBase, String apiKey, String apiKeyName) {
+ @Override
+ public String toString() {
+ return "ModelConfig[name="
+ + name
+ + ", provider="
+ + provider
+ + ", apiBase="
+ + apiBase
+ + ", apiKey="
+ + (apiKey.isBlank() ? "" : "")
+ + ", apiKeyName="
+ + apiKeyName
+ + "]";
+ }
+ }
+
+ public record EmbeddingConfig(String name, int dimension, String apiBase, String apiKey) {
+ @Override
+ public String toString() {
+ return "EmbeddingConfig[name="
+ + name
+ + ", dimension="
+ + dimension
+ + ", apiBase="
+ + apiBase
+ + ", apiKey="
+ + (apiKey.isBlank() ? "" : "")
+ + "]";
+ }
+ }
+
+ public record OpenTelemetryConfig(
+ boolean traceContent, String tlsEndpoint, String tlsRegion, String tlsServiceName) {}
+}
diff --git a/core/src/main/java/com/volcengine/veadk/knowledgebase/InMemoryKnowledgebaseBackend.java b/core/src/main/java/com/volcengine/veadk/knowledgebase/InMemoryKnowledgebaseBackend.java
new file mode 100644
index 0000000..bdd4780
--- /dev/null
+++ b/core/src/main/java/com/volcengine/veadk/knowledgebase/InMemoryKnowledgebaseBackend.java
@@ -0,0 +1,84 @@
+/**
+ * Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+ *
+ * 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 com.volcengine.veadk.knowledgebase;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+/** Deterministic in-memory backend suitable for local development and tests. */
+public final class InMemoryKnowledgebaseBackend implements KnowledgebaseBackend {
+
+ private final String index;
+ private final CopyOnWriteArrayList entries = new CopyOnWriteArrayList<>();
+
+ public InMemoryKnowledgebaseBackend(String index) {
+ if (index == null || index.isBlank()) {
+ throw new IllegalArgumentException("index must not be blank");
+ }
+ this.index = index;
+ }
+
+ @Override
+ public String index() {
+ return index;
+ }
+
+ @Override
+ public boolean add(List entries) {
+ this.entries.addAll(List.copyOf(entries));
+ return true;
+ }
+
+ @Override
+ public List search(String query, int topK) {
+ if (topK <= 0) {
+ throw new IllegalArgumentException("topK must be positive");
+ }
+ Set queryTerms = terms(query);
+ List ranked = new ArrayList<>(entries.size());
+ for (int index = 0; index < entries.size(); index++) {
+ KnowledgebaseEntry entry = entries.get(index);
+ Set contentTerms = terms(entry.content());
+ long overlap = queryTerms.stream().filter(contentTerms::contains).count();
+ ranked.add(new RankedEntry(entry, overlap, index));
+ }
+ ranked.sort(
+ Comparator.comparingLong(RankedEntry::score)
+ .reversed()
+ .thenComparingInt(RankedEntry::insertionOrder));
+ return ranked.stream().limit(topK).map(RankedEntry::entry).toList();
+ }
+
+ private static Set terms(String text) {
+ Set terms = new HashSet<>();
+ if (text == null) {
+ return terms;
+ }
+ for (String term : text.toLowerCase(Locale.ROOT).split("[^\\p{L}\\p{N}_]+")) {
+ if (!term.isBlank()) {
+ terms.add(term);
+ }
+ }
+ return terms;
+ }
+
+ private record RankedEntry(KnowledgebaseEntry entry, long score, int insertionOrder) {}
+}
diff --git a/core/src/main/java/com/volcengine/veadk/knowledgebase/KnowledgeBase.java b/core/src/main/java/com/volcengine/veadk/knowledgebase/KnowledgeBase.java
new file mode 100644
index 0000000..97a0379
--- /dev/null
+++ b/core/src/main/java/com/volcengine/veadk/knowledgebase/KnowledgeBase.java
@@ -0,0 +1,157 @@
+/**
+ * Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+ *
+ * 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 com.volcengine.veadk.knowledgebase;
+
+import io.reactivex.rxjava3.core.Single;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/** Backend-neutral knowledge-base facade aligned with the Python VeADK contract. */
+public final class KnowledgeBase implements BaseKnowledgebaseService, AutoCloseable {
+
+ public static final String DEFAULT_NAME = "user_knowledgebase";
+ public static final String DEFAULT_DESCRIPTION =
+ "This knowledgebase stores some user-related information.";
+
+ private final String name;
+ private final String description;
+ private final KnowledgebaseBackend backend;
+ private final int topK;
+
+ public KnowledgeBase(String index) {
+ this(new InMemoryKnowledgebaseBackend(index), DEFAULT_NAME, DEFAULT_DESCRIPTION, 10);
+ }
+
+ public KnowledgeBase(KnowledgebaseBackend backend) {
+ this(backend, DEFAULT_NAME, DEFAULT_DESCRIPTION, 10);
+ }
+
+ public KnowledgeBase(KnowledgebaseBackend backend, String name, String description, int topK) {
+ this.backend = Objects.requireNonNull(backend, "backend");
+ this.name = requireText(name, "name");
+ this.description = Objects.requireNonNullElse(description, "");
+ if (topK <= 0) {
+ throw new IllegalArgumentException("topK must be positive");
+ }
+ this.topK = topK;
+ }
+
+ public String name() {
+ return name;
+ }
+
+ public String description() {
+ return description;
+ }
+
+ public String index() {
+ return backend.index();
+ }
+
+ public String backend() {
+ return backend.backendName();
+ }
+
+ public int topK() {
+ return topK;
+ }
+
+ public boolean addFromText(String text) {
+ return addFromText(List.of(text));
+ }
+
+ public boolean addFromText(List texts) {
+ return backend.add(texts.stream().map(KnowledgebaseEntry::new).toList());
+ }
+
+ public boolean addFromFiles(List files) {
+ List entries = new ArrayList<>();
+ for (Path file : files) {
+ try {
+ entries.add(
+ new KnowledgebaseEntry(
+ Files.readString(file, StandardCharsets.UTF_8),
+ Map.of("file_path", file.toAbsolutePath().normalize().toString())));
+ } catch (IOException exception) {
+ throw new IllegalArgumentException(
+ "Unable to read knowledge file: " + file, exception);
+ }
+ }
+ return backend.add(entries);
+ }
+
+ public boolean addFromDirectory(Path directory) {
+ if (!Files.isDirectory(directory)) {
+ throw new IllegalArgumentException("Knowledge directory does not exist: " + directory);
+ }
+ try (java.util.stream.Stream paths = Files.walk(directory)) {
+ return addFromFiles(paths.filter(Files::isRegularFile).sorted().toList());
+ } catch (IOException exception) {
+ throw new IllegalArgumentException(
+ "Unable to read knowledge directory: " + directory, exception);
+ }
+ }
+
+ public List search(String query) {
+ return search(query, topK);
+ }
+
+ public List search(String query, int topK) {
+ return List.copyOf(backend.search(Objects.requireNonNull(query, "query"), topK));
+ }
+
+ @Override
+ public Single searchKnowledgebase(String query) {
+ return Single.fromCallable(
+ () -> {
+ List
+ legacyEntries =
+ search(query).stream()
+ .map(KnowledgeBase::toLegacyEntry)
+ .toList();
+ SearchKnowledgebaseResponse response = new SearchKnowledgebaseResponse();
+ response.setKnowledgebaseEntries(legacyEntries);
+ return response;
+ });
+ }
+
+ @Override
+ public void close() {
+ backend.close();
+ }
+
+ private static com.volcengine.veadk.integration.vikingknowledgebase.KnowledgebaseEntry
+ toLegacyEntry(KnowledgebaseEntry entry) {
+ Map metadata = new LinkedHashMap<>();
+ entry.metadata().forEach((key, value) -> metadata.put(key, String.valueOf(value)));
+ return new com.volcengine.veadk.integration.vikingknowledgebase.KnowledgebaseEntry(
+ entry.content(), metadata);
+ }
+
+ private static String requireText(String value, String field) {
+ if (value == null || value.isBlank()) {
+ throw new IllegalArgumentException(field + " must not be blank");
+ }
+ return value;
+ }
+}
diff --git a/core/src/main/java/com/volcengine/veadk/knowledgebase/KnowledgebaseBackend.java b/core/src/main/java/com/volcengine/veadk/knowledgebase/KnowledgebaseBackend.java
new file mode 100644
index 0000000..44e896b
--- /dev/null
+++ b/core/src/main/java/com/volcengine/veadk/knowledgebase/KnowledgebaseBackend.java
@@ -0,0 +1,35 @@
+/**
+ * Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+ *
+ * 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 com.volcengine.veadk.knowledgebase;
+
+import java.util.List;
+
+/** Storage and retrieval contract implemented by all knowledge-base backends. */
+public interface KnowledgebaseBackend extends AutoCloseable {
+
+ String index();
+
+ boolean add(List entries);
+
+ List search(String query, int topK);
+
+ default String backendName() {
+ return getClass().getSimpleName();
+ }
+
+ @Override
+ default void close() {}
+}
diff --git a/core/src/main/java/com/volcengine/veadk/knowledgebase/KnowledgebaseEntry.java b/core/src/main/java/com/volcengine/veadk/knowledgebase/KnowledgebaseEntry.java
new file mode 100644
index 0000000..14e3e5b
--- /dev/null
+++ b/core/src/main/java/com/volcengine/veadk/knowledgebase/KnowledgebaseEntry.java
@@ -0,0 +1,33 @@
+/**
+ * Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+ *
+ * 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 com.volcengine.veadk.knowledgebase;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Objects;
+
+/** Backend-neutral knowledge-base entry. */
+public record KnowledgebaseEntry(String content, Map metadata) {
+
+ public KnowledgebaseEntry {
+ content = Objects.requireNonNull(content, "content");
+ metadata = Map.copyOf(new LinkedHashMap<>(Objects.requireNonNullElse(metadata, Map.of())));
+ }
+
+ public KnowledgebaseEntry(String content) {
+ this(content, Map.of());
+ }
+}
diff --git a/core/src/main/java/com/volcengine/veadk/knowledgebase/KnowledgebaseServiceBackendAdapter.java b/core/src/main/java/com/volcengine/veadk/knowledgebase/KnowledgebaseServiceBackendAdapter.java
new file mode 100644
index 0000000..d4d98d5
--- /dev/null
+++ b/core/src/main/java/com/volcengine/veadk/knowledgebase/KnowledgebaseServiceBackendAdapter.java
@@ -0,0 +1,63 @@
+/**
+ * Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+ *
+ * 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 com.volcengine.veadk.knowledgebase;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/** Read-only adapter that exposes an existing Java knowledge-base service as a new backend. */
+public final class KnowledgebaseServiceBackendAdapter implements KnowledgebaseBackend {
+
+ private final String index;
+ private final BaseKnowledgebaseService service;
+
+ public KnowledgebaseServiceBackendAdapter(String index, BaseKnowledgebaseService service) {
+ if (index == null || index.isBlank()) {
+ throw new IllegalArgumentException("index must not be blank");
+ }
+ this.index = index;
+ this.service = Objects.requireNonNull(service, "service");
+ }
+
+ @Override
+ public String index() {
+ return index;
+ }
+
+ @Override
+ public boolean add(List entries) {
+ throw new UnsupportedOperationException("The legacy knowledge-base service is read-only");
+ }
+
+ @Override
+ public List search(String query, int topK) {
+ SearchKnowledgebaseResponse response = service.searchKnowledgebase(query).blockingGet();
+ if (response == null || response.getKnowledgebaseEntries() == null) {
+ return List.of();
+ }
+ return response.getKnowledgebaseEntries().stream()
+ .limit(topK)
+ .map(
+ entry ->
+ new KnowledgebaseEntry(
+ entry.getContent(),
+ entry.getMetadata() == null
+ ? Map.of()
+ : Map.copyOf(entry.getMetadata())))
+ .toList();
+ }
+}
diff --git a/core/src/main/java/com/volcengine/veadk/memory/InMemoryLongTermMemoryBackend.java b/core/src/main/java/com/volcengine/veadk/memory/InMemoryLongTermMemoryBackend.java
new file mode 100644
index 0000000..f796199
--- /dev/null
+++ b/core/src/main/java/com/volcengine/veadk/memory/InMemoryLongTermMemoryBackend.java
@@ -0,0 +1,109 @@
+/**
+ * Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+ *
+ * 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 com.volcengine.veadk.memory;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+/** Deterministic local backend suitable for development and tests. */
+public final class InMemoryLongTermMemoryBackend implements LongTermMemoryBackend {
+
+ private final String index;
+ private final Map> memoriesByUser =
+ new ConcurrentHashMap<>();
+
+ public InMemoryLongTermMemoryBackend(String index) {
+ if (index == null || index.isBlank()) {
+ throw new IllegalArgumentException("index must not be blank");
+ }
+ this.index = index;
+ }
+
+ @Override
+ public String index() {
+ return index;
+ }
+
+ @Override
+ public boolean saveMemory(
+ String userId, List eventStrings, Map options) {
+ String normalizedUserId = requireText(userId, "userId");
+ List events = List.copyOf(eventStrings);
+ if (events.stream().anyMatch(event -> event == null || event.isBlank())) {
+ throw new IllegalArgumentException("eventStrings must not contain blank values");
+ }
+ memoriesByUser
+ .computeIfAbsent(normalizedUserId, ignored -> new CopyOnWriteArrayList<>())
+ .addAll(events);
+ return true;
+ }
+
+ @Override
+ public List searchMemory(
+ String userId, String query, int topK, Map options) {
+ if (topK <= 0) {
+ throw new IllegalArgumentException("topK must be positive");
+ }
+ requireText(query, "query");
+ List memories =
+ memoriesByUser.getOrDefault(
+ requireText(userId, "userId"), new CopyOnWriteArrayList<>());
+ Set queryTerms = terms(query);
+ List ranked = new ArrayList<>(memories.size());
+ for (int position = 0; position < memories.size(); position++) {
+ String memory = memories.get(position);
+ Set memoryTerms = terms(memory);
+ long overlap = queryTerms.stream().filter(memoryTerms::contains).count();
+ if (memory.toLowerCase(Locale.ROOT).contains(query.toLowerCase(Locale.ROOT))) {
+ overlap++;
+ }
+ ranked.add(new RankedMemory(memory, overlap, position));
+ }
+ ranked.sort(
+ Comparator.comparingLong(RankedMemory::score)
+ .reversed()
+ .thenComparingInt(RankedMemory::insertionOrder));
+ return ranked.stream().limit(topK).map(RankedMemory::value).toList();
+ }
+
+ public int size(String userId) {
+ return memoriesByUser.getOrDefault(userId, new CopyOnWriteArrayList<>()).size();
+ }
+
+ private static Set terms(String text) {
+ Set terms = new HashSet<>();
+ for (String term : text.toLowerCase(Locale.ROOT).split("[^\\p{L}\\p{N}_]+")) {
+ if (!term.isBlank()) {
+ terms.add(term);
+ }
+ }
+ return terms;
+ }
+
+ private static String requireText(String value, String field) {
+ if (value == null || value.isBlank()) {
+ throw new IllegalArgumentException(field + " must not be blank");
+ }
+ return value;
+ }
+
+ private record RankedMemory(String value, long score, int insertionOrder) {}
+}
diff --git a/core/src/main/java/com/volcengine/veadk/memory/InMemoryShortTermMemoryBackend.java b/core/src/main/java/com/volcengine/veadk/memory/InMemoryShortTermMemoryBackend.java
new file mode 100644
index 0000000..6189c6b
--- /dev/null
+++ b/core/src/main/java/com/volcengine/veadk/memory/InMemoryShortTermMemoryBackend.java
@@ -0,0 +1,28 @@
+/**
+ * Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+ *
+ * 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 com.volcengine.veadk.memory;
+
+import com.google.adk.sessions.BaseSessionService;
+import com.google.adk.sessions.InMemorySessionService;
+
+/** Local backend that keeps sessions in process memory. */
+public final class InMemoryShortTermMemoryBackend implements ShortTermMemoryBackend {
+
+ private final InMemorySessionService sessionService = new InMemorySessionService();
+
+ @Override
+ public BaseSessionService sessionService() {
+ return sessionService;
+ }
+}
diff --git a/core/src/main/java/com/volcengine/veadk/memory/LongTermMemory.java b/core/src/main/java/com/volcengine/veadk/memory/LongTermMemory.java
new file mode 100644
index 0000000..d4ecc74
--- /dev/null
+++ b/core/src/main/java/com/volcengine/veadk/memory/LongTermMemory.java
@@ -0,0 +1,244 @@
+/**
+ * Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+ *
+ *
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 com.volcengine.veadk.memory;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.google.adk.events.Event;
+import com.google.adk.memory.BaseMemoryService;
+import com.google.adk.memory.MemoryEntry;
+import com.google.adk.memory.SearchMemoryResponse;
+import com.google.adk.sessions.Session;
+import com.google.genai.types.Content;
+import com.google.genai.types.Part;
+import com.volcengine.veadk.utils.JSONUtil;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Single;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Backend-neutral long-term-memory facade aligned with the Python VeADK contract. */
+public final class LongTermMemory implements BaseMemoryService, AutoCloseable {
+
+ public static final String DEFAULT_INDEX = "default_app";
+ public static final int DEFAULT_TOP_K = 5;
+
+ private static final Logger log = LoggerFactory.getLogger(LongTermMemory.class);
+
+ private final LongTermMemoryBackend backend;
+ private final String appName;
+ private final int topK;
+
+ public LongTermMemory() {
+ this(DEFAULT_INDEX);
+ }
+
+ public LongTermMemory(String index) {
+ this(new InMemoryLongTermMemoryBackend(index), index, DEFAULT_TOP_K);
+ }
+
+ public LongTermMemory(LongTermMemoryBackend backend) {
+ this(backend, backend.index(), DEFAULT_TOP_K);
+ }
+
+ public LongTermMemory(LongTermMemoryBackend backend, int topK) {
+ this(backend, backend.index(), topK);
+ }
+
+ public LongTermMemory(LongTermMemoryBackend backend, String appName, int topK) {
+ this.backend = Objects.requireNonNull(backend, "backend");
+ this.appName = requireText(appName, "appName");
+ if (topK <= 0) {
+ throw new IllegalArgumentException("topK must be positive");
+ }
+ this.topK = topK;
+ }
+
+ public String index() {
+ return backend.index();
+ }
+
+ public String appName() {
+ return appName;
+ }
+
+ public String backend() {
+ return backend.backendName();
+ }
+
+ public int topK() {
+ return topK;
+ }
+
+ @Override
+ public Completable addSessionToMemory(Session session) {
+ Objects.requireNonNull(session, "session");
+ return Completable.fromAction(
+ () -> {
+ List eventStrings = serializeUserEvents(session.events());
+ if (eventStrings.isEmpty()) {
+ return;
+ }
+ Map options = new LinkedHashMap<>();
+ options.put(
+ "appName",
+ session.appName() == null || session.appName().isBlank()
+ ? appName
+ : session.appName());
+ options.put("sessionId", session.id());
+ backend.saveMemory(session.userId(), eventStrings, Map.copyOf(options));
+ });
+ }
+
+ @Override
+ public Single searchMemory(
+ String requestedAppName, String userId, String query) {
+ return Single.fromCallable(
+ () -> {
+ List chunks;
+ try {
+ chunks =
+ backend.searchMemory(
+ userId,
+ query,
+ topK,
+ Map.of(
+ "appName",
+ requestedAppName == null
+ || requestedAppName.isBlank()
+ ? appName
+ : requestedAppName));
+ } catch (Exception exception) {
+ log.warn(
+ "Long-term memory search failed; returning no memories", exception);
+ chunks = List.of();
+ }
+ List entries = new ArrayList<>();
+ chunks.forEach(chunk -> entries.addAll(toMemoryEntries(chunk)));
+ return SearchMemoryResponse.builder().setMemories(entries).build();
+ });
+ }
+
+ @Override
+ public void close() {
+ backend.close();
+ }
+
+ private static List serializeUserEvents(List events) {
+ List serialized = new ArrayList<>();
+ for (Event event : events) {
+ if (!"user".equals(event.author()) || event.content().isEmpty()) {
+ continue;
+ }
+ List parts = event.content().get().parts().orElse(List.of());
+ List> textParts =
+ parts.stream()
+ .filter(part -> part.text().isPresent())
+ .map(part -> Map.of("text", part.text().get()))
+ .toList();
+ if (textParts.isEmpty()) {
+ continue;
+ }
+ String value = JSONUtil.toJson(Map.of("role", "user", "parts", textParts));
+ if (!value.isBlank()) {
+ serialized.add(value);
+ }
+ }
+ return serialized;
+ }
+
+ private static List toMemoryEntries(String chunk) {
+ try {
+ return toMemoryEntries(JSONUtil.parseJson(chunk));
+ } catch (IOException exception) {
+ return List.of(memoryEntry("user", chunk));
+ }
+ }
+
+ private static List toMemoryEntries(JsonNode node) {
+ if (node == null || node.isNull()) {
+ return List.of();
+ }
+ if (node.isTextual() || node.isNumber() || node.isBoolean()) {
+ return List.of(memoryEntry("user", node.asText()));
+ }
+ if (node.isArray()) {
+ List entries = new ArrayList<>();
+ node.forEach(item -> entries.addAll(toMemoryEntries(item)));
+ return entries;
+ }
+
+ JsonNode memories = node.get("memories");
+ if (memories != null && memories.isArray()) {
+ return toMemoryEntries(memories);
+ }
+
+ JsonNode content = node.path("content");
+ String role = firstText(content.path("role"), node.path("role"), "user");
+ JsonNode parts = content.has("parts") ? content.path("parts") : node.path("parts");
+ List textParts = new ArrayList<>();
+ if (parts.isArray()) {
+ for (JsonNode part : parts) {
+ String text = part.isTextual() ? part.asText() : part.path("text").asText("");
+ if (!text.isBlank()) {
+ textParts.add(text);
+ }
+ }
+ }
+ if (textParts.isEmpty()) {
+ for (String field : List.of("text", "abstract", "summary")) {
+ String text = node.path(field).asText("");
+ if (!text.isBlank()) {
+ textParts.add(text);
+ break;
+ }
+ }
+ }
+ if (textParts.isEmpty() && content.isTextual()) {
+ textParts.add(content.asText());
+ }
+ return textParts.isEmpty()
+ ? List.of()
+ : List.of(memoryEntry(role, String.join("\n", textParts)));
+ }
+
+ private static String firstText(JsonNode first, JsonNode second, String fallback) {
+ if (first != null && first.isTextual() && !first.asText().isBlank()) {
+ return first.asText();
+ }
+ if (second != null && second.isTextual() && !second.asText().isBlank()) {
+ return second.asText();
+ }
+ return fallback;
+ }
+
+ private static MemoryEntry memoryEntry(String role, String text) {
+ return MemoryEntry.builder()
+ .author(role)
+ .content(Content.builder().role(role).parts(List.of(Part.fromText(text))).build())
+ .build();
+ }
+
+ private static String requireText(String value, String field) {
+ if (value == null || value.isBlank()) {
+ throw new IllegalArgumentException(field + " must not be blank");
+ }
+ return value;
+ }
+}
diff --git a/core/src/main/java/com/volcengine/veadk/memory/LongTermMemoryBackend.java b/core/src/main/java/com/volcengine/veadk/memory/LongTermMemoryBackend.java
new file mode 100644
index 0000000..c014344
--- /dev/null
+++ b/core/src/main/java/com/volcengine/veadk/memory/LongTermMemoryBackend.java
@@ -0,0 +1,42 @@
+/**
+ * Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+ *
+ * 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 com.volcengine.veadk.memory;
+
+import java.util.List;
+import java.util.Map;
+
+/** Storage and retrieval contract implemented by long-term-memory backends. */
+public interface LongTermMemoryBackend extends AutoCloseable {
+
+ String index();
+
+ boolean saveMemory(String userId, List eventStrings, Map options);
+
+ List searchMemory(String userId, String query, int topK, Map options);
+
+ default boolean saveMemory(String userId, List eventStrings) {
+ return saveMemory(userId, eventStrings, Map.of());
+ }
+
+ default List searchMemory(String userId, String query, int topK) {
+ return searchMemory(userId, query, topK, Map.of());
+ }
+
+ default String backendName() {
+ return getClass().getSimpleName();
+ }
+
+ @Override
+ default void close() {}
+}
diff --git a/core/src/main/java/com/volcengine/veadk/memory/MemoryServiceBackendAdapter.java b/core/src/main/java/com/volcengine/veadk/memory/MemoryServiceBackendAdapter.java
new file mode 100644
index 0000000..78e2663
--- /dev/null
+++ b/core/src/main/java/com/volcengine/veadk/memory/MemoryServiceBackendAdapter.java
@@ -0,0 +1,59 @@
+/**
+ * Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+ *
+ * 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 com.volcengine.veadk.memory;
+
+import com.google.adk.memory.BaseMemoryService;
+import com.google.adk.memory.SearchMemoryResponse;
+import com.volcengine.veadk.utils.JSONUtil;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/** Read-only adapter that exposes an existing ADK memory service as a new backend. */
+public final class MemoryServiceBackendAdapter implements LongTermMemoryBackend {
+
+ private final String index;
+ private final BaseMemoryService service;
+
+ public MemoryServiceBackendAdapter(String index, BaseMemoryService service) {
+ if (index == null || index.isBlank()) {
+ throw new IllegalArgumentException("index must not be blank");
+ }
+ this.index = index;
+ this.service = Objects.requireNonNull(service, "service");
+ }
+
+ @Override
+ public String index() {
+ return index;
+ }
+
+ @Override
+ public boolean saveMemory(
+ String userId, List eventStrings, Map options) {
+ throw new UnsupportedOperationException("The legacy memory service is read-only");
+ }
+
+ @Override
+ public List searchMemory(
+ String userId, String query, int topK, Map options) {
+ Object configuredAppName = options.get("appName");
+ String appName = configuredAppName == null ? index : configuredAppName.toString();
+ SearchMemoryResponse response = service.searchMemory(appName, userId, query).blockingGet();
+ if (response == null || response.memories() == null) {
+ return List.of();
+ }
+ return response.memories().stream().limit(topK).map(JSONUtil::toJson).toList();
+ }
+}
diff --git a/core/src/main/java/com/volcengine/veadk/memory/SessionServiceBackendAdapter.java b/core/src/main/java/com/volcengine/veadk/memory/SessionServiceBackendAdapter.java
new file mode 100644
index 0000000..56dd5d0
--- /dev/null
+++ b/core/src/main/java/com/volcengine/veadk/memory/SessionServiceBackendAdapter.java
@@ -0,0 +1,32 @@
+/**
+ * Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+ *
+ * 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 com.volcengine.veadk.memory;
+
+import com.google.adk.sessions.BaseSessionService;
+import java.util.Objects;
+
+/** Adapts an existing ADK session service to the VeADK short-term-memory backend SPI. */
+public final class SessionServiceBackendAdapter implements ShortTermMemoryBackend {
+
+ private final BaseSessionService sessionService;
+
+ public SessionServiceBackendAdapter(BaseSessionService sessionService) {
+ this.sessionService = Objects.requireNonNull(sessionService, "sessionService");
+ }
+
+ @Override
+ public BaseSessionService sessionService() {
+ return sessionService;
+ }
+}
diff --git a/core/src/main/java/com/volcengine/veadk/memory/ShortTermMemory.java b/core/src/main/java/com/volcengine/veadk/memory/ShortTermMemory.java
new file mode 100644
index 0000000..74fa20b
--- /dev/null
+++ b/core/src/main/java/com/volcengine/veadk/memory/ShortTermMemory.java
@@ -0,0 +1,153 @@
+/**
+ * Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+ *
+ *
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 com.volcengine.veadk.memory;
+
+import com.google.adk.events.Event;
+import com.google.adk.sessions.BaseSessionService;
+import com.google.adk.sessions.GetSessionConfig;
+import com.google.adk.sessions.ListEventsResponse;
+import com.google.adk.sessions.ListSessionsResponse;
+import com.google.adk.sessions.Session;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Maybe;
+import io.reactivex.rxjava3.core.Single;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.function.Consumer;
+
+/** ADK-compatible short-term-memory facade aligned with the Python VeADK contract. */
+public final class ShortTermMemory implements BaseSessionService, AutoCloseable {
+
+ private final ShortTermMemoryBackend backend;
+ private final BaseSessionService delegate;
+ private final ShortTermMemoryProcessor processor;
+ private final Consumer afterLoadMemoryCallback;
+
+ public ShortTermMemory() {
+ this(new InMemoryShortTermMemoryBackend());
+ }
+
+ public ShortTermMemory(BaseSessionService sessionService) {
+ this(new SessionServiceBackendAdapter(sessionService));
+ }
+
+ public ShortTermMemory(ShortTermMemoryBackend backend) {
+ this(backend, null, session -> {});
+ }
+
+ public ShortTermMemory(
+ ShortTermMemoryBackend backend, Consumer afterLoadMemoryCallback) {
+ this(backend, null, afterLoadMemoryCallback);
+ }
+
+ public ShortTermMemory(ShortTermMemoryBackend backend, ShortTermMemoryProcessor processor) {
+ this(backend, processor, session -> {});
+ }
+
+ public ShortTermMemory(
+ ShortTermMemoryBackend backend,
+ ShortTermMemoryProcessor processor,
+ Consumer afterLoadMemoryCallback) {
+ this.backend = Objects.requireNonNull(backend, "backend");
+ this.delegate = Objects.requireNonNull(backend.sessionService(), "sessionService");
+ this.processor = processor;
+ this.afterLoadMemoryCallback =
+ Objects.requireNonNull(afterLoadMemoryCallback, "afterLoadMemoryCallback");
+ }
+
+ public BaseSessionService sessionService() {
+ return this;
+ }
+
+ public String backend() {
+ return backend.backendName();
+ }
+
+ /** Creates a session, or returns the existing session when the identifier is already present. */
+ public Single createSession(String appName, String userId, String sessionId) {
+ return getSession(appName, userId, sessionId, Optional.empty())
+ .switchIfEmpty(
+ createSession(appName, userId, new ConcurrentHashMap<>(), sessionId));
+ }
+
+ @Override
+ public Single createSession(
+ String appName, String userId, ConcurrentMap state, String sessionId) {
+ return delegate.createSession(appName, userId, state, sessionId);
+ }
+
+ @Override
+ public Maybe getSession(
+ String appName, String userId, String sessionId, Optional config) {
+ return delegate.getSession(appName, userId, sessionId, config)
+ .map(session -> processor == null ? session : processor.afterLoadSession(session))
+ .doOnSuccess(afterLoadMemoryCallback::accept);
+ }
+
+ @Override
+ public Single listSessions(String appName, String userId) {
+ return delegate.listSessions(appName, userId);
+ }
+
+ @Override
+ public Completable deleteSession(String appName, String userId, String sessionId) {
+ return delegate.deleteSession(appName, userId, sessionId);
+ }
+
+ @Override
+ public Single listEvents(String appName, String userId, String sessionId) {
+ return delegate.listEvents(appName, userId, sessionId);
+ }
+
+ @Override
+ public Completable closeSession(Session session) {
+ return delegate.closeSession(session);
+ }
+
+ @Override
+ public Single appendEvent(Session session, Event event) {
+ if (processor != null) {
+ // The processor returns a presentation-only session. Keep that view current for the
+ // active invocation, but append to a freshly loaded canonical session so optimized or
+ // filtered history never replaces the persisted history.
+ return BaseSessionService.super
+ .appendEvent(session, event)
+ .flatMap(
+ appendedEvent ->
+ delegate.getSession(
+ session.appName(),
+ session.userId(),
+ session.id(),
+ Optional.empty())
+ .switchIfEmpty(
+ Single.error(
+ new IllegalStateException(
+ "Session not found: "
+ + session.id())))
+ .flatMap(
+ persistedSession ->
+ delegate.appendEvent(
+ persistedSession,
+ appendedEvent)));
+ }
+ return delegate.appendEvent(session, event);
+ }
+
+ @Override
+ public void close() {
+ backend.close();
+ }
+}
diff --git a/core/src/main/java/com/volcengine/veadk/memory/ShortTermMemoryBackend.java b/core/src/main/java/com/volcengine/veadk/memory/ShortTermMemoryBackend.java
new file mode 100644
index 0000000..59938db
--- /dev/null
+++ b/core/src/main/java/com/volcengine/veadk/memory/ShortTermMemoryBackend.java
@@ -0,0 +1,29 @@
+/**
+ * Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+ *
+ * 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 com.volcengine.veadk.memory;
+
+import com.google.adk.sessions.BaseSessionService;
+
+/** Contract implemented by short-term-memory storage backends. */
+public interface ShortTermMemoryBackend extends AutoCloseable {
+
+ BaseSessionService sessionService();
+
+ default String backendName() {
+ return getClass().getSimpleName();
+ }
+
+ @Override
+ default void close() {}
+}
diff --git a/core/src/main/java/com/volcengine/veadk/memory/ShortTermMemoryMessage.java b/core/src/main/java/com/volcengine/veadk/memory/ShortTermMemoryMessage.java
new file mode 100644
index 0000000..e09c155
--- /dev/null
+++ b/core/src/main/java/com/volcengine/veadk/memory/ShortTermMemoryMessage.java
@@ -0,0 +1,27 @@
+/**
+ * Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+ *
+ *
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 com.volcengine.veadk.memory;
+
+/** A provider-neutral chat message used by short-term-memory optimizers. */
+public record ShortTermMemoryMessage(String role, String content) {
+
+ public ShortTermMemoryMessage {
+ if (role == null || role.isBlank()) {
+ throw new IllegalArgumentException("role must not be blank");
+ }
+ if (content == null || content.isBlank()) {
+ throw new IllegalArgumentException("content must not be blank");
+ }
+ }
+}
diff --git a/core/src/main/java/com/volcengine/veadk/memory/ShortTermMemoryProcessor.java b/core/src/main/java/com/volcengine/veadk/memory/ShortTermMemoryProcessor.java
new file mode 100644
index 0000000..4a81dca
--- /dev/null
+++ b/core/src/main/java/com/volcengine/veadk/memory/ShortTermMemoryProcessor.java
@@ -0,0 +1,82 @@
+/**
+ * Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+ *
+ *
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 com.volcengine.veadk.memory;
+
+import com.google.adk.events.Event;
+import com.google.adk.sessions.Session;
+import com.google.genai.types.Content;
+import com.google.genai.types.Part;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+
+/** Rewrites loaded session history using a caller-provided message optimizer. */
+public final class ShortTermMemoryProcessor {
+
+ private final Function, List> optimizer;
+
+ public ShortTermMemoryProcessor(
+ Function, List> optimizer) {
+ this.optimizer = Objects.requireNonNull(optimizer, "optimizer");
+ }
+
+ public Session afterLoadSession(Session session) {
+ Objects.requireNonNull(session, "session");
+ List sourceMessages = extractMessages(session.events());
+ List optimizedMessages =
+ List.copyOf(Objects.requireNonNull(optimizer.apply(sourceMessages), "messages"));
+ List optimizedEvents =
+ optimizedMessages.stream().map(ShortTermMemoryProcessor::toEvent).toList();
+ return Session.builder(session.id())
+ .appName(session.appName())
+ .userId(session.userId())
+ .state(new ConcurrentHashMap<>(session.state()))
+ // ADK's default appendEvent implementation appends in place. Stream.toList()
+ // returns an unmodifiable list, so keep processor-created session views mutable.
+ .events(new ArrayList<>(optimizedEvents))
+ .lastUpdateTime(session.lastUpdateTime())
+ .build();
+ }
+
+ private static List extractMessages(List events) {
+ List messages = new ArrayList<>();
+ for (Event event : events) {
+ if (event.content().isEmpty()) {
+ continue;
+ }
+ Content content = event.content().get();
+ List parts = content.parts().orElse(List.of());
+ if (parts.isEmpty() || parts.get(0).text().isEmpty()) {
+ continue;
+ }
+ messages.add(
+ new ShortTermMemoryMessage(
+ content.role().orElse(event.author()), parts.get(0).text().get()));
+ }
+ return List.copyOf(messages);
+ }
+
+ private static Event toEvent(ShortTermMemoryMessage message) {
+ return Event.builder()
+ .author("memory_optimizer")
+ .content(
+ Content.builder()
+ .role(message.role())
+ .parts(List.of(Part.fromText(message.content())))
+ .build())
+ .build();
+ }
+}
diff --git a/core/src/main/java/com/volcengine/veadk/memory/viking/VikingMemoryService.java b/core/src/main/java/com/volcengine/veadk/memory/viking/VikingMemoryService.java
index 858925c..e429a14 100644
--- a/core/src/main/java/com/volcengine/veadk/memory/viking/VikingMemoryService.java
+++ b/core/src/main/java/com/volcengine/veadk/memory/viking/VikingMemoryService.java
@@ -15,6 +15,7 @@
*/
package com.volcengine.veadk.memory.viking;
+import com.fasterxml.jackson.databind.JsonNode;
import com.google.adk.memory.BaseMemoryService;
import com.google.adk.memory.MemoryEntry;
import com.google.adk.memory.SearchMemoryResponse;
@@ -22,15 +23,20 @@
import com.volcengine.veadk.integration.vikingmemory.Message;
import com.volcengine.veadk.integration.vikingmemory.Metadata;
import com.volcengine.veadk.integration.vikingmemory.VikingMemoryWrapper;
+import com.volcengine.veadk.memory.LongTermMemoryBackend;
import com.volcengine.veadk.utils.EnvUtil;
+import com.volcengine.veadk.utils.JSONUtil;
import io.reactivex.rxjava3.core.Completable;
import io.reactivex.rxjava3.core.Single;
+import java.io.IOException;
+import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
import java.util.stream.Collectors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class VikingMemoryService implements BaseMemoryService {
+public class VikingMemoryService implements BaseMemoryService, LongTermMemoryBackend {
private static final Logger log = LoggerFactory.getLogger(VikingMemoryService.class);
@@ -115,4 +121,74 @@ public Single searchMemory(String appName, String userId,
return SearchMemoryResponse.builder().setMemories(memoryEntries).build();
});
}
+
+ @Override
+ public String index() {
+ return appName;
+ }
+
+ @Override
+ public boolean saveMemory(
+ String userId, List eventStrings, Map options) {
+ List messages =
+ eventStrings.stream()
+ .map(VikingMemoryService::toMessage)
+ .collect(Collectors.toList());
+ if (messages.isEmpty()) {
+ return true;
+ }
+ try {
+ Metadata metadata = new Metadata(userId, "assistant", System.currentTimeMillis());
+ return vikingMemoryWrapper.addSession(appName, messages, metadata);
+ } catch (Exception exception) {
+ throw new IllegalStateException("Unable to save Viking long-term memory", exception);
+ }
+ }
+
+ @Override
+ public List searchMemory(
+ String userId, String query, int requestedTopK, Map options) {
+ try {
+ return vikingMemoryWrapper
+ .searchMemory(appName, userId, query, requestedTopK, builtinEventTypes)
+ .stream()
+ .map(JSONUtil::toJson)
+ .toList();
+ } catch (Exception exception) {
+ throw new IllegalStateException("Unable to search Viking long-term memory", exception);
+ }
+ }
+
+ private static Message toMessage(String eventString) {
+ try {
+ JsonNode event = JSONUtil.parseJson(eventString);
+ JsonNode content = event.path("content");
+ String role = content.path("role").asText(event.path("role").asText("user"));
+ JsonNode parts = content.has("parts") ? content.path("parts") : event.path("parts");
+ List textParts = new ArrayList<>();
+ if (parts.isArray()) {
+ for (JsonNode part : parts) {
+ String text = part.isTextual() ? part.asText() : part.path("text").asText("");
+ if (!text.isBlank()) {
+ textParts.add(text);
+ }
+ }
+ }
+ String text = String.join("\n", textParts);
+ if (!text.isBlank()) {
+ return new Message(normalizeRole(role), text);
+ }
+ } catch (IOException exception) {
+ // Plain text is a valid backend input and is handled below.
+ }
+ return new Message("user", eventString);
+ }
+
+ private static String normalizeRole(String role) {
+ return switch (role) {
+ case "assistant", "model" -> "assistant";
+ case "system" -> "system";
+ default -> "user";
+ };
+ }
}
diff --git a/core/src/main/java/com/volcengine/veadk/model/ArkEmbedding.java b/core/src/main/java/com/volcengine/veadk/model/ArkEmbedding.java
new file mode 100644
index 0000000..0897c3f
--- /dev/null
+++ b/core/src/main/java/com/volcengine/veadk/model/ArkEmbedding.java
@@ -0,0 +1,230 @@
+/**
+ * Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+ *
+ * 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 com.volcengine.veadk.model;
+
+import com.volcengine.ark.runtime.model.multimodalembeddings.MultimodalEmbeddingInput;
+import com.volcengine.ark.runtime.model.multimodalembeddings.MultimodalEmbeddingRequest;
+import com.volcengine.ark.runtime.model.multimodalembeddings.MultimodalEmbeddingResult;
+import com.volcengine.ark.runtime.model.multimodalembeddings.MultimodalEmbeddingUsage;
+import com.volcengine.ark.runtime.service.ArkService;
+import com.volcengine.veadk.config.VeADKConfig;
+import io.reactivex.rxjava3.core.Single;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+
+/** Ark multimodal embedding adapter with synchronous, reactive, and batched text APIs. */
+public final class ArkEmbedding {
+
+ public enum Model {
+ DOUBAO_EMBEDDING_VISION_251215("doubao-embedding-vision-251215"),
+ DOUBAO_EMBEDDING_VISION_250615("doubao-embedding-vision-250615");
+
+ private final String id;
+
+ Model(String id) {
+ this.id = id;
+ }
+
+ public String id() {
+ return id;
+ }
+ }
+
+ private final String modelName;
+ private final Integer dimensions;
+ private final ArkService arkService;
+
+ public ArkEmbedding(String modelName, String apiKey) {
+ this(builder().modelName(modelName).apiKey(apiKey));
+ }
+
+ public ArkEmbedding(String modelName, String apiKey, String apiBase, Integer dimensions) {
+ this(builder().modelName(modelName).apiKey(apiKey).apiBase(apiBase).dimensions(dimensions));
+ }
+
+ private ArkEmbedding(Builder builder) {
+ modelName = requireText(builder.modelName, "modelName");
+ dimensions = builder.dimensions;
+ if (dimensions != null && dimensions <= 0) {
+ throw new IllegalArgumentException("dimensions must be positive");
+ }
+ arkService = builder.arkService != null ? builder.arkService : createService(builder);
+ }
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ public static Builder builder(VeADKConfig config) {
+ Objects.requireNonNull(config, "config");
+ VeADKConfig.EmbeddingConfig embedding = config.embedding();
+ return new Builder()
+ .modelName(embedding.name())
+ .apiKey(embedding.apiKey())
+ .apiBase(embedding.apiBase())
+ .dimensions(embedding.dimension());
+ }
+
+ public String modelName() {
+ return modelName;
+ }
+
+ public Integer dimensions() {
+ return dimensions;
+ }
+
+ public List embed(String text) {
+ return embedWithUsage(text).embedding();
+ }
+
+ public EmbeddingResponse embedWithUsage(String text) {
+ Objects.requireNonNull(text, "text");
+ MultimodalEmbeddingInput input =
+ MultimodalEmbeddingInput.builder().type("text").text(text).build();
+ MultimodalEmbeddingRequest.Builder request =
+ MultimodalEmbeddingRequest.builder().model(modelName).input(List.of(input));
+ if (dimensions != null) {
+ request.dimensions(dimensions);
+ }
+ MultimodalEmbeddingResult result = arkService.createMultiModalEmbeddings(request.build());
+ if (result == null || result.getData() == null || result.getData().getEmbedding() == null) {
+ throw new IllegalStateException("Ark embedding response did not contain an embedding");
+ }
+ MultimodalEmbeddingUsage usage = result.getUsage();
+ return new EmbeddingResponse(
+ result.getModel() == null ? modelName : result.getModel(),
+ result.getData().getEmbedding(),
+ usage == null ? 0 : usage.getPromptTokens(),
+ usage == null ? 0 : usage.getTotalTokens());
+ }
+
+ public List> embedAll(List texts) {
+ Objects.requireNonNull(texts, "texts");
+ List> embeddings = new ArrayList<>(texts.size());
+ for (String text : texts) {
+ embeddings.add(embed(text));
+ }
+ return List.copyOf(embeddings);
+ }
+
+ public Single> embedAsync(String text) {
+ return Single.fromCallable(() -> embed(text));
+ }
+
+ public Single>> embedAllAsync(List texts) {
+ return Single.fromCallable(() -> embedAll(texts));
+ }
+
+ public List getTextEmbedding(String text) {
+ return embed(text);
+ }
+
+ public List> getTextEmbeddings(List texts) {
+ return embedAll(texts);
+ }
+
+ public List getQueryEmbedding(String query) {
+ return embed(query);
+ }
+
+ private static ArkService createService(Builder builder) {
+ ArkService.Builder service =
+ ArkService.builder()
+ .apiKey(requireText(builder.apiKey, "apiKey"))
+ .timeout(builder.timeout)
+ .retryTimes(builder.maxRetries);
+ if (builder.apiBase != null && !builder.apiBase.isBlank()) {
+ service.baseUrl(builder.apiBase);
+ }
+ return service.build();
+ }
+
+ private static String requireText(String value, String field) {
+ if (value == null || value.isBlank()) {
+ throw new IllegalArgumentException(field + " must not be blank");
+ }
+ return value;
+ }
+
+ public record EmbeddingResponse(
+ String model, List embedding, long promptTokens, long totalTokens) {
+ public EmbeddingResponse {
+ model = Objects.requireNonNullElse(model, "");
+ embedding = List.copyOf(embedding);
+ }
+ }
+
+ public static final class Builder {
+ private String modelName = VeADKConfig.DEFAULT_EMBEDDING_MODEL;
+ private String apiKey;
+ private String apiBase;
+ private Integer dimensions;
+ private int maxRetries = 10;
+ private Duration timeout = Duration.ofSeconds(60);
+ private ArkService arkService;
+
+ private Builder() {}
+
+ public Builder modelName(String modelName) {
+ this.modelName = modelName;
+ return this;
+ }
+
+ public Builder model(Model model) {
+ return modelName(Objects.requireNonNull(model, "model").id());
+ }
+
+ public Builder apiKey(String apiKey) {
+ this.apiKey = apiKey;
+ return this;
+ }
+
+ public Builder apiBase(String apiBase) {
+ this.apiBase = apiBase;
+ return this;
+ }
+
+ public Builder dimensions(Integer dimensions) {
+ this.dimensions = dimensions;
+ return this;
+ }
+
+ public Builder maxRetries(int maxRetries) {
+ if (maxRetries < 0) {
+ throw new IllegalArgumentException("maxRetries must not be negative");
+ }
+ this.maxRetries = maxRetries;
+ return this;
+ }
+
+ public Builder timeout(Duration timeout) {
+ this.timeout = Objects.requireNonNull(timeout, "timeout");
+ return this;
+ }
+
+ /** Supplies a preconfigured client, primarily for custom transports and deterministic tests. */
+ public Builder arkService(ArkService arkService) {
+ this.arkService = Objects.requireNonNull(arkService, "arkService");
+ return this;
+ }
+
+ public ArkEmbedding build() {
+ return new ArkEmbedding(this);
+ }
+ }
+}
diff --git a/core/src/main/java/com/volcengine/veadk/model/ArkLlm.java b/core/src/main/java/com/volcengine/veadk/model/ArkLlm.java
index c9d8196..90b0204 100644
--- a/core/src/main/java/com/volcengine/veadk/model/ArkLlm.java
+++ b/core/src/main/java/com/volcengine/veadk/model/ArkLlm.java
@@ -24,27 +24,40 @@
import com.google.adk.tools.BaseTool;
import com.google.common.collect.ImmutableMap;
import com.google.genai.types.Content;
+import com.google.genai.types.FinishReason;
+import com.google.genai.types.FunctionCall;
import com.google.genai.types.FunctionDeclaration;
import com.google.genai.types.FunctionResponse;
+import com.google.genai.types.GenerateContentConfig;
+import com.google.genai.types.GenerateContentResponseUsageMetadata;
import com.google.genai.types.Part;
import com.google.genai.types.Schema;
+import com.google.genai.types.VideoMetadata;
+import com.volcengine.ark.runtime.model.Usage;
import com.volcengine.ark.runtime.model.completion.chat.ChatCompletionChunk;
+import com.volcengine.ark.runtime.model.completion.chat.ChatCompletionContentPart;
import com.volcengine.ark.runtime.model.completion.chat.ChatCompletionRequest;
import com.volcengine.ark.runtime.model.completion.chat.ChatCompletionResult;
import com.volcengine.ark.runtime.model.completion.chat.ChatFunction;
+import com.volcengine.ark.runtime.model.completion.chat.ChatFunctionCall;
import com.volcengine.ark.runtime.model.completion.chat.ChatMessage;
import com.volcengine.ark.runtime.model.completion.chat.ChatMessageRole;
import com.volcengine.ark.runtime.model.completion.chat.ChatTool;
import com.volcengine.ark.runtime.model.completion.chat.ChatToolCall;
+import com.volcengine.ark.runtime.model.completion.chat.ResponseFormatJSONSchemaJSONSchemaParam;
import com.volcengine.ark.runtime.service.ArkService;
import com.volcengine.veadk.utils.EnvUtil;
import com.volcengine.veadk.utils.JSONUtil;
import io.reactivex.rxjava3.core.Flowable;
import java.util.ArrayList;
+import java.util.Base64;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
+import java.util.TreeMap;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.commons.lang3.StringUtils;
@@ -70,6 +83,7 @@ public final class ArkLlm extends BaseLlm {
.build();
private final ArkService arkService;
+ private final List fallbacks;
private ChatCompletionRequest.ChatCompletionRequestThinking thinking = null;
public ArkLlm(String modelName) {
@@ -77,14 +91,88 @@ public ArkLlm(String modelName) {
}
public ArkLlm(String modelName, String thinking) {
+ this(modelName, thinking, EnvUtil.getAgentApiKey(), null);
+ }
+
+ public ArkLlm(String modelName, String apiKey, String apiBase) {
+ this(modelName, null, apiKey, apiBase);
+ }
+
+ public ArkLlm(String modelName, String thinking, String apiKey, String apiBase) {
+ this(modelName, thinking, apiKey, apiBase, List.of());
+ }
+
+ public ArkLlm(List modelNames) {
+ this(modelNames, null);
+ }
+
+ public ArkLlm(List modelNames, String thinking) {
+ this(primaryModelName(modelNames), thinking, EnvUtil.getAgentApiKey(), null, modelNames);
+ }
+
+ public ArkLlm(List modelNames, String apiKey, String apiBase) {
+ this(primaryModelName(modelNames), null, apiKey, apiBase, modelNames);
+ }
+
+ public ArkLlm(List modelNames, String thinking, String apiKey, String apiBase) {
+ this(primaryModelName(modelNames), thinking, apiKey, apiBase, modelNames);
+ }
+
+ private ArkLlm(
+ String modelName, String thinking, String apiKey, String apiBase, List models) {
super(modelName);
- Objects.requireNonNull(modelName, "modelName must be set.");
- this.arkService = ArkService.builder().apiKey(EnvUtil.getAgentApiKey()).build();
+ requireModelName(modelName);
+ this.fallbacks = fallbackModelNames(models);
+ ArkService.Builder serviceBuilder =
+ ArkService.builder().apiKey(Objects.requireNonNull(apiKey, "apiKey must be set."));
+ if (StringUtils.isNotBlank(apiBase)) {
+ serviceBuilder.baseUrl(apiBase);
+ }
+ this.arkService = serviceBuilder.build();
if (StringUtils.isNotBlank(thinking)) {
this.thinking = new ChatCompletionRequest.ChatCompletionRequestThinking(thinking);
}
}
+ public List fallbacks() {
+ return fallbacks;
+ }
+
+ private static String primaryModelName(List modelNames) {
+ Objects.requireNonNull(modelNames, "modelNames must be set.");
+ if (modelNames.isEmpty()) {
+ throw new IllegalArgumentException("modelNames must not be empty");
+ }
+ return requireModelName(modelNames.get(0));
+ }
+
+ private static List fallbackModelNames(List modelNames) {
+ if (modelNames == null || modelNames.size() <= 1) {
+ return List.of();
+ }
+ return modelNames.subList(1, modelNames.size()).stream()
+ .map(ArkLlm::requireModelName)
+ .toList();
+ }
+
+ private static String requireModelName(String modelName) {
+ Objects.requireNonNull(modelName, "modelName must be set.");
+ if (modelName.isBlank()) {
+ throw new IllegalArgumentException("modelName must not be blank");
+ }
+ return modelName;
+ }
+
+ private List modelCandidates(LlmRequest llmRequest) {
+ Optional requestedModel = llmRequest.model().filter(StringUtils::isNotBlank);
+ // ADK's Basic request processor always copies BaseLlm.model() into the request. Treat that
+ // value as the configured primary model so fallbacks remain available in normal Agent runs.
+ if (requestedModel.isPresent() && !requestedModel.get().equals(model())) {
+ return List.of(requestedModel.get());
+ }
+ return Stream.concat(Stream.of(model()), fallbacks.stream()).toList();
+ }
+
/**
* Generate content based on LLM request
* @param llmRequest The request containing prompts and parameters
@@ -93,19 +181,49 @@ public ArkLlm(String modelName, String thinking) {
*/
@Override
public Flowable generateContent(LlmRequest llmRequest, boolean stream) {
- // Convert ADK request to Ark request format
- ChatCompletionRequest arkRequest = toArkRequest(llmRequest);
+ return Flowable.defer(
+ () ->
+ generateContentWithFallbacks(
+ llmRequest, stream, modelCandidates(llmRequest), 0));
+ }
+
+ private Flowable generateContentWithFallbacks(
+ LlmRequest llmRequest, boolean stream, List modelCandidates, int index) {
+ String modelName = modelCandidates.get(index);
+ ChatCompletionRequest arkRequest = toArkRequest(llmRequest, modelName);
+ Flowable attempt =
+ Flowable.defer(() -> generateContentWithModel(arkRequest, stream));
+ AtomicBoolean emittedResponse = new AtomicBoolean(false);
+ return attempt.doOnNext(response -> emittedResponse.set(true))
+ .onErrorResumeNext(
+ error -> {
+ if (emittedResponse.get() || index + 1 >= modelCandidates.size()) {
+ return Flowable.error(error);
+ }
+ String nextModel = modelCandidates.get(index + 1);
+ log.warn(
+ "Ark request with model {} failed before emitting a response;"
+ + " falling back to {}",
+ modelName,
+ nextModel,
+ error);
+ return generateContentWithFallbacks(
+ llmRequest, stream, modelCandidates, index + 1);
+ });
+ }
+
+ private Flowable generateContentWithModel(
+ ChatCompletionRequest arkRequest, boolean stream) {
if (stream) {
log.debug(
"Sending streaming generateContent request to model {}", arkRequest.getModel());
- // Handle streaming response
+ arkRequest.setStreamOptions(
+ ChatCompletionRequest.ChatCompletionRequestStreamOptions.of(true));
return generateContentStreaming(arkRequest);
- } else {
- log.debug("Sending generateContent request to model {}", arkRequest.getModel());
- // Handle non-streaming response
- return Flowable.fromCallable(() -> arkService.createChatCompletion(arkRequest))
- .map(this::toLlmResponse);
}
+ log.debug("Sending generateContent request to model {}", arkRequest.getModel());
+ return Flowable.fromCallable(() -> arkService.createChatCompletion(arkRequest))
+ .map(this::toLlmResponse);
}
/**
@@ -114,70 +232,64 @@ public Flowable generateContent(LlmRequest llmRequest, boolean stre
* @return Flowable of LlmResponse objects
*/
private Flowable generateContentStreaming(ChatCompletionRequest arkRequest) {
- // Get streaming response from Ark service
io.reactivex.Flowable streamResponse =
arkService.streamChatCompletion(arkRequest);
return Flowable.defer(
() -> {
- // Accumulate complete text response
final StringBuilder accumulatedText = new StringBuilder();
- // Buffer partial text for incremental responses
final StringBuilder partialText = new StringBuilder();
- // Hold the last chunk for final processing
- final ChatCompletionChunk[] lastChunkHolder = {null};
- // Accumulate tool calls if any
- final List accumulatedToolCalls = new ArrayList<>();
+ final Map accumulatedToolCalls = new TreeMap<>();
+ final String[] finishReason = {null};
+ final Usage[] usage = {null};
return Flowable.fromPublisher(streamResponse)
.concatMap(
chunk -> {
- lastChunkHolder[0] = chunk;
log.debug("Raw Ark streaming chunk: {}", chunk);
+ if (chunk.getUsage() != null) {
+ usage[0] = chunk.getUsage();
+ }
+ String chunkFinishReason = finishReason(chunk);
+ if (StringUtils.isNotBlank(chunkFinishReason)) {
+ finishReason[0] = chunkFinishReason;
+ }
- // Prepare list of responses to emit
List responsesToEmit = new ArrayList<>();
- // Process text content from chunk
processTextContent(
chunk,
accumulatedText,
partialText,
responsesToEmit);
- // Process tool calls from chunk
processToolCalls(chunk, accumulatedToolCalls);
- // Handle stop chunk (final chunk)
- if (isStopChunk(chunk)) {
+ if (StringUtils.isNotBlank(chunkFinishReason)) {
processStopChunk(partialText, responsesToEmit);
}
- // Emit responses if any
if (responsesToEmit.isEmpty()) {
return Flowable.empty();
- } else {
- log.debug("Responses to emit: {}", responsesToEmit);
- return Flowable.fromIterable(responsesToEmit);
}
+ log.debug("Responses to emit: {}", responsesToEmit);
+ return Flowable.fromIterable(responsesToEmit);
})
.concatWith(
Flowable.defer(
() -> {
- // Process final response after stream ends
- ChatCompletionChunk lastChunk = lastChunkHolder[0];
- if (lastChunk == null
- || accumulatedText.length() == 0) {
+ if (StringUtils.isBlank(finishReason[0])
+ || (accumulatedText.isEmpty()
+ && accumulatedToolCalls
+ .isEmpty())) {
return Flowable.empty();
}
-
- if (isStopChunk(lastChunk)) {
- // Build and emit final aggregated response
- return Flowable.just(
- buildFinalResponse(
- accumulatedText.toString(),
- accumulatedToolCalls));
- }
-
- return Flowable.empty();
+ return Flowable.just(
+ buildFinalResponse(
+ accumulatedText.toString(),
+ new ArrayList<>(
+ accumulatedToolCalls
+ .values()),
+ finishReason[0],
+ usage[0]));
}));
});
}
@@ -194,18 +306,18 @@ private void processTextContent(
StringBuilder accumulatedText,
StringBuilder partialText,
List responsesToEmit) {
- // Extract text content from chunk
- String content = (String) chunk.getChoices().get(0).getMessage().getContent();
+ if (!hasMessageChoice(chunk)) {
+ return;
+ }
+ Object rawContent = chunk.getChoices().get(0).getMessage().getContent();
+ String content = rawContent instanceof String ? (String) rawContent : null;
if (StringUtils.isNotEmpty(content)) {
- // Add to accumulated text
accumulatedText.append(content);
- // Add to partial text buffer
partialText.append(content);
- // Emit partial response when buffer exceeds threshold
if (partialText.length() > 30) {
responsesToEmit.add(buildPartialResponse(partialText.toString()));
- partialText.setLength(0); // Clear buffer
+ partialText.setLength(0);
}
}
}
@@ -228,22 +340,73 @@ private LlmResponse buildPartialResponse(String text) {
* @param accumulatedToolCalls List of accumulated tool calls
*/
private void processToolCalls(
- ChatCompletionChunk chunk, List accumulatedToolCalls) {
- // Extract tool calls from chunk
+ ChatCompletionChunk chunk, Map accumulatedToolCalls) {
+ if (!hasMessageChoice(chunk)) {
+ return;
+ }
List toolCalls = chunk.getChoices().get(0).getMessage().getToolCalls();
- if (null != toolCalls && !toolCalls.isEmpty()) {
- ChatToolCall toolCall = toolCalls.get(0);
- // If tool call has ID, it's a new tool call
- if (StringUtils.isNotBlank(toolCall.getId())) {
- accumulatedToolCalls.add(toolCall);
- } else {
- // Otherwise, it's continuation of existing tool call
- int index = toolCall.getIndex();
- String arguments =
- accumulatedToolCalls.get(index).getFunction().getArguments()
- + toolCall.getFunction().getArguments();
- accumulatedToolCalls.get(index).getFunction().setArguments(arguments);
+ if (toolCalls == null) {
+ return;
+ }
+ for (ChatToolCall toolCall : toolCalls) {
+ if (toolCall == null) {
+ continue;
}
+ int index = resolveToolCallIndex(toolCall, accumulatedToolCalls);
+ ChatToolCall accumulated =
+ accumulatedToolCalls.computeIfAbsent(index, unused -> new ChatToolCall());
+ mergeToolCall(accumulated, toolCall);
+ }
+ }
+
+ private int resolveToolCallIndex(
+ ChatToolCall toolCall, Map accumulatedToolCalls) {
+ if (toolCall.getIndex() != null && toolCall.getIndex() >= 0) {
+ return toolCall.getIndex();
+ }
+ if (StringUtils.isNotBlank(toolCall.getId())) {
+ return accumulatedToolCalls.entrySet().stream()
+ .filter(entry -> toolCall.getId().equals(entry.getValue().getId()))
+ .map(Map.Entry::getKey)
+ .findFirst()
+ .orElseGet(() -> nextToolCallIndex(accumulatedToolCalls));
+ }
+ return accumulatedToolCalls.size() == 1
+ ? accumulatedToolCalls.keySet().iterator().next()
+ : nextToolCallIndex(accumulatedToolCalls);
+ }
+
+ private int nextToolCallIndex(Map accumulatedToolCalls) {
+ int index = 0;
+ while (accumulatedToolCalls.containsKey(index)) {
+ index++;
+ }
+ return index;
+ }
+
+ private void mergeToolCall(ChatToolCall target, ChatToolCall fragment) {
+ if (StringUtils.isNotBlank(fragment.getId())) {
+ target.setId(fragment.getId());
+ }
+ if (StringUtils.isNotBlank(fragment.getType())) {
+ target.setType(fragment.getType());
+ }
+ ChatFunctionCall fragmentFunction = fragment.getFunction();
+ if (fragmentFunction == null) {
+ return;
+ }
+ ChatFunctionCall targetFunction = target.getFunction();
+ if (targetFunction == null) {
+ targetFunction = new ChatFunctionCall();
+ target.setFunction(targetFunction);
+ }
+ if (StringUtils.isNotBlank(fragmentFunction.getName())) {
+ targetFunction.setName(fragmentFunction.getName());
+ }
+ if (fragmentFunction.getArguments() != null) {
+ targetFunction.setArguments(
+ Objects.requireNonNullElse(targetFunction.getArguments(), "")
+ + fragmentFunction.getArguments());
}
}
@@ -256,7 +419,7 @@ private void processStopChunk(StringBuilder partialText, List respo
// Emit any remaining partial text
if (!partialText.isEmpty()) {
responsesToEmit.add(buildPartialResponse(partialText.toString()));
- partialText.setLength(0); // Clear buffer
+ partialText.setLength(0);
}
}
@@ -267,22 +430,26 @@ private void processStopChunk(StringBuilder partialText, List respo
* @return Final LlmResponse object
*/
private LlmResponse buildFinalResponse(
- String accumulatedText, List accumulatedToolCalls) {
+ String accumulatedText,
+ List accumulatedToolCalls,
+ String finishReason,
+ Usage usage) {
List parts = new ArrayList<>();
- // Add text part
- parts.add(Part.fromText(accumulatedText));
+ if (StringUtils.isNotEmpty(accumulatedText)) {
+ parts.add(Part.fromText(accumulatedText));
+ }
- // Add tool call parts if any
if (!accumulatedToolCalls.isEmpty()) {
parts.addAll(parseToolCalls(accumulatedToolCalls));
}
- // Build final response
- LlmResponse finalAggregatedResponse =
+ LlmResponse.Builder builder =
LlmResponse.builder()
.content(Content.builder().role("model").parts(parts).build())
- .partial(false) // Mark as complete response
- .build();
+ .partial(false);
+ toFinishReason(finishReason).ifPresent(builder::finishReason);
+ toUsageMetadata(usage).ifPresent(builder::usageMetadata);
+ LlmResponse finalAggregatedResponse = builder.build();
log.debug("finalAggregatedResponse to emit: {}", finalAggregatedResponse);
return finalAggregatedResponse;
}
@@ -292,9 +459,16 @@ private LlmResponse buildFinalResponse(
* @param chunk The streaming chunk
* @return True if chunk is stop chunk, false otherwise
*/
- private boolean isStopChunk(ChatCompletionChunk chunk) {
- String finishReason = chunk.getChoices().get(0).getFinishReason();
- return StringUtils.isNotBlank(finishReason);
+ private String finishReason(ChatCompletionChunk chunk) {
+ return chunk.getChoices() == null || chunk.getChoices().isEmpty()
+ ? null
+ : chunk.getChoices().get(0).getFinishReason();
+ }
+
+ private boolean hasMessageChoice(ChatCompletionChunk chunk) {
+ return chunk.getChoices() != null
+ && !chunk.getChoices().isEmpty()
+ && chunk.getChoices().get(0).getMessage() != null;
}
/**
@@ -304,54 +478,89 @@ private boolean isStopChunk(ChatCompletionChunk chunk) {
*/
private LlmResponse toLlmResponse(ChatCompletionResult arkResponse) {
log.debug("Raw Ark response:{}", arkResponse);
- LlmResponse response = null;
// Check finish reason to determine response type
String finishReason = arkResponse.getChoices().get(0).getFinishReason();
+ List parts = new ArrayList<>();
+ String text = (String) arkResponse.getChoices().get(0).getMessage().getContent();
+ if (StringUtils.isNotEmpty(text)) {
+ parts.add(Part.fromText(text));
+ }
if ("tool_calls".equalsIgnoreCase(finishReason)) {
- // Handle tool call response
- List parts = new ArrayList<>();
-
- // Add text content if any
- String text = (String) arkResponse.getChoices().get(0).getMessage().getContent();
- if (StringUtils.isNotEmpty(text)) {
- parts.add(Part.fromText(text));
- }
-
// Add tool call parts
parts.addAll(
parseToolCalls(arkResponse.getChoices().get(0).getMessage().getToolCalls()));
-
- response =
- LlmResponse.builder()
- .content(Content.builder().role("model").parts(parts).build())
- .build();
- } else {
- // Handle regular text response
- String text = (String) arkResponse.getChoices().get(0).getMessage().getContent();
- response =
- LlmResponse.builder()
- .content(
- Content.builder()
- .role("model")
- .parts(Part.fromText(text))
- .build())
- .build();
}
+ LlmResponse response = buildLlmResponse(parts, finishReason, arkResponse.getUsage());
log.debug("LlmResponse:{}", response);
return response;
}
+ private LlmResponse buildLlmResponse(String text, String finishReason, Usage usage) {
+ List parts = new ArrayList<>();
+ if (StringUtils.isNotEmpty(text)) {
+ parts.add(Part.fromText(text));
+ }
+ return buildLlmResponse(parts, finishReason, usage);
+ }
+
+ private LlmResponse buildLlmResponse(List parts, String finishReason, Usage usage) {
+ LlmResponse.Builder builder =
+ LlmResponse.builder().content(Content.builder().role("model").parts(parts).build());
+ toFinishReason(finishReason).ifPresent(builder::finishReason);
+ toUsageMetadata(usage).ifPresent(builder::usageMetadata);
+ return builder.build();
+ }
+
+ private Optional toFinishReason(String finishReason) {
+ if (StringUtils.isBlank(finishReason)) {
+ return Optional.empty();
+ }
+ FinishReason.Known known =
+ switch (finishReason.toLowerCase()) {
+ case "stop", "tool_calls", "function_call" -> FinishReason.Known.STOP;
+ case "length" -> FinishReason.Known.MAX_TOKENS;
+ case "content_filter" -> FinishReason.Known.SAFETY;
+ default -> FinishReason.Known.OTHER;
+ };
+ return Optional.of(new FinishReason(known));
+ }
+
+ private Optional toUsageMetadata(Usage usage) {
+ if (usage == null) {
+ return Optional.empty();
+ }
+ GenerateContentResponseUsageMetadata.Builder builder =
+ GenerateContentResponseUsageMetadata.builder()
+ .promptTokenCount(toIntTokenCount(usage.getPromptTokens()))
+ .candidatesTokenCount(toIntTokenCount(usage.getCompletionTokens()))
+ .totalTokenCount(toIntTokenCount(usage.getTotalTokens()));
+ if (usage.getPromptTokensDetails() != null
+ && usage.getPromptTokensDetails().getCachedTokens() != null) {
+ builder.cachedContentTokenCount(usage.getPromptTokensDetails().getCachedTokens());
+ }
+ if (usage.getCompletionTokensDetails() != null
+ && usage.getCompletionTokensDetails().getReasoningTokens() != null) {
+ builder.thoughtsTokenCount(usage.getCompletionTokensDetails().getReasoningTokens());
+ }
+ return Optional.of(builder.build());
+ }
+
+ private Integer toIntTokenCount(long value) {
+ return Math.toIntExact(Math.min(value, Integer.MAX_VALUE));
+ }
+
/**
* Convert ADK LlmRequest to Ark ChatCompletionRequest
* @param llmRequest The ADK request
* @return ChatCompletionRequest object for Ark API
*/
private ChatCompletionRequest toArkRequest(LlmRequest llmRequest) {
- // Determine model name to use
- String effectiveModelName = llmRequest.model().orElse(model());
+ return toArkRequest(llmRequest, llmRequest.model().orElse(model()));
+ }
+ private ChatCompletionRequest toArkRequest(LlmRequest llmRequest, String effectiveModelName) {
// Build chat messages from request
List messages = buildChatMessages(llmRequest);
@@ -367,6 +576,8 @@ private ChatCompletionRequest toArkRequest(LlmRequest llmRequest) {
request.setThinking(thinking);
}
+ llmRequest.config().ifPresent(config -> applyGenerateContentConfig(request, config));
+
// Add tools if any
if (llmRequest.tools() != null && !llmRequest.tools().isEmpty()) {
List chatTools = buildChatTools(llmRequest);
@@ -378,6 +589,73 @@ private ChatCompletionRequest toArkRequest(LlmRequest llmRequest) {
return request;
}
+ private void applyGenerateContentConfig(
+ ChatCompletionRequest request, GenerateContentConfig config) {
+ config.temperature().map(Float::doubleValue).ifPresent(request::setTemperature);
+ config.topP().map(Float::doubleValue).ifPresent(request::setTopP);
+ config.maxOutputTokens().ifPresent(request::setMaxTokens);
+ config.stopSequences().filter(stop -> !stop.isEmpty()).ifPresent(request::setStop);
+ config.presencePenalty().map(Float::doubleValue).ifPresent(request::setPresencePenalty);
+ config.frequencyPenalty().map(Float::doubleValue).ifPresent(request::setFrequencyPenalty);
+ config.candidateCount().ifPresent(request::setN);
+ config.responseLogprobs().ifPresent(request::setLogprobs);
+ config.logprobs()
+ .ifPresent(
+ topLogprobs -> {
+ request.setLogprobs(true);
+ request.setTopLogprobs(topLogprobs);
+ });
+ config.responseSchema()
+ .ifPresentOrElse(
+ schema -> request.setResponseFormat(buildJsonSchemaResponseFormat(schema)),
+ () ->
+ config.responseJsonSchema()
+ .ifPresentOrElse(
+ schema ->
+ request.setResponseFormat(
+ buildJsonSchemaResponseFormat(
+ schema)),
+ () ->
+ applyJsonObjectResponseFormat(
+ request, config)));
+ }
+
+ private ChatCompletionRequest.ChatCompletionRequestResponseFormat buildJsonSchemaResponseFormat(
+ Schema schema) {
+ Map schemaMap =
+ JSONUtil.convertValue(schema, new TypeReference>() {});
+ updateTypeString(schemaMap);
+ ResponseFormatJSONSchemaJSONSchemaParam jsonSchema =
+ new ResponseFormatJSONSchemaJSONSchemaParam(
+ schema.title().filter(StringUtils::isNotBlank).orElse("response_schema"),
+ schema.description().orElse(null),
+ JSONUtil.valueToTree(schemaMap),
+ true);
+ return new ChatCompletionRequest.ChatCompletionRequestResponseFormat(
+ "json_schema", jsonSchema);
+ }
+
+ private ChatCompletionRequest.ChatCompletionRequestResponseFormat buildJsonSchemaResponseFormat(
+ Object schema) {
+ ResponseFormatJSONSchemaJSONSchemaParam jsonSchema =
+ new ResponseFormatJSONSchemaJSONSchemaParam(
+ "response_schema", null, JSONUtil.valueToTree(schema), true);
+ return new ChatCompletionRequest.ChatCompletionRequestResponseFormat(
+ "json_schema", jsonSchema);
+ }
+
+ private void applyJsonObjectResponseFormat(
+ ChatCompletionRequest request, GenerateContentConfig config) {
+ config.responseMimeType()
+ .filter(mimeType -> "application/json".equalsIgnoreCase(mimeType))
+ .ifPresent(
+ unused ->
+ request.setResponseFormat(
+ new ChatCompletionRequest
+ .ChatCompletionRequestResponseFormat(
+ "json_object")));
+ }
+
/**
* Build chat messages from LlmRequest
* @param llmRequest The ADK request
@@ -415,12 +693,95 @@ private Stream buildSystemMessages(LlmRequest llmRequest) {
*/
private Stream buildContentMessages(LlmRequest llmRequest) {
return llmRequest.contents().stream()
- .map(
- content ->
- ChatMessage.builder()
- .role(toArkRole(content.role().orElse("user")))
- .content(extractText(content))
- .build());
+ .flatMap(content -> buildContentMessages(content).stream());
+ }
+
+ private List buildContentMessages(Content content) {
+ if (hasFunctionResponsePart(content)) {
+ return buildFunctionResponseMessages(content);
+ }
+ if (hasFunctionCallPart(content)) {
+ return List.of(buildFunctionCallMessage(content));
+ }
+ return List.of(buildContentMessage(content));
+ }
+
+ private ChatMessage buildContentMessage(Content content) {
+ ChatMessage.Builder builder =
+ ChatMessage.builder().role(toArkRole(content.role().orElse("user")));
+ if (hasSupportedMediaPart(content)) {
+ return builder.multiContent(extractContentParts(content)).build();
+ }
+ return builder.content(extractText(content)).build();
+ }
+
+ private boolean hasFunctionResponsePart(Content content) {
+ return content.parts().stream()
+ .flatMap(List::stream)
+ .filter(Objects::nonNull)
+ .anyMatch(part -> part.functionResponse().isPresent());
+ }
+
+ private boolean hasFunctionCallPart(Content content) {
+ return content.parts().stream()
+ .flatMap(List::stream)
+ .filter(Objects::nonNull)
+ .anyMatch(part -> part.functionCall().isPresent());
+ }
+
+ private List buildFunctionResponseMessages(Content content) {
+ return content.parts().stream()
+ .flatMap(List::stream)
+ .filter(Objects::nonNull)
+ .flatMap(part -> part.functionResponse().stream())
+ .map(this::buildFunctionResponseMessage)
+ .toList();
+ }
+
+ private ChatMessage buildFunctionResponseMessage(FunctionResponse functionResponse) {
+ ChatMessage.Builder builder =
+ ChatMessage.builder()
+ .role(ChatMessageRole.TOOL)
+ .content(functionResponse.response().map(JSONUtil::toJson).orElse("{}"));
+ functionResponse.id().filter(StringUtils::isNotBlank).ifPresent(builder::toolCallId);
+ functionResponse.name().filter(StringUtils::isNotBlank).ifPresent(builder::name);
+ return builder.build();
+ }
+
+ private ChatMessage buildFunctionCallMessage(Content content) {
+ ChatMessage.Builder builder = ChatMessage.builder().role(ChatMessageRole.ASSISTANT);
+ String text = extractText(content);
+ if (StringUtils.isNotEmpty(text)) {
+ builder.content(text);
+ }
+ List toolCalls =
+ content.parts().stream()
+ .flatMap(List::stream)
+ .filter(Objects::nonNull)
+ .flatMap(part -> part.functionCall().stream())
+ .map(this::toChatToolCall)
+ .filter(Optional::isPresent)
+ .map(Optional::get)
+ .toList();
+ if (!toolCalls.isEmpty()) {
+ builder.toolCalls(toolCalls);
+ }
+ return builder.build();
+ }
+
+ private Optional toChatToolCall(FunctionCall functionCall) {
+ Optional name = functionCall.name().filter(StringUtils::isNotBlank);
+ if (name.isEmpty()) {
+ return Optional.empty();
+ }
+ ChatFunctionCall chatFunctionCall = new ChatFunctionCall();
+ chatFunctionCall.setName(name.get());
+ chatFunctionCall.setArguments(functionCall.args().map(JSONUtil::toJson).orElse("{}"));
+ ChatToolCall chatToolCall = new ChatToolCall();
+ functionCall.id().filter(StringUtils::isNotBlank).ifPresent(chatToolCall::setId);
+ chatToolCall.setType("function");
+ chatToolCall.setFunction(chatFunctionCall);
+ return Optional.of(chatToolCall);
}
/**
@@ -442,27 +803,49 @@ private List buildChatTools(LlmRequest llmRequest) {
* @return Optional ChatTool object
*/
private Optional convertToChatTool(BaseTool tool) {
- // Get tool parameters schema
- Optional parameters = tool.declaration().flatMap(FunctionDeclaration::parameters);
- return parameters.map(
- schema -> {
- // Convert schema to map
- Map schemaMap =
- JSONUtil.convertValue(
- schema, new TypeReference>() {});
-
- // Normalize type strings in schema
- updateTypeString(schemaMap);
-
- // Create chat function
- ChatFunction chatFunction = new ChatFunction();
- chatFunction.setName(tool.name());
- chatFunction.setDescription(tool.description());
- chatFunction.setParameters(JSONUtil.valueToTree(schemaMap));
-
- // Return chat tool
- return new ChatTool("function", chatFunction);
- });
+ Optional declaration = tool.declaration();
+ if (declaration.isEmpty()) {
+ return Optional.empty();
+ }
+ Map schemaMap =
+ declaration
+ .get()
+ .parameters()
+ .map(
+ schema ->
+ JSONUtil.convertValue(
+ schema,
+ new TypeReference>() {}))
+ .orElseGet(
+ () ->
+ declaration
+ .get()
+ .parametersJsonSchema()
+ .map(
+ schema ->
+ JSONUtil.convertValue(
+ schema,
+ new TypeReference<
+ Map<
+ String,
+ Object>>() {}))
+ .orElseGet(
+ () -> {
+ Map emptySchema =
+ new LinkedHashMap<>();
+ emptySchema.put("type", "object");
+ emptySchema.put(
+ "properties",
+ new LinkedHashMap<>());
+ return emptySchema;
+ }));
+ updateTypeString(schemaMap);
+
+ ChatFunction chatFunction = new ChatFunction();
+ chatFunction.setName(tool.name());
+ chatFunction.setDescription(tool.description());
+ chatFunction.setParameters(JSONUtil.valueToTree(schemaMap));
+ return Optional.of(new ChatTool("function", chatFunction));
}
/**
@@ -499,11 +882,39 @@ private void updateTypeString(Map valueDict) {
}
}
- /**
- * Extract text content from Content object
- * @param content The Content object
- * @return Extracted text
- */
+ private boolean hasSupportedMediaPart(Content content) {
+ return content.parts().stream()
+ .flatMap(List::stream)
+ .filter(Objects::nonNull)
+ .anyMatch(
+ part ->
+ part.inlineData()
+ .flatMap(blob -> blob.mimeType())
+ .filter(this::isSupportedMediaMimeType)
+ .isPresent()
+ || part.fileData()
+ .flatMap(fileData -> fileData.mimeType())
+ .filter(this::isSupportedMediaMimeType)
+ .isPresent());
+ }
+
+ private List extractContentParts(Content content) {
+ List contentParts = new ArrayList<>();
+ content.parts()
+ .ifPresent(
+ parts -> {
+ for (Part part : parts) {
+ if (part == null) {
+ continue;
+ }
+ appendTextContentPart(part, contentParts);
+ appendInlineDataContentPart(part, contentParts);
+ appendFileDataContentPart(part, contentParts);
+ }
+ });
+ return contentParts;
+ }
+
private String extractText(Content content) {
StringBuilder textBuilder = new StringBuilder();
// Use ifPresent with a lambda for a more functional and readable style
@@ -523,6 +934,78 @@ private String extractText(Content content) {
return textBuilder.toString();
}
+ private void appendTextContentPart(Part part, List contentParts) {
+ part.text()
+ .filter(StringUtils::isNotEmpty)
+ .ifPresent(
+ text ->
+ contentParts.add(
+ ChatCompletionContentPart.builder()
+ .type("text")
+ .text(text)
+ .build()));
+ }
+
+ private void appendInlineDataContentPart(
+ Part part, List contentParts) {
+ part.inlineData()
+ .filter(blob -> blob.data().isPresent())
+ .flatMap(
+ blob ->
+ toMediaContentPart(
+ dataUri(
+ blob.mimeType().orElse("application/octet-stream"),
+ blob.data().get()),
+ blob.mimeType().orElse("application/octet-stream"),
+ part.videoMetadata()))
+ .ifPresent(contentParts::add);
+ }
+
+ private void appendFileDataContentPart(
+ Part part, List contentParts) {
+ part.fileData()
+ .filter(fileData -> fileData.fileUri().isPresent())
+ .flatMap(
+ fileData ->
+ toMediaContentPart(
+ fileData.fileUri().get(),
+ fileData.mimeType().orElse("application/octet-stream"),
+ part.videoMetadata()))
+ .ifPresent(contentParts::add);
+ }
+
+ private Optional toMediaContentPart(
+ String url, String mimeType, Optional videoMetadata) {
+ if (mimeType.startsWith("image/")) {
+ return Optional.of(
+ ChatCompletionContentPart.builder()
+ .type("image_url")
+ .imageUrl(
+ new ChatCompletionContentPart.ChatCompletionContentPartImageURL(
+ url, "auto"))
+ .build());
+ }
+ if (mimeType.startsWith("video/")) {
+ return Optional.of(
+ ChatCompletionContentPart.builder()
+ .type("video_url")
+ .videoUrl(
+ new ChatCompletionContentPart.ChatCompletionContentPartVideoURL(
+ url,
+ videoMetadata.flatMap(VideoMetadata::fps).orElse(1.0)))
+ .build());
+ }
+ return Optional.empty();
+ }
+
+ private boolean isSupportedMediaMimeType(String mimeType) {
+ return mimeType.startsWith("image/") || mimeType.startsWith("video/");
+ }
+
+ private String dataUri(String mimeType, byte[] data) {
+ return "data:" + mimeType + ";base64," + Base64.getEncoder().encodeToString(data);
+ }
+
/**
* Append text part to StringBuilder
* @param part The Part object
@@ -571,11 +1054,18 @@ private ChatMessageRole toArkRole(String adkRole) {
* @throws JsonProcessingException If JSON parsing fails
*/
private Part parseToolCallPart(ChatToolCall toolCall) throws JsonProcessingException {
- return Part.fromFunctionCall(
- toolCall.getFunction().getName(),
- JSONUtil.fromJson(
- toolCall.getFunction().getArguments(),
- new TypeReference>() {}));
+ ChatFunctionCall function = toolCall.getFunction();
+ Map args =
+ StringUtils.isBlank(function.getArguments())
+ ? Map.of()
+ : JSONUtil.fromJson(
+ function.getArguments(),
+ new TypeReference>() {});
+ FunctionCall.Builder builder = FunctionCall.builder().name(function.getName()).args(args);
+ if (StringUtils.isNotBlank(toolCall.getId())) {
+ builder.id(toolCall.getId());
+ }
+ return Part.builder().functionCall(builder.build()).build();
}
/**
diff --git a/core/src/main/java/com/volcengine/veadk/processors/BaseRunProcessor.java b/core/src/main/java/com/volcengine/veadk/processors/BaseRunProcessor.java
new file mode 100644
index 0000000..ed8c941
--- /dev/null
+++ b/core/src/main/java/com/volcengine/veadk/processors/BaseRunProcessor.java
@@ -0,0 +1,27 @@
+/**
+ * Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+ *
+ * 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 com.volcengine.veadk.processors;
+
+import com.google.adk.events.Event;
+import io.reactivex.rxjava3.core.Flowable;
+import java.util.function.Supplier;
+
+/** Intercepts an agent run and may observe, transform, retry, or replace its event stream. */
+@FunctionalInterface
+public interface BaseRunProcessor {
+
+ Flowable processRun(RunContext context, Supplier> eventGenerator);
+}
diff --git a/core/src/main/java/com/volcengine/veadk/processors/NoOpRunProcessor.java b/core/src/main/java/com/volcengine/veadk/processors/NoOpRunProcessor.java
new file mode 100644
index 0000000..d398e5c
--- /dev/null
+++ b/core/src/main/java/com/volcengine/veadk/processors/NoOpRunProcessor.java
@@ -0,0 +1,34 @@
+/**
+ * Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+ *
+ * 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 com.volcengine.veadk.processors;
+
+import com.google.adk.events.Event;
+import io.reactivex.rxjava3.core.Flowable;
+import java.util.function.Supplier;
+
+/** Default run processor that preserves the original event stream. */
+public final class NoOpRunProcessor implements BaseRunProcessor {
+
+ public static final NoOpRunProcessor INSTANCE = new NoOpRunProcessor();
+
+ private NoOpRunProcessor() {}
+
+ @Override
+ public Flowable processRun(
+ RunContext context, Supplier> eventGenerator) {
+ return Flowable.defer(eventGenerator::get);
+ }
+}
diff --git a/core/src/main/java/com/volcengine/veadk/processors/RunContext.java b/core/src/main/java/com/volcengine/veadk/processors/RunContext.java
new file mode 100644
index 0000000..fd26060
--- /dev/null
+++ b/core/src/main/java/com/volcengine/veadk/processors/RunContext.java
@@ -0,0 +1,57 @@
+/**
+ * Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+ *
+ * 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 com.volcengine.veadk.processors;
+
+import com.google.adk.agents.RunConfig;
+import com.google.adk.sessions.Session;
+import com.google.genai.types.Content;
+import com.volcengine.veadk.runner.Runner;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Objects;
+
+/** Immutable inputs made available to a {@link BaseRunProcessor}. */
+public record RunContext(
+ Runner runner,
+ Session session,
+ Content message,
+ RunConfig runConfig,
+ Map invocationState) {
+
+ public RunContext {
+ Objects.requireNonNull(runner, "runner");
+ Objects.requireNonNull(session, "session");
+ Objects.requireNonNull(message, "message");
+ Objects.requireNonNull(runConfig, "runConfig");
+ invocationState =
+ invocationState == null
+ ? Map.of()
+ : Collections.unmodifiableMap(new LinkedHashMap<>(invocationState));
+ }
+
+ public String appName() {
+ return session.appName();
+ }
+
+ public String userId() {
+ return session.userId();
+ }
+
+ public String sessionId() {
+ return session.id();
+ }
+}
diff --git a/core/src/main/java/com/volcengine/veadk/runner/Runner.java b/core/src/main/java/com/volcengine/veadk/runner/Runner.java
index ca554ff..cec1de5 100644
--- a/core/src/main/java/com/volcengine/veadk/runner/Runner.java
+++ b/core/src/main/java/com/volcengine/veadk/runner/Runner.java
@@ -16,20 +16,36 @@
package com.volcengine.veadk.runner;
import com.google.adk.agents.BaseAgent;
+import com.google.adk.agents.RunConfig;
+import com.google.adk.artifacts.BaseArtifactService;
import com.google.adk.artifacts.InMemoryArtifactService;
+import com.google.adk.events.Event;
import com.google.adk.memory.BaseMemoryService;
import com.google.adk.memory.InMemoryMemoryService;
+import com.google.adk.plugins.BasePlugin;
+import com.google.adk.sessions.BaseSessionService;
import com.google.adk.sessions.InMemorySessionService;
-import com.google.common.collect.ImmutableList;
+import com.google.adk.sessions.Session;
+import com.google.genai.types.Content;
+import com.volcengine.veadk.agent.Agent;
+import com.volcengine.veadk.processors.BaseRunProcessor;
+import com.volcengine.veadk.processors.NoOpRunProcessor;
+import com.volcengine.veadk.processors.RunContext;
+import io.reactivex.rxjava3.core.Flowable;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
public class Runner extends com.google.adk.runner.Runner {
+ private final BaseRunProcessor runProcessor;
+
public Runner(BaseAgent agent) {
this(agent, agent.name());
}
public Runner(BaseAgent agent, String appName) {
- this(agent, appName, null);
+ this(agent, appName, (BaseMemoryService) null);
}
public Runner(BaseAgent agent, BaseMemoryService baseMemoryService) {
@@ -37,12 +53,104 @@ public Runner(BaseAgent agent, BaseMemoryService baseMemoryService) {
}
public Runner(BaseAgent agent, String appName, BaseMemoryService baseMemoryService) {
- super(
+ this(
agent,
appName,
new InMemoryArtifactService(),
- new InMemorySessionService(),
- null != baseMemoryService ? baseMemoryService : new InMemoryMemoryService(),
- ImmutableList.of());
+ defaultSessionService(agent),
+ null != baseMemoryService ? baseMemoryService : defaultMemoryService(agent),
+ defaultPlugins(agent),
+ defaultRunProcessor(agent));
+ }
+
+ public static Runner withProcessor(BaseAgent agent, BaseRunProcessor runProcessor) {
+ return withProcessor(agent, agent.name(), runProcessor);
+ }
+
+ public static Runner withProcessor(
+ BaseAgent agent, String appName, BaseRunProcessor runProcessor) {
+ return new Runner(
+ agent,
+ appName,
+ new InMemoryArtifactService(),
+ defaultSessionService(agent),
+ defaultMemoryService(agent),
+ defaultPlugins(agent),
+ runProcessor);
+ }
+
+ public Runner(
+ BaseAgent agent,
+ String appName,
+ BaseArtifactService artifactService,
+ BaseSessionService sessionService,
+ BaseMemoryService memoryService,
+ List plugins) {
+ this(
+ agent,
+ appName,
+ artifactService,
+ sessionService,
+ memoryService,
+ plugins,
+ NoOpRunProcessor.INSTANCE);
+ }
+
+ public Runner(
+ BaseAgent agent,
+ String appName,
+ BaseArtifactService artifactService,
+ BaseSessionService sessionService,
+ BaseMemoryService memoryService,
+ List