diff --git a/py/genai_client/message_builders/semoss_base/document_input.py b/py/genai_client/message_builders/semoss_base/document_input.py index 8490cd1dda2..316bf9b9d4a 100644 --- a/py/genai_client/message_builders/semoss_base/document_input.py +++ b/py/genai_client/message_builders/semoss_base/document_input.py @@ -38,6 +38,10 @@ class DocumentInputError(ValueError): """An attachment could not be delivered without losing its contents.""" +class ContextBudgetError(DocumentInputError): + """The request is too large for the model's context; callers may compact and retry.""" + + @dataclass(frozen=True) class ExtractedDocument: text: str @@ -207,7 +211,7 @@ def prepare( message.type = SEMOSSMessageType.INPUT_TEXT if extracted_chars > self.max_text_chars: - raise DocumentInputError( + raise ContextBudgetError( "The attached documents contain too much text for one conversation " f"({self.max_text_chars:,} character limit). Use fewer pages or files." ) @@ -472,7 +476,7 @@ def _check_context_budget( ] text_bytes += len(json.dumps(payload, ensure_ascii=False).encode("utf-8")) if text_bytes + int(output_tokens) > settings.context_window: - raise DocumentInputError( + raise ContextBudgetError( "The extracted documents and conversation exceed the conservative " "text budget for this model. Attach fewer pages, start a new conversation, " "or reduce the requested output length. No content was truncated." diff --git a/py/genai_client/text_generation/model_engine_exception.py b/py/genai_client/text_generation/model_engine_exception.py index 68b2f6aa95d..61d63597889 100644 --- a/py/genai_client/text_generation/model_engine_exception.py +++ b/py/genai_client/text_generation/model_engine_exception.py @@ -1,10 +1,25 @@ -from typing import Any +from typing import Any, Optional import traceback, re, json import openai from pydantic import BaseModel from anthropic import APIStatusError, APIConnectionError, APITimeoutError from botocore.exceptions import ClientError, BotoCoreError +from ..message_builders.semoss_base.document_input import ContextBudgetError + +# reason Java checks to compact/fork and retry instead of failing the room +CONTEXT_OVERFLOW = "CONTEXT_OVERFLOW" + +# provider wording for "request exceeds the context window" (lowercase) +_CONTEXT_OVERFLOW_PATTERNS = ( + "maximum context length", # openai / vllm / openai-compatible servers + "prompt is too long", # anthropic + "input is too long", # bedrock + "too many input tokens", # bedrock + "exceeds the maximum number of tokens", # google + "input token count", # google +) + class AnthropicRefusalError(RuntimeError): """Raised when Anthropic returns stop_reason='refusal'.""" @@ -18,6 +33,9 @@ class ErrorDetails(BaseModel): client: str model: str traceback: str + reason: Optional[str] = None + # which rule set reason, e.g. "code:context_length_exceeded" or "text:prompt is too long" + reason_detail: Optional[str] = None class ModelEngineException: @@ -28,6 +46,14 @@ def __init__(self, error: Any, client: str, model: str): self.traceback = traceback.format_exc() def parse_error(self) -> ErrorDetails: + details = self._parse_by_client() + rule = self._context_overflow_rule(details) + if rule: + details.reason = CONTEXT_OVERFLOW + details.reason_detail = rule + return details + + def _parse_by_client(self) -> ErrorDetails: if self.client == "anthropic": return self._parse_anthropic_error() elif self.client in ["google", "vertex", "gemini"]: @@ -44,6 +70,23 @@ def parse_error(self) -> ErrorDetails: error_type="Internal Server Error", ) + def _context_overflow_rule(self, details: ErrorDetails) -> Optional[str]: + # structured signals first, provider wording as the fallback + if isinstance(self.error, ContextBudgetError): + return "type:ContextBudgetError" + if getattr(self.error, "code", None) == "context_length_exceeded": + return "code:context_length_exceeded" # openai / azure + if isinstance(self.error, APIStatusError): + body = self.error.body if isinstance(self.error.body, dict) else {} + inner = body.get("error") + if isinstance(inner, dict) and inner.get("type") == "request_too_large": + return "type:request_too_large" # anthropic 413, byte limit + text = f"{details.error_type} {details.message}".lower() + for pattern in _CONTEXT_OVERFLOW_PATTERNS: + if pattern in text: + return f"text:{pattern}" + return None + def _parse_google_error(self) -> ErrorDetails: """ Handles Google GenAI errors which often look like: diff --git a/src/prerna/engine/api/IModelEngine.java b/src/prerna/engine/api/IModelEngine.java index ade14a53043..c036d7abe15 100644 --- a/src/prerna/engine/api/IModelEngine.java +++ b/src/prerna/engine/api/IModelEngine.java @@ -163,6 +163,14 @@ default MultiModalEmbeddingsModelEngineResponse multiModalEmbeddings(List results = new ArrayList<>(); + for (MessagePart part : fork.getParts()) { + if (part instanceof ToolResultMessagePart trPart && trPart.getToolResult() != null + && trPart.getToolResult().getOutput() != null + && trPart.getToolResult().getOutput().length() >= minStubChars) { + results.add(trPart.getToolResult()); + } + } + results.sort((a, b) -> Integer.compare(b.getOutput().length(), a.getOutput().length())); + int replaced = 0; + long freed = 0; + for (ToolResultPart result : results) { + if (replaced > 0 && freed >= tokensToFree) { + break; + } + int chars = result.getOutput().length(); + result.setOutput("[tool result too large for the model context: " + chars + + " chars, not sent. Re-run with a narrower query or read it in parts" + + " (e.g. ReadFile offset/limit).]"); + freed += chars / 4; + replaced++; + } + if (replaced == 0) { + return 0; + } + fork.setMessageId(GUID.v7().toUUID().toString()); + fork.setParentMessageId(original.getParentMessageId()); + fork.setTokensInMessage(0); + fork.setRoom(this); + messages.add(fork); + RoomMessageStore.persist(this, userId); + classLogger.info("Forked tool results room={} original={} fork={} replaced={} freedTokens~{}", getId(), + original.getMessageId(), fork.getMessageId(), replaced, freed); + return replaced; + } + } finally { + lock.unlock(); + } + } + /** * Persist a harness-owned completion after all tools finish, without another * model request. diff --git a/src/prerna/engine/impl/model/message/MessageUtils.java b/src/prerna/engine/impl/model/message/MessageUtils.java index f3fdaec5aa5..e9fb9e58f05 100644 --- a/src/prerna/engine/impl/model/message/MessageUtils.java +++ b/src/prerna/engine/impl/model/message/MessageUtils.java @@ -740,9 +740,30 @@ public static List getMessageBranchFromParent(List T deepCopy(T m) { + @SuppressWarnings("unchecked") + T copy = (T) GSON_FOR_DB.fromJson(GSON_FOR_DB.toJson(m), m.getClass()); + copy.roomId = m.roomId; + copy.roomFolderPath = m.roomFolderPath; + // base64Data and the media room folder are not serialized; without them the + // provider gets attachments with no data + List from = m.getParts(); + List to = copy.getParts(); + if (from != null && to != null && from.size() == to.size()) { + for (int i = 0; i < from.size(); i++) { + if (from.get(i) instanceof MediaMessagePart src && to.get(i) instanceof MediaMessagePart dst + && src.getMediaInfo() != null && dst.getMediaInfo() != null) { + dst.getMediaInfo().setBase64Data(src.getMediaInfo().getBase64Data()); + } + } + } + return copy; + } + + // only messages with tool parts change, so the rest stay as-is private static AbstractMessage deepCopyForPruning(AbstractMessage m) { - String json = GSON_FOR_DB.toJson(m); - return GSON_FOR_DB.fromJson(json, m.getClass()); + return m.hasToolResultPart() || m.hasToolCallPart() ? deepCopy(m) : m; } private static void pruneToolsPartFromMessage(AbstractMessage m) { diff --git a/src/prerna/engine/impl/model/responses/AskErrorModelEngineResponse.java b/src/prerna/engine/impl/model/responses/AskErrorModelEngineResponse.java index 8e61170abc8..de23dc0a63c 100644 --- a/src/prerna/engine/impl/model/responses/AskErrorModelEngineResponse.java +++ b/src/prerna/engine/impl/model/responses/AskErrorModelEngineResponse.java @@ -36,12 +36,18 @@ public class AskErrorModelEngineResponse extends AskModelEngineResponse public static final String CLIENT = "client"; public static final String MODEL = "model"; public static final String TRACEBACK = "traceback"; + public static final String REASON = "reason"; + public static final String REASON_DETAIL = "reason_detail"; + // request did not fit the model context; callers may compact/fork and retry + public static final String REASON_CONTEXT_OVERFLOW = "CONTEXT_OVERFLOW"; protected String errorType; protected int code; protected String client; protected String model; protected String traceback; + protected String reason; + protected String reasonDetail; public AskErrorModelEngineResponse(String message, String errorType, int code, String client, String model, String traceback) { super(message, 0, 0); @@ -66,6 +72,16 @@ public String getStringResponse() { public String getTraceback() { return this.traceback; } + public String getReason() { return this.reason; } + + public void setReason(String reason) { this.reason = reason; } + + public String getReasonDetail() { return this.reasonDetail; } + + public void setReasonDetail(String reasonDetail) { this.reasonDetail = reasonDetail; } + + public boolean isContextOverflow() { return REASON_CONTEXT_OVERFLOW.equals(this.reason); } + @Override public Map toMap() { Map map = super.toMap(); @@ -74,6 +90,8 @@ public Map toMap() { map.put(CLIENT, this.client); map.put(MODEL, this.model); map.put(TRACEBACK, this.traceback); + map.put(REASON, this.reason); + map.put(REASON_DETAIL, this.reasonDetail); return map; } } \ No newline at end of file diff --git a/src/prerna/engine/impl/model/responses/AskModelEngineResponse.java b/src/prerna/engine/impl/model/responses/AskModelEngineResponse.java index 20671822c18..8682eaab85b 100644 --- a/src/prerna/engine/impl/model/responses/AskModelEngineResponse.java +++ b/src/prerna/engine/impl/model/responses/AskModelEngineResponse.java @@ -231,8 +231,10 @@ public static AskModelEngineResponse fromMap(Object responseObject) { String model = safeString(modelResponse.get("model")); String traceback = safeString(modelResponse.get("traceback")); - AskModelEngineResponse errorResponse = new AskErrorModelEngineResponse(message, errorType, code, client, - model, traceback); + AskErrorModelEngineResponse errorResponse = new AskErrorModelEngineResponse(message, errorType, code, + client, model, traceback); + errorResponse.setReason(safeString(modelResponse.get(AskErrorModelEngineResponse.REASON))); + errorResponse.setReasonDetail(safeString(modelResponse.get(AskErrorModelEngineResponse.REASON_DETAIL))); if (schemaVersion != null) { errorResponse.setSchemaVersion(schemaVersion); } diff --git a/src/prerna/reactor/agent/runtime/HarnessToolExecutor.java b/src/prerna/reactor/agent/runtime/HarnessToolExecutor.java index 5820fc66cab..ba2d7fe0853 100644 --- a/src/prerna/reactor/agent/runtime/HarnessToolExecutor.java +++ b/src/prerna/reactor/agent/runtime/HarnessToolExecutor.java @@ -77,6 +77,8 @@ import prerna.sablecc2.om.GenRowStruct; import prerna.sablecc2.om.PixelDataType; import prerna.sablecc2.om.ReactorKeysEnum; +import prerna.engine.impl.model.responses.AskErrorModelEngineResponse; +import prerna.sablecc2.om.execptions.SemossModelEngineException; import prerna.sablecc2.om.nounmeta.NounMetadata; /** @@ -194,6 +196,8 @@ static ResponseMessage executeToolBatch(ResponseMessage toolResponse, AgentLoopS return (ResponseMessage) room.getMessages().getLast(); } } + // tool results are stored; prune older tool payloads if the context is near full + SemossAgentHarness.pruneToolsBeforeContinuation(ctx, parentMsgId); Map nextParams = new HashMap<>(paramMap); if (ctx.getAgentConfig().getFinishingTurns() > 0 && state.getIterations() >= ctx.getMaxTurns()) { nextParams.put("tool_choice", "none"); @@ -201,8 +205,23 @@ static ResponseMessage executeToolBatch(ResponseMessage toolResponse, AgentLoopS AgentRunStreamService.get().beginModelCall(ctx.getRunId()); state.progress().beginModel(); try { - nextModelResp = room.continueAfterToolExecutionResultsWithRuntimeContext(nextParams, parentMsgId, - ctx.getModelEngine(), ctx.getInsight(), state.systemPrompt(), state.runtimeContext()); + try { + nextModelResp = room.continueAfterToolExecutionResultsWithRuntimeContext(nextParams, parentMsgId, + ctx.getModelEngine(), ctx.getInsight(), state.systemPrompt(), state.runtimeContext()); + } catch (RuntimeException e) { + // too big for the context: fork with the largest results stubbed and retry once + AskErrorModelEngineResponse overflow = SemossModelEngineException.contextOverflowError(e); + if (overflow == null) { + throw e; + } + logger.warn("HarnessToolExecutor: context overflow after tool batch room={} client={} rule={}", + room.getId(), overflow.getClient(), overflow.getReasonDetail()); + if (SemossAgentHarness.forkOversizedToolResults(ctx, parentMsgId) == 0) { + throw e; + } + nextModelResp = room.continueAfterToolExecutionResultsWithRuntimeContext(nextParams, parentMsgId, + ctx.getModelEngine(), ctx.getInsight(), state.systemPrompt(), state.runtimeContext()); + } } finally { state.progress().endModel(); } diff --git a/src/prerna/reactor/agent/runtime/PlatformAgentToolHandlers.java b/src/prerna/reactor/agent/runtime/PlatformAgentToolHandlers.java index 2e2605c8c52..378be71c965 100644 --- a/src/prerna/reactor/agent/runtime/PlatformAgentToolHandlers.java +++ b/src/prerna/reactor/agent/runtime/PlatformAgentToolHandlers.java @@ -30,6 +30,9 @@ import java.io.File; import java.io.RandomAccessFile; import java.math.BigDecimal; +import java.nio.ByteBuffer; +import java.nio.charset.CharacterCodingException; +import java.nio.charset.CodingErrorAction; import java.nio.charset.StandardCharsets; import java.nio.file.FileSystems; import java.nio.file.Files; @@ -45,6 +48,7 @@ import java.util.HashSet; import java.util.LinkedHashMap; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Set; import java.util.regex.Matcher; @@ -52,6 +56,7 @@ import java.util.regex.PatternSyntaxException; import java.util.stream.Stream; +import org.apache.commons.io.FilenameUtils; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import org.json.JSONArray; @@ -84,6 +89,7 @@ final class PlatformAgentToolHandlers { private static final String TOOL_KIND = "semoss_platform_default"; private static final int DEFAULT_READ_MAX_LINES = 2000; + private static final List DOCLING_READ_EXTENSIONS = List.of("pdf", "docx", "pptx", "xlsx"); private static final int MAX_GLOB_RESULTS = 500; private static final int DEFAULT_GREP_HEAD_LIMIT = 200; private static final int MAX_MULTI_EDITS = 200; @@ -140,7 +146,8 @@ static Map handlersByName() { add(tools, handler("ReadFile", """ Reads a file from the working directory. Returns content with line numbers \ - and a continuation marker when more lines remain.\ + and a continuation marker when more lines remain. PDF, DOCX, PPTX and XLSX files \ + are returned as extracted Markdown text; other binary files are rejected.\ """, objectSchema( props(prop(PARAM_PATH, stringProp("Path to read, relative to the working directory.")), @@ -359,7 +366,25 @@ private static String readFile(Map params, ToolContext tc) throw if (!file.isFile()) { return "Error: not a file: " + filePath; } - String content = new String(Files.readAllBytes(file.toPath()), StandardCharsets.UTF_8); + String content; + String extension = FilenameUtils.getExtension(file.getName()).toLowerCase(Locale.ROOT); + if (DOCLING_READ_EXTENSIONS.contains(extension)) { + // office/pdf files are zips or binary; read them as markdown via docling + try { + content = extractDocumentMarkdown(file, tc); + } catch (Exception e) { + logger.warn("ReadFile document extraction failed for " + filePath, e); + return "Error: could not extract text from " + filePath + ": " + lastLine(e.getMessage()); + } + } else { + byte[] bytes = Files.readAllBytes(file.toPath()); + content = decodeTextOrNull(bytes); + if (content == null) { + return "Error: " + filePath + " is a binary file (" + bytes.length + + " bytes); ReadFile only returns text. Supported documents: " + + String.join(", ", DOCLING_READ_EXTENSIONS) + "."; + } + } List lines = content.lines().toList(); int start = offset - 1; if (start >= lines.size()) { @@ -377,6 +402,59 @@ private static String readFile(Map params, ToolContext tc) throw return sb.toString(); } + // strict utf-8 decode; null for binary (nul bytes or invalid utf-8) + private static String decodeTextOrNull(byte[] bytes) { + for (int i = 0, n = Math.min(bytes.length, 8192); i < n; i++) { + if (bytes[i] == 0) { + return null; + } + } + try { + return StandardCharsets.UTF_8.newDecoder().onMalformedInput(CodingErrorAction.REPORT) + .onUnmappableCharacter(CodingErrorAction.REPORT).decode(ByteBuffer.wrap(bytes)).toString(); + } catch (CharacterCodingException e) { + return null; + } + } + + // runs smssutil.get_document_markdown in the room's python process + private static String extractDocumentMarkdown(File file, ToolContext tc) throws Exception { + if (!isPythonToolEnabled()) { + throw new IllegalStateException("Document extraction requires Python, which is disabled on this instance."); + } + if (tc.ctx.getInsight() == null || tc.ctx.getInsight().getUser() == null) { + throw new IllegalStateException("no user is associated with this agent run"); + } + // pass the path as base64 so no quoting/escaping can break the script + String encodedPath = Base64.getEncoder() + .encodeToString(file.getAbsolutePath().getBytes(StandardCharsets.UTF_8)); + String script = "from smssutil import get_document_markdown as _smss_doc_md\n" + + "import base64 as _smss_b64\n" + + "print(_smss_doc_md(_smss_b64.b64decode('" + encodedPath + "').decode('utf-8')), end='')"; + Insight executionInsight = tc.ctx.getInsight(); + User user = executionInsight.getUser(); + Object output = AgentCodeExecutionContext.executeForRoom(tc.ctx.getRoom(), executionInsight, tc.root, + roomInsight -> { + SocketClient sc = user.getPythonSocketClient(true); + PyTranslator translator = new PyTranslator(sc, roomInsight); + try { + return translator.runScript(script); + } catch (RuntimeException e) { + interruptRoomExecutionIfCancelled(sc, roomInsight, tc.ctx); + throw e; + } + }); + return output == null ? "" : output.toString(); + } + + private static String lastLine(String message) { + if (message == null || message.isBlank()) { + return "extraction failed"; + } + String[] lines = message.strip().split("\\R"); + return lines[lines.length - 1].strip(); + } + private static String writeFile(Map params, ToolContext tc) { String filePath = stringParam(params, PARAM_PATH); String content = stringParam(params, "content"); diff --git a/src/prerna/reactor/agent/runtime/SemossAgentHarness.java b/src/prerna/reactor/agent/runtime/SemossAgentHarness.java index 1f24c97b1b9..35ba2c234b3 100644 --- a/src/prerna/reactor/agent/runtime/SemossAgentHarness.java +++ b/src/prerna/reactor/agent/runtime/SemossAgentHarness.java @@ -80,6 +80,8 @@ import prerna.sablecc2.om.GenRowStruct; import prerna.sablecc2.om.PixelDataType; import prerna.sablecc2.om.ReactorKeysEnum; +import prerna.engine.impl.model.responses.AskErrorModelEngineResponse; +import prerna.sablecc2.om.execptions.SemossModelEngineException; import prerna.sablecc2.om.nounmeta.NounMetadata; /** @@ -112,7 +114,11 @@ public class SemossAgentHarness implements IAgentHarness { /** Harness-only paramMap key stripped before provider model calls. */ public static final String PARAM_MAX_SECONDS = "max_seconds"; - private static final double AUTO_COMPACTION_TRIGGER_RATIO = 0.80; + // compaction triggers when estimate + reserve reaches the window; reserve = + // reply allowance + margin for the next tool round and estimate error + private static final int COMPACTION_MARGIN_TOKENS = 20_000; + private static final int DEFAULT_MAX_OUTPUT_TOKENS = 16_000; + private static final double MAX_RESERVE_RATIO = 0.25; private static final String PARAM_FILE_PATH = "file_path"; private static final String PARAM_FILE_PATH_CAMEL = "filePath"; @@ -365,7 +371,7 @@ public AgentHarnessResult execute(AgentRunContext ctx) throws Exception { AgentRunStreamService.get().beginModelCall(ctx.getRunId()); state.progress().beginModel(); try { - response = requireModelResponse(room.ask(firstMsg, ctx.getModelEngine(), null), + response = requireModelResponse(askWithOverflowRecovery(ctx, firstMsg), "during initial model call"); } finally { state.progress().endModel(); @@ -579,12 +585,23 @@ private static AutoCompactionOutcome autoCompactIfNeeded(AgentRunContext ctx, bo AbstractMessage leaf = messages.getLast(); List branch = RoomMessageStore .providerContext(MessageUtils.getMessageBranchFromParent(messages, leaf.getMessageId())); - int contextTokens = currentContextTokens(branch); - double usageRatio = (double) contextTokens / contextWindow; - if (usageRatio < AUTO_COMPACTION_TRIGGER_RATIO) { + int contextTokens = estimateContextTokens(branch); + if (!needsCompaction(ctx, contextTokens, contextWindow)) { return AutoCompactionOutcome.NOT_NEEDED; } + // a failed run left unanswered tool results as the leaf; prune tool payloads + // from it up so the next request is not stuck on them + if (leaf instanceof InputMessage && leaf.hasToolResultPart()) { + boolean pruned = room.markPruneToolsAbove(leaf.getMessageId(), ctx.getInsight()); + logger.info( + "SemossAgentHarness: pruned tool results under unanswered leaf run={} room={} model={} " + + "estimatedTokens={} contextWindow={} pruned={}", + ctx.getRunId(), room.getId(), ctx.getModelEngine().getEngineId(), contextTokens, contextWindow, + pruned); + return pruned ? AutoCompactionOutcome.COMPACTED : AutoCompactionOutcome.SKIPPED; + } + if (leaf instanceof InputMessage || leaf.hasToolCallPart()) { String reason = leaf instanceof InputMessage ? "the active leaf is an input message" : "the active leaf has unresolved tool calls"; @@ -607,7 +624,7 @@ private static AutoCompactionOutcome autoCompactIfNeeded(AgentRunContext ctx, bo for (Map result : results) { if (Boolean.TRUE.equals(result.get("success"))) { - int tokensAfter = currentContextTokens(MessageUtils.getMessageBranchFromParent(room.getMessages(), + int tokensAfter = estimateContextTokens(MessageUtils.getMessageBranchFromParent(room.getMessages(), room.getMessages().getLast().getMessageId())); logger.info( "SemossAgentHarness: auto compaction completed run={} room={} model={} type={} " @@ -656,11 +673,161 @@ private static void addReactorStringInput(CompactRoomMessagesReactor reactor, St reactor.getNounStore().addNoun(key, row); } - private static int currentContextTokens(List branch) { - if (branch == null || branch.size() < 2) { + static boolean needsCompaction(AgentRunContext ctx, int contextTokens, int contextWindow) { + return (long) contextTokens + compactionReserve(ctx, contextWindow) >= contextWindow; + } + + // reply allowance (request max_tokens, else engine max, else default) + margin, + // capped so small windows do not compact every round + static int compactionReserve(AgentRunContext ctx, int contextWindow) { + long maxOutput = 0; + Map params = ctx.getAgentConfig().getModelParams(); + Object requested = params == null ? null : params.get("max_tokens"); + if (requested instanceof Number n) { + maxOutput = n.longValue(); + } else if (requested != null) { + try { + maxOutput = Long.parseLong(requested.toString().trim()); + } catch (NumberFormatException e) { + // fall through to engine/default + } + } + if (maxOutput <= 0) { + maxOutput = ctx.getModelEngine().getMaxTokens(); + } + if (maxOutput <= 0) { + maxOutput = DEFAULT_MAX_OUTPUT_TOKENS; + } + long reserve = maxOutput + COMPACTION_MARGIN_TOKENS; + return (int) Math.min(reserve, (long) (contextWindow * MAX_RESERVE_RATIO)); + } + + /** + * First model call of a run. On a context overflow, prune tool payloads in the + * stored history and retry once; if there is nothing to prune or it still + * overflows, the new message itself is too large. The input is only stored on + * success, so the room is never left stuck. + */ + private static ResponseMessage askWithOverflowRecovery(AgentRunContext ctx, InputMessage firstMsg) { + Room room = ctx.getRoom(); + try { + return room.ask(firstMsg, ctx.getModelEngine(), null); + } catch (RuntimeException e) { + AskErrorModelEngineResponse overflow = SemossModelEngineException.contextOverflowError(e); + if (overflow == null) { + throw e; + } + List messages = room.getMessages(); + AbstractMessage leaf = messages == null || messages.isEmpty() ? null : messages.getLast(); + boolean historyHasTools = leaf != null && MessageUtils + .getMessageBranchFromParent(messages, leaf.getMessageId()).stream() + .anyMatch(m -> m.hasToolResultPart() || m.hasToolCallPart()); + boolean pruned = historyHasTools && room.markPruneToolsAbove(leaf.getMessageId(), ctx.getInsight()); + logger.warn("SemossAgentHarness: initial ask context overflow run={} room={} client={} rule={} " + + "prunedHistory={}", ctx.getRunId(), room.getId(), overflow.getClient(), + overflow.getReasonDetail(), pruned); + if (pruned) { + try { + return room.ask(firstMsg, ctx.getModelEngine(), null); + } catch (RuntimeException retryError) { + if (!SemossModelEngineException.isContextOverflow(retryError)) { + throw retryError; + } + e = retryError; + } + } + throw new IllegalStateException("This message and its attachments are too large for the model's " + + "context window. Send less text, fewer pages or smaller files, or start a new conversation.", e); + } + } + + // results smaller than this are never stubbed; they are not what overflows the context + private static final int MIN_STUB_RESULT_CHARS = 2_000; + + /** + * After a context-overflow error on a tool continuation: fork the current + * tool-results message with the largest results stubbed until estimate + + * reserve fits. Returns how many results were replaced (0 = nothing to retry). + */ + static int forkOversizedToolResults(AgentRunContext ctx, String toolCallMessageId) { + Room room = ctx.getRoom(); + long tokensToFree = 0; + try { + int contextWindow = ctx.getModelEngine().getContextWindow(); + List messages = room.getMessages(); + if (contextWindow > 0 && messages != null && !messages.isEmpty()) { + int estimate = estimateContextTokens(RoomMessageStore.providerContext( + MessageUtils.getMessageBranchFromParent(messages, messages.getLast().getMessageId()))); + tokensToFree = Math.max(0L, + (long) estimate + compactionReserve(ctx, contextWindow) - contextWindow); + } + } catch (RuntimeException e) { + // unknown window; the fork still stubs the largest result + } + int replaced = room.forkToolResultsWithStubs(toolCallMessageId, tokensToFree, MIN_STUB_RESULT_CHARS, + ctx.getInsight()); + logger.info("SemossAgentHarness: context overflow fork run={} room={} toolCall={} tokensToFree={} replaced={}", + ctx.getRunId(), room.getId(), toolCallMessageId, tokensToFree, replaced); + return replaced; + } + + // last provider-measured prompt plus everything after it; unmeasured messages + // (e.g. tool results not yet sent) are estimated at chars / 4 + static int estimateContextTokens(List branch) { + if (branch == null || branch.isEmpty()) { return 0; } - return branch.getLast().getTokensInMessage() + branch.get(branch.size() - 2).getTokensInMessage(); + long total = 0; + for (int i = branch.size() - 1; i >= 0; i--) { + AbstractMessage message = branch.get(i); + int tokens = message.getTokensInMessage(); + // an input's recorded tokens are the full prompt at that call + if (message instanceof InputMessage && tokens > 0) { + total += tokens; + break; + } + total += tokens > 0 ? tokens : MessageUtils.toJson(message).length() / 4; + } + return (int) Math.min(Integer.MAX_VALUE, total); + } + + /** + * In-loop check after a tool batch's results are stored and before the next + * model call. Over the trigger ratio, older tool payloads are pruned from the + * provider view; the current tool call and its results stay intact. + */ + static void pruneToolsBeforeContinuation(AgentRunContext ctx, String toolCallMessageId) { + Room room = ctx.getRoom(); + int contextWindow; + try { + contextWindow = ctx.getModelEngine().getContextWindow(); + } catch (RuntimeException e) { + return; + } + List messages = room.getMessages(); + if (contextWindow <= 0 || messages == null || messages.isEmpty()) { + return; + } + List branch = RoomMessageStore + .providerContext(MessageUtils.getMessageBranchFromParent(messages, messages.getLast().getMessageId())); + int contextTokens = estimateContextTokens(branch); + if (!needsCompaction(ctx, contextTokens, contextWindow)) { + return; + } + // flag the message above the current tool call so pruning starts there + String pruneFromId = null; + for (AbstractMessage message : branch) { + if (toolCallMessageId.equals(message.getMessageId())) { + pruneFromId = message.getParentMessageId(); + break; + } + } + boolean pruned = pruneFromId != null && !pruneFromId.isEmpty() + && room.markPruneToolsAbove(pruneFromId, ctx.getInsight()); + logger.info( + "SemossAgentHarness: in-loop tool pruning run={} room={} model={} estimatedTokens={} " + + "contextWindow={} pruned={}", + ctx.getRunId(), room.getId(), ctx.getModelEngine().getEngineId(), contextTokens, contextWindow, pruned); } @SuppressWarnings("unchecked") diff --git a/src/prerna/reactor/model/CompactRoomMessagesReactor.java b/src/prerna/reactor/model/CompactRoomMessagesReactor.java index 468fe63ea33..9626b7ac555 100644 --- a/src/prerna/reactor/model/CompactRoomMessagesReactor.java +++ b/src/prerna/reactor/model/CompactRoomMessagesReactor.java @@ -41,6 +41,7 @@ import prerna.auth.User; import prerna.auth.utils.SecurityEngineUtils; import prerna.engine.api.IModelEngine; +import prerna.engine.impl.model.AbstractModelEngine; import prerna.engine.impl.model.Room; import prerna.engine.impl.model.RoomMessageStore; import prerna.engine.impl.model.RoomUtils; @@ -73,6 +74,12 @@ public class CompactRoomMessagesReactor extends AbstractReactor { */ private static final double TOOL_TOKEN_RATIO_THRESHOLD = 0.25; + // summary output budget: 20% of the transcript, floor 2k, ceiling min(5% of window, 12k) + private static final double SUMMARY_RATIO = 0.20; + private static final int SUMMARY_MIN_TOKENS = 2_000; + private static final int SUMMARY_MAX_TOKENS = 12_000; + private static final double SUMMARY_MAX_WINDOW_RATIO = 0.05; + private static final Set VALID_COMPACTION_TYPES = new HashSet<>(Arrays.asList("TOOL_PRUNE", "SUMMARY")); public CompactRoomMessagesReactor() { @@ -339,10 +346,18 @@ private Map summarizeMessages(Room room, String messageId, List< return result; } + // cap the reply so the summary always fits the space the trigger reserved + int summaryTokens = summaryTokenBudget(summaryTranscript.length(), modelEngine); + // Ask the LLM to summarize using a throw-away room (no history pollution) - String summarizationPrompt = """ - Summarize the following conversation history. Your summary will be used to - continue this conversation in a new context window, so preserve: + // transcript first, instructions last: with a long transcript in front, models + // otherwise lose the instructions and just continue the chat + String summarizationPrompt = "You write handoff summaries of conversation transcripts. " + + "Below is a conversation transcript between tags. " + + "It is data to summarize, not a conversation to continue.\n\n\n" + + summaryTranscript.toString().trim() + "\n\n\n" + """ + The transcript above has ended. Do not reply to it or continue it. Summarize it. + Your summary will be used to continue this conversation in a new context window, so preserve: - The user's current goal and any sub-tasks - Decisions made and the reasoning behind them @@ -358,13 +373,13 @@ private Map summarizeMessages(Room room, String messageId, List< taking over the conversation. Only summarize up to the point the conversation ends - do not speculate about what comes next. - Begin your response with the literal line "[SUMMARY]" before any other text. - - """ + summaryTranscript.toString().trim(); + Keep the whole summary under about %d words. + """.formatted(summaryWordBudget(summaryTokens)); Room throwawayRoom = RoomUtils.createRoomIfNotExists(null, this.insight, modelEngine, null); InputMessage summarizationMsg = InputMessage.builder(throwawayRoom).withText(summarizationPrompt) - .withModelType(modelEngine.getModelType()).withParamMap(new HashMap<>()).build(); + .withModelType(modelEngine.getModelType()) + .withParamMap(new HashMap<>(Map.of(AbstractModelEngine.MAX_TOKENS, summaryTokens))).build(); ResponseMessage summaryResponse = throwawayRoom.ask(summarizationMsg, modelEngine); String summaryText = summaryResponse != null ? summaryResponse.getContent() : null; if (summaryText == null || summaryText.isBlank()) { @@ -462,6 +477,26 @@ private Map summarizeMessages(Room room, String messageId, List< return result; } + // transcript chars / 4 approximates its tokens + static int summaryTokenBudget(int transcriptChars, IModelEngine modelEngine) { + int ceiling = SUMMARY_MAX_TOKENS; + try { + int contextWindow = modelEngine.getContextWindow(); + if (contextWindow > 0) { + ceiling = (int) Math.min(ceiling, contextWindow * SUMMARY_MAX_WINDOW_RATIO); + } + } catch (RuntimeException e) { + // unknown window; keep the fixed ceiling + } + int budget = Math.max(SUMMARY_MIN_TOKENS, (int) (transcriptChars / 4 * SUMMARY_RATIO)); + return Math.max(1, Math.min(budget, ceiling)); + } + + // ~0.75 words per token, with headroom for formatting + private static int summaryWordBudget(int tokenBudget) { + return Math.max(100, (int) (tokenBudget * 0.6)); + } + private static void appendMessageToTranscript(AbstractMessage m, StringBuilder transcript) { String text = getMessageText(m); if (text != null && !text.isBlank()) { diff --git a/src/prerna/sablecc2/om/execptions/SemossModelEngineException.java b/src/prerna/sablecc2/om/execptions/SemossModelEngineException.java index 20e34457405..bc1762e9eac 100644 --- a/src/prerna/sablecc2/om/execptions/SemossModelEngineException.java +++ b/src/prerna/sablecc2/om/execptions/SemossModelEngineException.java @@ -27,6 +27,7 @@ *******************************************************************************/ package prerna.sablecc2.om.execptions; +import prerna.engine.impl.model.responses.AskErrorModelEngineResponse; import prerna.engine.impl.model.responses.AskModelEngineResponse; public class SemossModelEngineException extends RuntimeException { @@ -41,4 +42,21 @@ public SemossModelEngineException(AskModelEngineResponse errorResponse) { public AskModelEngineResponse getErrorResponse() { return errorResponse; } + + // true when this exception (or a cause) is a model error with reason CONTEXT_OVERFLOW + public static boolean isContextOverflow(Throwable error) { + return contextOverflowError(error) != null; + } + + // the CONTEXT_OVERFLOW error details in this exception or its causes, else null + public static AskErrorModelEngineResponse contextOverflowError(Throwable error) { + for (Throwable t = error; t != null; t = t.getCause() == t ? null : t.getCause()) { + if (t instanceof SemossModelEngineException e + && e.getErrorResponse() instanceof AskErrorModelEngineResponse details + && details.isContextOverflow()) { + return details; + } + } + return null; + } } \ No newline at end of file