wehub-resource-sync
fed8b2eed7
Build and push multi-arch DocsGPT Docker image / build (linux/amd64, ubuntu-latest, amd64) (push) Has been cancelled
Backend release / release (push) Has been cancelled
Bandit Security Scan / bandit_scan (push) Has been cancelled
Build and push multi-arch DocsGPT Docker image / build (linux/arm64, ubuntu-24.04-arm, arm64) (push) Has been cancelled
Build and push multi-arch DocsGPT Docker image / manifest (push) Has been cancelled
Build and push DocsGPT FE Docker image for development / build (linux/amd64, ubuntu-latest, amd64) (push) Has been cancelled
Build and push DocsGPT FE Docker image for development / build (linux/arm64, ubuntu-24.04-arm, arm64) (push) Has been cancelled
Build and push DocsGPT FE Docker image for development / manifest (push) Has been cancelled
Python linting / ruff (push) Has been cancelled
Run python tests with pytest / Run tests and count coverage (3.12) (push) Has been cancelled
React Widget Build / build (push) Has been cancelled
1350 lines
56 KiB
Python
1350 lines
56 KiB
Python
import json
|
|
import logging
|
|
import uuid
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass
|
|
from typing import Any, Dict, Generator, List, Optional, Union
|
|
|
|
from application.logging import build_stack_data
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# Cap the agent tool-call loop. Without this an LLM that keeps
|
|
# requesting more tool calls (preview models, sparse tool results,
|
|
# under-specified prompts) can chain searches indefinitely and the
|
|
# stream never finalises. 25 mirrors Dify's default.
|
|
MAX_TOOL_ITERATIONS = 25
|
|
_FINALIZE_INSTRUCTION = (
|
|
f"You have made {MAX_TOOL_ITERATIONS} tool calls. Provide a final "
|
|
"response to the user based on what you have, without making any "
|
|
"additional tool calls."
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class ToolCall:
|
|
"""Represents a tool/function call from the LLM."""
|
|
|
|
id: str
|
|
name: str
|
|
arguments: Union[str, Dict]
|
|
index: Optional[int] = None
|
|
thought_signature: Optional[str] = None
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: Dict) -> "ToolCall":
|
|
"""Create ToolCall from dictionary."""
|
|
return cls(
|
|
id=data.get("id", ""),
|
|
name=data.get("name", ""),
|
|
arguments=data.get("arguments", {}),
|
|
index=data.get("index"),
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class LLMResponse:
|
|
"""Represents a response from the LLM."""
|
|
|
|
content: str
|
|
tool_calls: List[ToolCall]
|
|
finish_reason: str
|
|
raw_response: Any
|
|
reasoning_content: str = ""
|
|
|
|
@property
|
|
def requires_tool_call(self) -> bool:
|
|
"""Check if the response requires tool calls."""
|
|
return bool(self.tool_calls) and self.finish_reason == "tool_calls"
|
|
|
|
|
|
class LLMHandler(ABC):
|
|
"""Abstract base class for LLM handlers."""
|
|
|
|
def __init__(self):
|
|
self.llm_calls = []
|
|
self.tool_calls = []
|
|
# Cache of provider-name -> handler used by ``_parse_for_response``
|
|
# to parse chunks from a model that a cross-provider fallback
|
|
# swapped in underneath this handler.
|
|
self._parser_by_provider = {}
|
|
|
|
@abstractmethod
|
|
def parse_response(self, response: Any) -> LLMResponse:
|
|
"""Parse raw LLM response into standardized format."""
|
|
pass
|
|
|
|
def _parse_for_response(self, agent, response: Any) -> "LLMResponse":
|
|
"""Parse ``response`` with the handler matching the model that
|
|
actually produced it.
|
|
|
|
``BaseLLM`` runs model fallback *below* the agent (see
|
|
``BaseLLM._stream_with_fallback``): a Google-primary agent that is
|
|
rate-limited can fail over to an OpenAI-compatible backup inside the
|
|
same ``gen_stream`` call. This handler was built for the primary
|
|
provider, so its ``parse_response`` cannot read the backup's chunk
|
|
shape and silently drops tool calls — the agent then stops after the
|
|
first text instead of running the tool loop.
|
|
|
|
``parse_response`` is the only provider-specific step that matters
|
|
here: both providers' ``_iterate_stream`` and ``create_tool_message``
|
|
are identical, so only this call is routed. The orchestration state
|
|
(buffers, ``tool_calls``, ``llm_calls``) stays on ``self``.
|
|
"""
|
|
provider = getattr(getattr(agent, "llm", None), "_responding_provider", None)
|
|
if not isinstance(provider, str):
|
|
return self.parse_response(response)
|
|
return self._handler_for_provider(provider).parse_response(response)
|
|
|
|
def _handler_for_provider(self, provider: str) -> "LLMHandler":
|
|
"""Resolve (and cache) the handler for ``provider``. Reuses ``self``
|
|
when it already matches that provider, so the common no-fallback path
|
|
is unchanged."""
|
|
cached = self._parser_by_provider.get(provider)
|
|
if cached is not None:
|
|
return cached
|
|
from application.llm.handlers.handler_creator import LLMHandlerCreator
|
|
|
|
handler = LLMHandlerCreator.create_handler(provider)
|
|
if type(handler) is type(self):
|
|
handler = self
|
|
self._parser_by_provider[provider] = handler
|
|
return handler
|
|
|
|
@abstractmethod
|
|
def create_tool_message(self, tool_call: ToolCall, result: Any) -> Dict:
|
|
"""Create a tool result message for the conversation history."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def _iterate_stream(self, response: Any) -> Generator:
|
|
"""Iterate through streaming response chunks."""
|
|
pass
|
|
|
|
def process_message_flow(
|
|
self,
|
|
agent,
|
|
initial_response,
|
|
tools_dict: Dict,
|
|
messages: List[Dict],
|
|
attachments: Optional[List] = None,
|
|
stream: bool = False,
|
|
) -> Union[str, Generator]:
|
|
"""
|
|
Main orchestration method for processing LLM message flow.
|
|
|
|
Args:
|
|
agent: The agent instance
|
|
initial_response: Initial LLM response
|
|
tools_dict: Dictionary of available tools
|
|
messages: Conversation history
|
|
attachments: Optional attachments
|
|
stream: Whether to use streaming
|
|
|
|
Returns:
|
|
Final response or generator for streaming
|
|
"""
|
|
messages = self.prepare_messages(agent, messages, attachments)
|
|
|
|
if stream:
|
|
return self.handle_streaming(agent, initial_response, tools_dict, messages)
|
|
else:
|
|
return self.handle_non_streaming(
|
|
agent, initial_response, tools_dict, messages
|
|
)
|
|
|
|
def prepare_messages(
|
|
self, agent, messages: List[Dict], attachments: Optional[List] = None
|
|
) -> List[Dict]:
|
|
"""
|
|
Prepare messages with attachments and provider-specific formatting.
|
|
|
|
|
|
Args:
|
|
agent: The agent instance
|
|
messages: Original messages
|
|
attachments: List of attachments
|
|
|
|
Returns:
|
|
Prepared messages list
|
|
"""
|
|
if not attachments:
|
|
return messages
|
|
logger.info(f"Preparing messages with {len(attachments)} attachments")
|
|
supported_types = agent.llm.get_supported_attachment_types()
|
|
|
|
# Check if provider supports images but not PDF (synthetic PDF support)
|
|
supports_images = any(t.startswith("image/") for t in supported_types)
|
|
supports_pdf = "application/pdf" in supported_types
|
|
|
|
# Process attachments, converting PDFs to images if needed
|
|
processed_attachments = []
|
|
for attachment in attachments:
|
|
mime_type = attachment.get("mime_type")
|
|
|
|
# Synthetic PDF support: convert PDF to images if LLM supports images but not PDF
|
|
if mime_type == "application/pdf" and supports_images and not supports_pdf:
|
|
logger.info(
|
|
f"Converting PDF to images for synthetic PDF support: {attachment.get('path', 'unknown')}"
|
|
)
|
|
try:
|
|
converted_images = self._convert_pdf_to_images(attachment)
|
|
processed_attachments.extend(converted_images)
|
|
logger.info(
|
|
f"Converted PDF to {len(converted_images)} images"
|
|
)
|
|
except Exception as e:
|
|
logger.error(
|
|
f"Failed to convert PDF to images, falling back to text: {e}"
|
|
)
|
|
# Fall back to treating as unsupported (text extraction)
|
|
processed_attachments.append(attachment)
|
|
else:
|
|
processed_attachments.append(attachment)
|
|
|
|
supported_attachments = [
|
|
a for a in processed_attachments if a.get("mime_type") in supported_types
|
|
]
|
|
unsupported_attachments = [
|
|
a for a in processed_attachments if a.get("mime_type") not in supported_types
|
|
]
|
|
|
|
# Process supported attachments with the LLM's custom method
|
|
|
|
if supported_attachments:
|
|
logger.info(
|
|
f"Processing {len(supported_attachments)} supported attachments"
|
|
)
|
|
messages = agent.llm.prepare_messages_with_attachments(
|
|
messages, supported_attachments
|
|
)
|
|
# Process unsupported attachments with default method
|
|
|
|
if unsupported_attachments:
|
|
logger.info(
|
|
f"Processing {len(unsupported_attachments)} unsupported attachments"
|
|
)
|
|
messages = self._append_unsupported_attachments(
|
|
messages, unsupported_attachments
|
|
)
|
|
return messages
|
|
|
|
def _convert_pdf_to_images(self, attachment: Dict) -> List[Dict]:
|
|
"""
|
|
Convert a PDF attachment to a list of image attachments.
|
|
|
|
This enables synthetic PDF support for LLMs that support images but not PDFs.
|
|
|
|
Args:
|
|
attachment: PDF attachment dictionary with 'path' and optional 'content'
|
|
|
|
Returns:
|
|
List of image attachment dictionaries with 'data', 'mime_type', and 'page'
|
|
"""
|
|
from application.utils import convert_pdf_to_images
|
|
from application.storage.storage_creator import StorageCreator
|
|
|
|
file_path = attachment.get("path")
|
|
if not file_path:
|
|
raise ValueError("No file path provided in PDF attachment")
|
|
|
|
storage = StorageCreator.get_storage()
|
|
|
|
# Convert PDF to images
|
|
images_data = convert_pdf_to_images(
|
|
file_path=file_path,
|
|
storage=storage,
|
|
max_pages=20,
|
|
dpi=150,
|
|
)
|
|
|
|
return images_data
|
|
|
|
def _append_unsupported_attachments(
|
|
self, messages: List[Dict], attachments: List[Dict]
|
|
) -> List[Dict]:
|
|
"""
|
|
Default method to append unsupported attachment content to system prompt.
|
|
|
|
Args:
|
|
messages: Current messages
|
|
attachments: List of unsupported attachments
|
|
|
|
Returns:
|
|
Updated messages list
|
|
"""
|
|
prepared_messages = messages.copy()
|
|
attachment_texts = []
|
|
|
|
for attachment in attachments:
|
|
logger.info(f"Adding attachment {attachment.get('id')} to context")
|
|
if "content" in attachment:
|
|
attachment_texts.append(
|
|
f"Attached file content:\n\n{attachment['content']}"
|
|
)
|
|
if attachment_texts:
|
|
combined_text = "\n\n".join(attachment_texts)
|
|
|
|
system_msg = next(
|
|
(msg for msg in prepared_messages if msg.get("role") == "system"),
|
|
{"role": "system", "content": ""},
|
|
)
|
|
|
|
if system_msg not in prepared_messages:
|
|
prepared_messages.insert(0, system_msg)
|
|
system_msg["content"] += f"\n\n{combined_text}"
|
|
return prepared_messages
|
|
|
|
def _prune_messages_minimal(self, messages: List[Dict]) -> Optional[List[Dict]]:
|
|
"""
|
|
Build a minimal context: system prompt + latest user message only.
|
|
Drops all tool/function messages to shrink context aggressively.
|
|
"""
|
|
system_message = next((m for m in messages if m.get("role") == "system"), None)
|
|
if not system_message:
|
|
logger.warning("Cannot prune messages minimally: missing system message.")
|
|
return None
|
|
last_non_system = None
|
|
for m in reversed(messages):
|
|
if m.get("role") == "user":
|
|
last_non_system = m
|
|
break
|
|
if not last_non_system and m.get("role") not in ("system", None):
|
|
last_non_system = m
|
|
if not last_non_system:
|
|
logger.warning("Cannot prune messages minimally: missing user/assistant messages.")
|
|
return None
|
|
logger.info("Pruning context to system + latest user/assistant message to proceed.")
|
|
return [system_message, last_non_system]
|
|
|
|
def _extract_text_from_content(self, content: Any) -> str:
|
|
"""
|
|
Convert message content (str or list of parts) to plain text for compression.
|
|
"""
|
|
if isinstance(content, str):
|
|
return content
|
|
if isinstance(content, list):
|
|
parts_text = []
|
|
for item in content:
|
|
if isinstance(item, dict):
|
|
if "text" in item and item["text"] is not None:
|
|
parts_text.append(str(item["text"]))
|
|
elif "function_call" in item or "function_response" in item:
|
|
# Keep serialized function calls/responses so the compressor sees actions
|
|
parts_text.append(str(item))
|
|
elif "files" in item:
|
|
# Image attachments arrive with raw bytes / base64
|
|
# inline (see GoogleLLM.prepare_messages_with_attachments).
|
|
# ``str(item)`` would dump the whole byte/base64
|
|
# blob into the compression prompt and bust the
|
|
# compression LLM's input limit.
|
|
files = item.get("files") or []
|
|
descriptors = []
|
|
if isinstance(files, list):
|
|
for f in files:
|
|
if isinstance(f, dict):
|
|
descriptors.append(
|
|
f.get("mime_type") or "file"
|
|
)
|
|
elif isinstance(f, str):
|
|
descriptors.append(f)
|
|
if not descriptors:
|
|
descriptors = ["file"]
|
|
parts_text.append(
|
|
f"[attachment: {', '.join(descriptors)}]"
|
|
)
|
|
return "\n".join(parts_text)
|
|
return ""
|
|
|
|
def _build_conversation_from_messages(self, messages: List[Dict]) -> Optional[Dict]:
|
|
"""
|
|
Build a conversation-like dict from current messages so we can compress
|
|
even when the conversation isn't persisted yet. Includes tool calls/results.
|
|
"""
|
|
queries = []
|
|
current_prompt = None
|
|
current_tool_calls = {}
|
|
|
|
def _commit_query(response_text: str):
|
|
nonlocal current_prompt, current_tool_calls
|
|
if current_prompt is None and not response_text:
|
|
return
|
|
tool_calls_list = list(current_tool_calls.values())
|
|
queries.append(
|
|
{
|
|
"prompt": current_prompt or "",
|
|
"response": response_text,
|
|
"tool_calls": tool_calls_list,
|
|
}
|
|
)
|
|
current_prompt = None
|
|
current_tool_calls = {}
|
|
|
|
for message in messages:
|
|
role = message.get("role")
|
|
content = message.get("content")
|
|
|
|
if role == "user":
|
|
current_prompt = self._extract_text_from_content(content)
|
|
|
|
elif role in {"assistant", "model"}:
|
|
# Standard format: tool_calls array on assistant message
|
|
msg_tool_calls = message.get("tool_calls")
|
|
if msg_tool_calls:
|
|
for tc in msg_tool_calls:
|
|
call_id = tc.get("id") or str(uuid.uuid4())
|
|
func = tc.get("function", {})
|
|
args = func.get("arguments")
|
|
if isinstance(args, str):
|
|
try:
|
|
args = json.loads(args)
|
|
except (json.JSONDecodeError, TypeError):
|
|
pass
|
|
current_tool_calls[call_id] = {
|
|
"tool_name": "unknown_tool",
|
|
"action_name": func.get("name"),
|
|
"arguments": args,
|
|
"result": None,
|
|
"status": "called",
|
|
"call_id": call_id,
|
|
}
|
|
continue
|
|
|
|
# Legacy format: function_call/function_response in content list
|
|
if isinstance(content, list):
|
|
has_fc = False
|
|
for item in content:
|
|
if "function_call" in item:
|
|
has_fc = True
|
|
fc = item["function_call"]
|
|
call_id = fc.get("call_id") or str(uuid.uuid4())
|
|
current_tool_calls[call_id] = {
|
|
"tool_name": "unknown_tool",
|
|
"action_name": fc.get("name"),
|
|
"arguments": fc.get("args"),
|
|
"result": None,
|
|
"status": "called",
|
|
"call_id": call_id,
|
|
}
|
|
if has_fc:
|
|
continue
|
|
|
|
response_text = self._extract_text_from_content(content)
|
|
_commit_query(response_text)
|
|
|
|
elif role == "tool":
|
|
# Standard format: tool_call_id on tool message
|
|
call_id = message.get("tool_call_id")
|
|
tool_text = self._extract_text_from_content(content)
|
|
|
|
if call_id and call_id in current_tool_calls:
|
|
current_tool_calls[call_id]["result"] = tool_text
|
|
current_tool_calls[call_id]["status"] = "completed"
|
|
# Legacy: function_response in content list
|
|
elif isinstance(content, list):
|
|
for item in content:
|
|
if "function_response" in item:
|
|
legacy_id = item["function_response"].get("call_id")
|
|
if legacy_id and legacy_id in current_tool_calls:
|
|
current_tool_calls[legacy_id]["result"] = tool_text
|
|
current_tool_calls[legacy_id]["status"] = "completed"
|
|
break
|
|
elif call_id is None and queries:
|
|
queries[-1].setdefault("tool_calls", []).append(
|
|
{
|
|
"tool_name": "unknown_tool",
|
|
"action_name": "unknown_action",
|
|
"arguments": {},
|
|
"result": tool_text,
|
|
"status": "completed",
|
|
}
|
|
)
|
|
|
|
# If there's an unfinished prompt with tool_calls but no response yet, commit it
|
|
if current_prompt is not None or current_tool_calls:
|
|
_commit_query(response_text="")
|
|
|
|
if not queries:
|
|
return None
|
|
|
|
return {
|
|
"queries": queries,
|
|
"compression_metadata": {
|
|
"is_compressed": False,
|
|
"compression_points": [],
|
|
},
|
|
}
|
|
|
|
def _rebuild_messages_after_compression(
|
|
self,
|
|
messages: List[Dict],
|
|
compressed_summary: Optional[str],
|
|
recent_queries: List[Dict],
|
|
include_current_execution: bool = False,
|
|
include_tool_calls: bool = False,
|
|
) -> Optional[List[Dict]]:
|
|
"""
|
|
Rebuild the message list after compression so tool execution can continue.
|
|
|
|
Delegates to MessageBuilder for the actual reconstruction.
|
|
"""
|
|
from application.api.answer.services.compression.message_builder import (
|
|
MessageBuilder,
|
|
)
|
|
|
|
return MessageBuilder.rebuild_messages_after_compression(
|
|
messages=messages,
|
|
compressed_summary=compressed_summary,
|
|
recent_queries=recent_queries,
|
|
include_current_execution=include_current_execution,
|
|
include_tool_calls=include_tool_calls,
|
|
)
|
|
|
|
def _perform_mid_execution_compression(
|
|
self, agent, messages: List[Dict]
|
|
) -> tuple[bool, Optional[List[Dict]]]:
|
|
"""
|
|
Perform compression during tool execution and rebuild messages.
|
|
|
|
Uses the new orchestrator for simplified compression.
|
|
|
|
Args:
|
|
agent: The agent instance
|
|
messages: Current conversation messages
|
|
|
|
Returns:
|
|
(success: bool, rebuilt_messages: Optional[List[Dict]])
|
|
"""
|
|
try:
|
|
from application.api.answer.services.compression import (
|
|
CompressionOrchestrator,
|
|
)
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
|
|
conversation_service = ConversationService()
|
|
orchestrator = CompressionOrchestrator(conversation_service)
|
|
|
|
# Get conversation from database (may be None for new sessions)
|
|
conversation = conversation_service.get_conversation(
|
|
agent.conversation_id, agent.initial_user_id
|
|
)
|
|
|
|
if conversation:
|
|
# Merge current in-flight messages (including tool calls)
|
|
conversation_from_msgs = self._build_conversation_from_messages(messages)
|
|
if conversation_from_msgs:
|
|
conversation = conversation_from_msgs
|
|
else:
|
|
logger.warning(
|
|
"Could not load conversation for compression; attempting in-memory compression"
|
|
)
|
|
return self._perform_in_memory_compression(agent, messages)
|
|
|
|
# Use orchestrator to perform compression. ``model_user_id``
|
|
# keeps BYOM registry resolution scoped to the model owner
|
|
# (shared-agent dispatch) while ``user_id`` stays the caller
|
|
# for the conversation access check.
|
|
result = orchestrator.compress_mid_execution(
|
|
conversation_id=agent.conversation_id,
|
|
user_id=agent.initial_user_id,
|
|
model_user_id=getattr(agent, "model_user_id", None),
|
|
model_id=agent.model_id,
|
|
decoded_token=getattr(agent, "decoded_token", {}),
|
|
current_conversation=conversation,
|
|
)
|
|
|
|
if not result.success:
|
|
logger.warning(f"Mid-execution compression failed: {result.error}")
|
|
# Try minimal pruning as fallback
|
|
pruned = self._prune_messages_minimal(messages)
|
|
if pruned:
|
|
agent.context_limit_reached = False
|
|
agent.current_token_count = 0
|
|
return True, pruned
|
|
return False, None
|
|
|
|
if not result.compression_performed:
|
|
logger.warning("Compression not performed")
|
|
return False, None
|
|
|
|
# Check if compression actually reduced tokens
|
|
if result.metadata:
|
|
if result.metadata.compressed_token_count >= result.metadata.original_token_count:
|
|
logger.warning(
|
|
"Compression did not reduce token count; falling back to minimal pruning"
|
|
)
|
|
pruned = self._prune_messages_minimal(messages)
|
|
if pruned:
|
|
agent.context_limit_reached = False
|
|
agent.current_token_count = 0
|
|
return True, pruned
|
|
return False, None
|
|
|
|
logger.info(
|
|
f"Mid-execution compression successful - ratio: {result.metadata.compression_ratio:.1f}x, "
|
|
f"saved {result.metadata.original_token_count - result.metadata.compressed_token_count} tokens"
|
|
)
|
|
|
|
# Also store the compression summary as a visible message
|
|
if result.metadata:
|
|
conversation_service.append_compression_message(
|
|
agent.conversation_id, result.metadata.to_dict()
|
|
)
|
|
|
|
# Update agent's compressed summary for downstream persistence
|
|
agent.compressed_summary = result.compressed_summary
|
|
agent.compression_metadata = result.metadata.to_dict() if result.metadata else None
|
|
agent.compression_saved = False
|
|
|
|
# Reset the context limit flag so tools can continue
|
|
agent.context_limit_reached = False
|
|
agent.current_token_count = 0
|
|
|
|
# Rebuild messages
|
|
rebuilt_messages = self._rebuild_messages_after_compression(
|
|
messages,
|
|
result.compressed_summary,
|
|
result.recent_queries,
|
|
include_current_execution=False,
|
|
include_tool_calls=False,
|
|
)
|
|
|
|
if rebuilt_messages is None:
|
|
return False, None
|
|
|
|
return True, rebuilt_messages
|
|
|
|
except Exception as e:
|
|
logger.error(
|
|
f"Error performing mid-execution compression: {str(e)}", exc_info=True
|
|
)
|
|
return False, None
|
|
|
|
def _perform_in_memory_compression(
|
|
self, agent, messages: List[Dict]
|
|
) -> tuple[bool, Optional[List[Dict]]]:
|
|
"""
|
|
Fallback compression path when the conversation is not yet persisted.
|
|
|
|
Uses CompressionService directly without DB persistence.
|
|
"""
|
|
try:
|
|
from application.api.answer.services.compression.service import (
|
|
CompressionService,
|
|
)
|
|
from application.core.model_utils import (
|
|
get_api_key_for_provider,
|
|
get_provider_from_model_id,
|
|
)
|
|
from application.core.settings import settings
|
|
from application.llm.llm_creator import LLMCreator
|
|
|
|
conversation = self._build_conversation_from_messages(messages)
|
|
if not conversation:
|
|
logger.warning(
|
|
"Cannot perform in-memory compression: no user/assistant turns found"
|
|
)
|
|
return False, None
|
|
|
|
compression_model = (
|
|
settings.COMPRESSION_MODEL_OVERRIDE
|
|
if settings.COMPRESSION_MODEL_OVERRIDE
|
|
else agent.model_id
|
|
)
|
|
agent_decoded = getattr(agent, "decoded_token", None)
|
|
caller_sub = (
|
|
agent_decoded.get("sub")
|
|
if isinstance(agent_decoded, dict)
|
|
else None
|
|
)
|
|
# Use model-owner scope (mirrors orchestrator path) so
|
|
# shared-agent owner-BYOM resolves under the owner's layer.
|
|
compression_user_id = (
|
|
getattr(agent, "model_user_id", None) or caller_sub
|
|
)
|
|
provider = get_provider_from_model_id(
|
|
compression_model, user_id=compression_user_id
|
|
)
|
|
api_key = get_api_key_for_provider(provider)
|
|
compression_llm = LLMCreator.create_llm(
|
|
provider,
|
|
api_key,
|
|
getattr(agent, "user_api_key", None),
|
|
getattr(agent, "decoded_token", None),
|
|
model_id=compression_model,
|
|
agent_id=getattr(agent, "agent_id", None),
|
|
model_user_id=compression_user_id,
|
|
)
|
|
# Side-channel LLM tag — see ``orchestrator.py`` for rationale.
|
|
compression_llm._token_usage_source = "compression"
|
|
compression_llm._request_id = getattr(agent, "_request_id", None) \
|
|
or getattr(getattr(agent, "llm", None), "_request_id", None)
|
|
|
|
# Create service without DB persistence capability
|
|
compression_service = CompressionService(
|
|
llm=compression_llm,
|
|
model_id=compression_model,
|
|
conversation_service=None, # No DB updates for in-memory
|
|
)
|
|
|
|
queries_count = len(conversation.get("queries", []))
|
|
compress_up_to = queries_count - 1
|
|
|
|
if compress_up_to < 0 or queries_count == 0:
|
|
logger.warning("Not enough queries to compress in-memory context")
|
|
return False, None
|
|
|
|
metadata = compression_service.compress_conversation(
|
|
conversation,
|
|
compress_up_to_index=compress_up_to,
|
|
)
|
|
|
|
# If compression doesn't reduce tokens, fall back to minimal pruning
|
|
if (
|
|
metadata.compressed_token_count
|
|
>= metadata.original_token_count
|
|
):
|
|
logger.warning(
|
|
"In-memory compression did not reduce token count; falling back to minimal pruning"
|
|
)
|
|
pruned = self._prune_messages_minimal(messages)
|
|
if pruned:
|
|
agent.context_limit_reached = False
|
|
agent.current_token_count = 0
|
|
return True, pruned
|
|
return False, None
|
|
|
|
# Attach metadata to synthetic conversation
|
|
conversation["compression_metadata"] = {
|
|
"is_compressed": True,
|
|
"compression_points": [metadata.to_dict()],
|
|
}
|
|
|
|
compressed_summary, recent_queries = (
|
|
compression_service.get_compressed_context(conversation)
|
|
)
|
|
|
|
agent.compressed_summary = compressed_summary
|
|
agent.compression_metadata = metadata.to_dict()
|
|
agent.compression_saved = False
|
|
agent.context_limit_reached = False
|
|
agent.current_token_count = 0
|
|
|
|
rebuilt_messages = self._rebuild_messages_after_compression(
|
|
messages,
|
|
compressed_summary,
|
|
recent_queries,
|
|
include_current_execution=False,
|
|
include_tool_calls=False,
|
|
)
|
|
if rebuilt_messages is None:
|
|
return False, None
|
|
|
|
logger.info(
|
|
f"In-memory compression successful - ratio: {metadata.compression_ratio:.1f}x, "
|
|
f"saved {metadata.original_token_count - metadata.compressed_token_count} tokens"
|
|
)
|
|
return True, rebuilt_messages
|
|
|
|
except Exception as e:
|
|
logger.error(
|
|
f"Error performing in-memory compression: {str(e)}", exc_info=True
|
|
)
|
|
return False, None
|
|
|
|
def handle_tool_calls(
|
|
self,
|
|
agent,
|
|
tool_calls: List[ToolCall],
|
|
tools_dict: Dict,
|
|
messages: List[Dict],
|
|
reasoning_content: str = "",
|
|
) -> Generator:
|
|
"""
|
|
Execute tool calls and update conversation history.
|
|
|
|
When a tool requires approval or client-side execution, it is
|
|
collected as a pending action instead of being executed. The
|
|
generator returns ``(updated_messages, pending_actions)`` where
|
|
*pending_actions* is ``None`` when every tool was executed
|
|
normally, or a list of dicts describing actions the client must
|
|
resolve before the LLM loop can continue.
|
|
|
|
Args:
|
|
agent: The agent instance
|
|
tool_calls: List of tool calls to execute
|
|
tools_dict: Available tools dictionary
|
|
messages: Current conversation history
|
|
reasoning_content: Reasoning text emitted by the model
|
|
before these tool calls. Attached to the recorded
|
|
assistant message so providers that require reasoning
|
|
to round-trip (DeepSeek thinking mode) accept the
|
|
follow-up request.
|
|
|
|
Returns:
|
|
Tuple of (updated_messages, pending_actions).
|
|
pending_actions is None if all tools executed, otherwise a list.
|
|
"""
|
|
updated_messages = messages.copy()
|
|
pending_actions: List[Dict] = []
|
|
|
|
for i, call in enumerate(tool_calls):
|
|
# Check context limit before executing tool call
|
|
if hasattr(agent, '_check_context_limit') and agent._check_context_limit(updated_messages):
|
|
# Context limit reached - attempt mid-execution compression
|
|
compression_attempted = False
|
|
compression_successful = False
|
|
|
|
try:
|
|
from application.core.settings import settings
|
|
compression_enabled = settings.ENABLE_CONVERSATION_COMPRESSION
|
|
except Exception:
|
|
compression_enabled = False
|
|
|
|
if compression_enabled:
|
|
compression_attempted = True
|
|
try:
|
|
logger.info(
|
|
f"Context limit reached with {len(tool_calls) - i} remaining tool calls. "
|
|
f"Attempting mid-execution compression..."
|
|
)
|
|
|
|
# Trigger mid-execution compression (DB-backed if available, otherwise in-memory)
|
|
compression_successful, rebuilt_messages = self._perform_mid_execution_compression(
|
|
agent, updated_messages
|
|
)
|
|
|
|
if compression_successful and rebuilt_messages is not None:
|
|
# Update the messages list with rebuilt compressed version
|
|
updated_messages = rebuilt_messages
|
|
|
|
# Yield compression success message
|
|
yield {
|
|
"type": "info",
|
|
"data": {
|
|
"message": "Context window limit reached. Compressed conversation history to continue processing."
|
|
}
|
|
}
|
|
|
|
logger.info(
|
|
f"Mid-execution compression successful. Continuing with {len(tool_calls) - i} remaining tool calls."
|
|
)
|
|
# Proceed to execute the current tool call with the reduced context
|
|
else:
|
|
logger.warning("Mid-execution compression attempted but failed. Skipping remaining tools.")
|
|
except Exception as e:
|
|
logger.error(f"Error during mid-execution compression: {str(e)}", exc_info=True)
|
|
compression_attempted = True
|
|
compression_successful = False
|
|
|
|
# If compression wasn't attempted or failed, skip remaining tools
|
|
if not compression_successful:
|
|
if i == 0:
|
|
# Special case: limit reached before executing any tools
|
|
# This can happen when previous tool responses pushed context over limit
|
|
if compression_attempted:
|
|
logger.warning(
|
|
f"Context limit reached before executing any tools. "
|
|
f"Compression attempted but failed. "
|
|
f"Skipping all {len(tool_calls)} pending tool call(s). "
|
|
f"This typically occurs when previous tool responses contained large amounts of data."
|
|
)
|
|
else:
|
|
logger.warning(
|
|
f"Context limit reached before executing any tools. "
|
|
f"Skipping all {len(tool_calls)} pending tool call(s). "
|
|
f"This typically occurs when previous tool responses contained large amounts of data. "
|
|
f"Consider enabling compression or using a model with larger context window."
|
|
)
|
|
else:
|
|
# Normal case: executed some tools, now stopping
|
|
tool_word = "tool call" if i == 1 else "tool calls"
|
|
remaining = len(tool_calls) - i
|
|
remaining_word = "tool call" if remaining == 1 else "tool calls"
|
|
if compression_attempted:
|
|
logger.warning(
|
|
f"Context limit reached after executing {i} {tool_word}. "
|
|
f"Compression attempted but failed. "
|
|
f"Skipping remaining {remaining} {remaining_word}."
|
|
)
|
|
else:
|
|
logger.warning(
|
|
f"Context limit reached after executing {i} {tool_word}. "
|
|
f"Skipping remaining {remaining} {remaining_word}. "
|
|
f"Consider enabling compression or using a model with larger context window."
|
|
)
|
|
|
|
# Mark remaining tools as skipped
|
|
for remaining_call in tool_calls[i:]:
|
|
skip_message = {
|
|
"type": "tool_call",
|
|
"data": {
|
|
"tool_name": "system",
|
|
"call_id": remaining_call.id,
|
|
"action_name": remaining_call.name,
|
|
"arguments": {},
|
|
"result": "Skipped: Context limit reached. Too many tool calls in conversation.",
|
|
"status": "skipped"
|
|
}
|
|
}
|
|
yield skip_message
|
|
|
|
# Set flag on agent
|
|
agent.context_limit_reached = True
|
|
break
|
|
|
|
# ---- Pause check: approval / client-side execution ----
|
|
llm_class = agent.llm.__class__.__name__
|
|
pause_info = agent.tool_executor.check_pause(
|
|
tools_dict, call, llm_class
|
|
)
|
|
if pause_info:
|
|
# Headless (scheduled / webhook): synthesize a denial tool message
|
|
# so the LLM finishes gracefully instead of stalling on a pause
|
|
# nobody will resolve, then journal so the reconciler sees it.
|
|
if pause_info.get("pause_type") == "headless_denied":
|
|
deny_reason = pause_info.get(
|
|
"deny_reason", "Tool blocked in headless mode."
|
|
)
|
|
args_str = (
|
|
json.dumps(call.arguments)
|
|
if isinstance(call.arguments, dict)
|
|
else (call.arguments or "{}")
|
|
)
|
|
tool_call_obj = {
|
|
"id": pause_info["call_id"],
|
|
"type": "function",
|
|
"function": {
|
|
"name": call.name,
|
|
"arguments": args_str,
|
|
},
|
|
}
|
|
if getattr(call, "thought_signature", None):
|
|
tool_call_obj["thought_signature"] = call.thought_signature
|
|
assistant_msg: Dict[str, Any] = {
|
|
"role": "assistant",
|
|
"content": None,
|
|
"tool_calls": [tool_call_obj],
|
|
}
|
|
if reasoning_content:
|
|
assistant_msg["reasoning_content"] = reasoning_content
|
|
updated_messages.append(assistant_msg)
|
|
denial_call = ToolCall(
|
|
id=pause_info["call_id"],
|
|
name=call.name,
|
|
arguments=call.arguments,
|
|
)
|
|
updated_messages.append(
|
|
self.create_tool_message(
|
|
denial_call,
|
|
f"Tool denied (headless): {deny_reason}",
|
|
)
|
|
)
|
|
if hasattr(agent.tool_executor, "headless_denials"):
|
|
agent.tool_executor.headless_denials.append(pause_info)
|
|
from application.agents.tool_executor import (
|
|
_mark_failed,
|
|
_record_proposed,
|
|
)
|
|
|
|
if _record_proposed(
|
|
pause_info["call_id"],
|
|
pause_info["tool_name"],
|
|
pause_info["action_name"],
|
|
pause_info.get("arguments") or {},
|
|
tool_id=pause_info.get("tool_id"),
|
|
message_id=agent.tool_executor.message_id,
|
|
user_id=agent.tool_executor.user,
|
|
agent_id=agent.tool_executor.agent_id,
|
|
):
|
|
_mark_failed(
|
|
pause_info["call_id"],
|
|
f"headless: {deny_reason}",
|
|
message_id=agent.tool_executor.message_id,
|
|
user_id=agent.tool_executor.user,
|
|
)
|
|
yield {
|
|
"type": "tool_call",
|
|
"data": {
|
|
"tool_name": pause_info["tool_name"],
|
|
"call_id": pause_info["call_id"],
|
|
"action_name": pause_info.get(
|
|
"llm_name", pause_info["name"]
|
|
),
|
|
"arguments": pause_info["arguments"],
|
|
"status": "denied",
|
|
"error": deny_reason,
|
|
"error_type": pause_info.get(
|
|
"error_type", "tool_not_allowed"
|
|
),
|
|
},
|
|
}
|
|
continue
|
|
# Yield pause event so the client knows this tool is waiting
|
|
pause_data = {
|
|
"tool_name": pause_info["tool_name"],
|
|
"call_id": pause_info["call_id"],
|
|
"action_name": pause_info.get("llm_name", pause_info["name"]),
|
|
"arguments": pause_info["arguments"],
|
|
"status": pause_info["pause_type"],
|
|
}
|
|
# Surface device_id for remote_device pauses so the approval UI
|
|
# can wire the sticky "don't ask again" button.
|
|
if pause_info.get("device_id"):
|
|
pause_data["device_id"] = pause_info["device_id"]
|
|
yield {"type": "tool_call", "data": pause_data}
|
|
pending_actions.append(pause_info)
|
|
# Do NOT add messages for pending tools here.
|
|
# They will be added on resume to keep call/result pairs together.
|
|
continue
|
|
|
|
# One assistant(tool_calls) message per call: track whether the
|
|
# success path already appended it so the except below doesn't
|
|
# add a second one when create_tool_message fails post-append.
|
|
assistant_appended = False
|
|
try:
|
|
self.tool_calls.append(call)
|
|
tool_executor_gen = agent._execute_tool_action(tools_dict, call)
|
|
while True:
|
|
try:
|
|
yield next(tool_executor_gen)
|
|
except StopIteration as e:
|
|
tool_response, call_id = e.value
|
|
break
|
|
|
|
# Standard internal format: assistant message with tool_calls array
|
|
args_str = (
|
|
json.dumps(call.arguments)
|
|
if isinstance(call.arguments, dict)
|
|
else call.arguments
|
|
)
|
|
tool_call_obj = {
|
|
"id": call_id,
|
|
"type": "function",
|
|
"function": {
|
|
"name": call.name,
|
|
"arguments": args_str,
|
|
},
|
|
}
|
|
# Preserve thought_signature for Google Gemini 3 models
|
|
if call.thought_signature:
|
|
tool_call_obj["thought_signature"] = call.thought_signature
|
|
|
|
assistant_msg: Dict[str, Any] = {
|
|
"role": "assistant",
|
|
"content": None,
|
|
"tool_calls": [tool_call_obj],
|
|
}
|
|
# Each call in a parallel batch becomes its own
|
|
# assistant message here, so the same per-round
|
|
# reasoning has to ride on every one — DeepSeek
|
|
# thinking mode rejects any assistant message in the
|
|
# active turn that's missing reasoning_content.
|
|
if reasoning_content:
|
|
assistant_msg["reasoning_content"] = reasoning_content
|
|
updated_messages.append(assistant_msg)
|
|
assistant_appended = True
|
|
|
|
# The tool result's tool_call_id must match the id put on the
|
|
# assistant tool_call above (``call_id`` — a synthesized UUID
|
|
# when the provider omitted an id), not the raw ``call.id`` which
|
|
# may be empty. A mismatch orphans the tool message and 400s the
|
|
# next completion ("'tool' must be a response to a preceding
|
|
# message with 'tool_calls'").
|
|
resolved_call = ToolCall(
|
|
id=call_id, name=call.name, arguments=call.arguments
|
|
)
|
|
updated_messages.append(
|
|
self.create_tool_message(resolved_call, tool_response)
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Error executing tool: {str(e)}", exc_info=True)
|
|
# The error tool message's tool_call_id must match the assistant
|
|
# tool_call id that precedes it. When the success path already
|
|
# appended one (create_tool_message failed afterwards) that id is
|
|
# the executor-returned ``call_id``; otherwise the except builds
|
|
# its own assistant message below from ``call.id``.
|
|
error_id = call_id if assistant_appended else call.id
|
|
error_call = ToolCall(
|
|
id=error_id, name=call.name, arguments=call.arguments
|
|
)
|
|
error_response = f"Error executing tool: {str(e)}"
|
|
# Mirror the success path: a role:"tool" message must follow
|
|
# an assistant message carrying its tool_calls, or the next
|
|
# provider completion 400s ("'tool' must be a response to a
|
|
# preceding message with 'tool_calls'"). Skip it when the
|
|
# success path already appended one for this call — a
|
|
# create_tool_message failure after that append would
|
|
# otherwise duplicate the assistant message and 400 the
|
|
# same way an orphan tool message does.
|
|
if not assistant_appended:
|
|
args_str = (
|
|
json.dumps(call.arguments)
|
|
if isinstance(call.arguments, dict)
|
|
else call.arguments
|
|
)
|
|
tool_call_obj = {
|
|
"id": call.id,
|
|
"type": "function",
|
|
"function": {
|
|
"name": call.name,
|
|
"arguments": args_str,
|
|
},
|
|
}
|
|
if call.thought_signature:
|
|
tool_call_obj["thought_signature"] = call.thought_signature
|
|
assistant_msg: Dict[str, Any] = {
|
|
"role": "assistant",
|
|
"content": None,
|
|
"tool_calls": [tool_call_obj],
|
|
}
|
|
if reasoning_content:
|
|
assistant_msg["reasoning_content"] = reasoning_content
|
|
updated_messages.append(assistant_msg)
|
|
|
|
error_message = self.create_tool_message(error_call, error_response)
|
|
updated_messages.append(error_message)
|
|
|
|
mapping = agent.tool_executor._name_to_tool
|
|
if call.name in mapping:
|
|
resolved_tool_id, _ = mapping[call.name]
|
|
tool_name = tools_dict.get(resolved_tool_id, {}).get(
|
|
"name", "unknown_tool"
|
|
)
|
|
else:
|
|
tool_name = "unknown_tool"
|
|
full_action_name = call.name
|
|
yield {
|
|
"type": "tool_call",
|
|
"data": {
|
|
"tool_name": tool_name,
|
|
"call_id": call.id,
|
|
"action_name": full_action_name,
|
|
"arguments": call.arguments,
|
|
"error": error_response,
|
|
"status": "error",
|
|
},
|
|
}
|
|
return updated_messages, pending_actions if pending_actions else None
|
|
|
|
def handle_non_streaming(
|
|
self, agent, response: Any, tools_dict: Dict, messages: List[Dict]
|
|
) -> Generator:
|
|
"""
|
|
Handle non-streaming response flow.
|
|
|
|
Args:
|
|
agent: The agent instance
|
|
response: Current LLM response
|
|
tools_dict: Available tools dictionary
|
|
messages: Conversation history
|
|
|
|
Returns:
|
|
Final response after processing all tool calls
|
|
"""
|
|
parsed = self._parse_for_response(agent, response)
|
|
self.llm_calls.append(build_stack_data(agent.llm))
|
|
|
|
iteration = 0
|
|
while parsed.requires_tool_call:
|
|
iteration += 1
|
|
reasoning_for_round = parsed.reasoning_content or ""
|
|
tool_handler_gen = self.handle_tool_calls(
|
|
agent,
|
|
parsed.tool_calls,
|
|
tools_dict,
|
|
messages,
|
|
reasoning_content=reasoning_for_round,
|
|
)
|
|
while True:
|
|
try:
|
|
yield next(tool_handler_gen)
|
|
except StopIteration as e:
|
|
messages, pending_actions = e.value
|
|
break
|
|
|
|
# If tools need approval or client execution, pause the loop
|
|
if pending_actions:
|
|
agent._pending_continuation = {
|
|
"messages": messages,
|
|
"pending_tool_calls": pending_actions,
|
|
"tools_dict": tools_dict,
|
|
"reasoning_content": reasoning_for_round,
|
|
}
|
|
yield {
|
|
"type": "tool_calls_pending",
|
|
"data": {"pending_tool_calls": pending_actions},
|
|
}
|
|
return ""
|
|
|
|
# Cap reached: force one final tool-less call so the stream
|
|
# always ends with content rather than cutting off.
|
|
if iteration >= MAX_TOOL_ITERATIONS:
|
|
logger.warning(
|
|
"agent tool loop hit cap (%d); forcing finalize",
|
|
MAX_TOOL_ITERATIONS,
|
|
)
|
|
messages.append(
|
|
{"role": "system", "content": _FINALIZE_INSTRUCTION},
|
|
)
|
|
response = agent.llm.gen(
|
|
model=getattr(agent.llm, "model_id", None) or agent.model_id,
|
|
messages=messages,
|
|
tools=None,
|
|
)
|
|
parsed = self._parse_for_response(agent, response)
|
|
self.llm_calls.append(build_stack_data(agent.llm))
|
|
break
|
|
|
|
# ``agent.model_id`` is the registry id (a UUID for BYOM
|
|
# records). Use the LLM's own model_id, which LLMCreator
|
|
# already resolved to the upstream model name. Built-ins:
|
|
# the two are equal; BYOM: the upstream name like
|
|
# "mistral-large-latest" instead of the UUID.
|
|
response = agent.llm.gen(
|
|
model=getattr(agent.llm, "model_id", None) or agent.model_id,
|
|
messages=messages,
|
|
tools=agent.tools,
|
|
)
|
|
parsed = self._parse_for_response(agent, response)
|
|
self.llm_calls.append(build_stack_data(agent.llm))
|
|
return parsed.content
|
|
|
|
def handle_streaming(
|
|
self,
|
|
agent,
|
|
response: Any,
|
|
tools_dict: Dict,
|
|
messages: List[Dict],
|
|
_iteration: int = 0,
|
|
) -> Generator:
|
|
"""
|
|
Handle streaming response flow.
|
|
|
|
Args:
|
|
agent: The agent instance
|
|
response: Current LLM response
|
|
tools_dict: Available tools dictionary
|
|
messages: Conversation history
|
|
|
|
Yields:
|
|
Streaming response chunks
|
|
"""
|
|
buffer = ""
|
|
tool_calls = {}
|
|
reasoning_buffer = ""
|
|
|
|
for chunk in self._iterate_stream(response):
|
|
if isinstance(chunk, dict) and chunk.get("type") == "thought":
|
|
reasoning_buffer += chunk.get("thought") or ""
|
|
yield chunk
|
|
continue
|
|
if isinstance(chunk, str):
|
|
yield chunk
|
|
continue
|
|
parsed = self._parse_for_response(agent, chunk)
|
|
if parsed.reasoning_content:
|
|
reasoning_buffer += parsed.reasoning_content
|
|
|
|
if parsed.tool_calls:
|
|
for call in parsed.tool_calls:
|
|
if call.index not in tool_calls:
|
|
tool_calls[call.index] = call
|
|
else:
|
|
existing = tool_calls[call.index]
|
|
if call.id:
|
|
existing.id = call.id
|
|
if call.name:
|
|
existing.name = call.name
|
|
if call.arguments:
|
|
if existing.arguments is None:
|
|
existing.arguments = call.arguments
|
|
else:
|
|
existing.arguments += call.arguments
|
|
# Preserve thought_signature for Google Gemini 3 models
|
|
if call.thought_signature:
|
|
existing.thought_signature = call.thought_signature
|
|
if parsed.finish_reason == "tool_calls":
|
|
tool_handler_gen = self.handle_tool_calls(
|
|
agent,
|
|
list(tool_calls.values()),
|
|
tools_dict,
|
|
messages,
|
|
reasoning_content=reasoning_buffer,
|
|
)
|
|
while True:
|
|
try:
|
|
yield next(tool_handler_gen)
|
|
except StopIteration as e:
|
|
messages, pending_actions = e.value
|
|
break
|
|
tool_calls = {}
|
|
pause_reasoning = reasoning_buffer
|
|
reasoning_buffer = ""
|
|
|
|
# If tools need approval or client execution, pause the loop
|
|
if pending_actions:
|
|
agent._pending_continuation = {
|
|
"messages": messages,
|
|
"pending_tool_calls": pending_actions,
|
|
"tools_dict": tools_dict,
|
|
"reasoning_content": pause_reasoning,
|
|
}
|
|
yield {
|
|
"type": "tool_calls_pending",
|
|
"data": {"pending_tool_calls": pending_actions},
|
|
}
|
|
return
|
|
|
|
next_iteration = _iteration + 1
|
|
cap_reached = next_iteration >= MAX_TOOL_ITERATIONS
|
|
|
|
# Check if context limit was reached during tool execution
|
|
if hasattr(agent, 'context_limit_reached') and agent.context_limit_reached:
|
|
# Add system message warning about context limit
|
|
messages.append({
|
|
"role": "system",
|
|
"content": (
|
|
"WARNING: Context window limit has been reached. "
|
|
"Please provide a final response to the user without making additional tool calls. "
|
|
"Summarize the work completed so far."
|
|
)
|
|
})
|
|
logger.info("Context limit reached - instructing agent to wrap up")
|
|
elif cap_reached:
|
|
logger.warning(
|
|
"agent tool loop hit cap (%d); forcing finalize",
|
|
MAX_TOOL_ITERATIONS,
|
|
)
|
|
messages.append(
|
|
{"role": "system", "content": _FINALIZE_INSTRUCTION},
|
|
)
|
|
|
|
# See note above on agent.model_id vs llm.model_id.
|
|
response = agent.llm.gen_stream(
|
|
model=getattr(agent.llm, "model_id", None) or agent.model_id,
|
|
messages=messages,
|
|
tools=(
|
|
None
|
|
if cap_reached
|
|
or getattr(agent, "context_limit_reached", False)
|
|
else agent.tools
|
|
),
|
|
)
|
|
self.llm_calls.append(build_stack_data(agent.llm))
|
|
|
|
yield from self.handle_streaming(
|
|
agent, response, tools_dict, messages,
|
|
_iteration=next_iteration,
|
|
)
|
|
return
|
|
if parsed.content:
|
|
buffer += parsed.content
|
|
yield buffer
|
|
buffer = ""
|
|
if parsed.finish_reason == "stop":
|
|
return
|