wehub-resource-sync
e4dcfc49aa
Tests / Import Check (Python 3.13) (push) Has been cancelled
Tests / Import Check (Python 3.14) (push) Has been cancelled
Tests / Python Tests (Python 3.11) (push) Has been cancelled
Tests / Python Tests (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.14) (push) Has been cancelled
Tests / Test Summary (push) Has been cancelled
Tests / Lint and Format (push) Has been cancelled
Tests / Web Node Tests (push) Has been cancelled
Tests / Import Check (Python 3.11) (push) Has been cancelled
Tests / Import Check (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.13) (push) Has been cancelled
1302 lines
52 KiB
Python
1302 lines
52 KiB
Python
"""Chat capability assembly for the exploring-loop agent."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
from typing import Any
|
||
|
||
from deeptutor.agents._shared.tool_composition import (
|
||
ToolMountFlags,
|
||
compose_enabled_tools,
|
||
default_optional_tools,
|
||
user_has_memory,
|
||
user_has_notebooks,
|
||
)
|
||
from deeptutor.agents.chat.agent_loop import AgentLoop
|
||
from deeptutor.agents.chat.prompt_blocks import ChatPromptAssembler
|
||
from deeptutor.capabilities import (
|
||
LoopCapability,
|
||
active_loop_capabilities,
|
||
any_exclusive_capability_active,
|
||
)
|
||
from deeptutor.core.agentic import (
|
||
DispatchOutcome,
|
||
LLMClientConfig,
|
||
UsageTracker,
|
||
build_completion_kwargs,
|
||
build_openai_client,
|
||
can_use_native_tool_calling,
|
||
dispatch_tool_calls,
|
||
)
|
||
from deeptutor.core.agentic.tool_dispatch import MAX_PARALLEL_TOOL_CALLS
|
||
from deeptutor.core.context import UnifiedContext
|
||
from deeptutor.core.stream_bus import StreamBus
|
||
from deeptutor.core.trace import (
|
||
build_trace_metadata,
|
||
derive_trace_metadata,
|
||
merge_trace_metadata,
|
||
new_call_id,
|
||
)
|
||
from deeptutor.runtime.registry.deferred_tools import (
|
||
DeferredToolLoader,
|
||
render_deferred_tools_manifest,
|
||
)
|
||
from deeptutor.runtime.registry.tool_registry import get_tool_registry
|
||
from deeptutor.services.config import get_chat_params
|
||
from deeptutor.services.llm import (
|
||
get_llm_config,
|
||
get_token_limit_kwargs, # noqa: F401 (re-exported for tests)
|
||
prepare_multimodal_messages,
|
||
supports_tools, # noqa: F401 (re-exported for tests)
|
||
)
|
||
from deeptutor.services.llm.context_window import resolve_effective_context_window
|
||
from deeptutor.services.prompt import get_prompt_manager
|
||
from deeptutor.tools.builtin import PARTNER_BUILTIN_TOOL_NAMES
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Chat memory tools a partner turn replaces with the partner_* variants.
|
||
_PARTNER_SUPPRESSED_TOOLS: tuple[str, ...] = ("read_memory", "write_memory")
|
||
|
||
|
||
CHAT_EXCLUDED_TOOLS: set[str] = set()
|
||
CHAT_OPTIONAL_TOOLS = default_optional_tools(excluded=CHAT_EXCLUDED_TOOLS)
|
||
|
||
# Generation tools are user-toggleable + grant-gated, but only usable once an
|
||
# admin has configured an active model for the service. Drop them from a turn's
|
||
# tool list when unconfigured so the model never sees a tool that can only error.
|
||
_GENERATION_TOOL_SERVICES: dict[str, str] = {"imagegen": "imagegen", "videogen": "videogen"}
|
||
|
||
|
||
def _drop_unconfigured_generation_tools(tools: list[str]) -> list[str]:
|
||
present = [name for name in tools if name in _GENERATION_TOOL_SERVICES]
|
||
if not present:
|
||
return tools
|
||
try:
|
||
from deeptutor.services.config.model_catalog import get_model_catalog_service
|
||
|
||
service = get_model_catalog_service()
|
||
catalog = service.load()
|
||
configured = {
|
||
name
|
||
for name in present
|
||
if (service.get_active_model(catalog, _GENERATION_TOOL_SERVICES[name]) or {}).get(
|
||
"model"
|
||
)
|
||
}
|
||
except Exception:
|
||
logger.debug("generation-tool config probe failed; dropping them", exc_info=True)
|
||
configured = set()
|
||
return [name for name in tools if name not in _GENERATION_TOOL_SERVICES or name in configured]
|
||
|
||
|
||
KB_SEED_MAX_KBS = 3
|
||
KB_SEED_CHARS_PER_KB = 4000
|
||
# Exploring-loop budget: max LLM rounds in one turn's loop. A round without
|
||
# tool calls ends the loop early — that is the normal exit.
|
||
DEFAULT_MAX_ROUNDS = 8
|
||
CONTEXT_WINDOW_GUARD_RATIO = 0.9
|
||
_DispatchOutcome = DispatchOutcome
|
||
|
||
|
||
def _read_int(cfg: Any, *, key: str, default: int) -> int:
|
||
if isinstance(cfg, dict):
|
||
value = cfg.get(key, default)
|
||
else:
|
||
value = default
|
||
try:
|
||
return int(value)
|
||
except (TypeError, ValueError):
|
||
return default
|
||
|
||
|
||
def _normalise_user_reply(raw: Any) -> tuple[str, list[dict[str, str]] | None]:
|
||
if isinstance(raw, str):
|
||
return raw, None
|
||
if isinstance(raw, dict):
|
||
text = str(raw.get("text") or "")
|
||
answers_raw = raw.get("answers")
|
||
if isinstance(answers_raw, list) and answers_raw:
|
||
answers: list[dict[str, str]] = []
|
||
for entry in answers_raw:
|
||
if not isinstance(entry, dict):
|
||
continue
|
||
qid = str(entry.get("questionId") or entry.get("id") or "").strip()
|
||
if qid:
|
||
answers.append({"questionId": qid, "text": str(entry.get("text") or "")})
|
||
return text, answers or None
|
||
return text, None
|
||
return str(raw or ""), None
|
||
|
||
|
||
def _prompt_text(prompts: dict[str, Any], path: tuple[str, ...], default: str) -> str:
|
||
value: Any = prompts
|
||
for key in path:
|
||
if not isinstance(value, dict):
|
||
return default
|
||
value = value.get(key)
|
||
return value if isinstance(value, str) and value else default
|
||
|
||
|
||
def _format_user_reply_body(
|
||
text: str,
|
||
answers: list[dict[str, str]] | None,
|
||
ask_user_payload: dict[str, Any],
|
||
*,
|
||
prompts: dict[str, Any] | None = None,
|
||
) -> str:
|
||
prompt_map = prompts or {}
|
||
empty = _prompt_text(prompt_map, ("empty", "empty_reply"), "(empty reply)")
|
||
skipped = _prompt_text(prompt_map, ("empty", "skipped_reply"), "(skipped)")
|
||
question_fallback = _prompt_text(prompt_map, ("empty", "question_fallback"), "(question)")
|
||
user_answered = _prompt_text(prompt_map, ("empty", "user_answered"), "User answered:")
|
||
if answers:
|
||
prompts_by_id: dict[str, str] = {}
|
||
for q in ask_user_payload.get("questions") or []:
|
||
if isinstance(q, dict):
|
||
qid = str(q.get("id") or "")
|
||
prompts_by_id[qid] = str(q.get("prompt") or qid)
|
||
lines = [user_answered]
|
||
for entry in answers:
|
||
qid = entry.get("questionId", "")
|
||
prompt = prompts_by_id.get(qid) or qid or question_fallback
|
||
value = (entry.get("text") or "").strip() or skipped
|
||
lines.append(f"- {prompt}\n -> {value}")
|
||
return "\n".join(lines)
|
||
flat = (text or "").strip() or empty
|
||
return f"{user_answered} {flat}"
|
||
|
||
|
||
def _flatten_ask_user_summary(ask_user_payload: dict[str, Any]) -> str:
|
||
questions = ask_user_payload.get("questions") or []
|
||
if isinstance(questions, list) and questions:
|
||
prompts = [str(q.get("prompt") or "") for q in questions if isinstance(q, dict)]
|
||
prompts = [p for p in prompts if p]
|
||
if prompts:
|
||
return " | ".join(prompts)
|
||
return str(ask_user_payload.get("question") or "")
|
||
|
||
|
||
class AgenticChatPipeline:
|
||
"""Run chat as one exploring agent loop followed by a respond stage."""
|
||
|
||
def __init__(
|
||
self,
|
||
language: str = "en",
|
||
*,
|
||
max_rounds: int | None = None,
|
||
temperature: float | None = None,
|
||
max_tokens: int | None = None,
|
||
) -> None:
|
||
self.language = "zh" if language.lower().startswith("zh") else "en"
|
||
self.llm_config = get_llm_config()
|
||
self.binding = getattr(self.llm_config, "binding", None) or "openai"
|
||
self.model = getattr(self.llm_config, "model", None)
|
||
self.api_key = getattr(self.llm_config, "api_key", None)
|
||
self.base_url = getattr(self.llm_config, "base_url", None)
|
||
self.api_version = getattr(self.llm_config, "api_version", None)
|
||
self.extra_headers = getattr(self.llm_config, "extra_headers", None) or {}
|
||
self.reasoning_effort = getattr(self.llm_config, "reasoning_effort", None)
|
||
self.registry = get_tool_registry()
|
||
self._usage = UsageTracker(model=self.model)
|
||
self._deferred_loader: DeferredToolLoader | None = None
|
||
self._deferred_pool: list[Any] = []
|
||
self._exec_enabled = False
|
||
|
||
try:
|
||
chat_cfg = get_chat_params()
|
||
except Exception as exc:
|
||
logger.warning("Failed to load chat params, using defaults: %s", exc)
|
||
chat_cfg = {}
|
||
try:
|
||
self._chat_temperature = float(chat_cfg.get("temperature", 0.2))
|
||
except (TypeError, ValueError):
|
||
self._chat_temperature = 0.2
|
||
self._max_rounds = _read_int(chat_cfg, key="max_rounds", default=DEFAULT_MAX_ROUNDS)
|
||
self._exploring_max_tokens = _read_int(
|
||
chat_cfg.get("exploring"), key="max_tokens", default=1600
|
||
)
|
||
self._respond_max_tokens = _read_int(
|
||
chat_cfg.get("responding"), key="max_tokens", default=8000
|
||
)
|
||
# Per-capability overrides (e.g. deep solve forwards its own round
|
||
# budget / temperature / answer-token cap, read from the solve
|
||
# settings). Chat itself passes none and keeps the chat_cfg values.
|
||
if max_rounds is not None:
|
||
self._max_rounds = max(1, int(max_rounds))
|
||
if temperature is not None:
|
||
self._chat_temperature = float(temperature)
|
||
if max_tokens is not None:
|
||
self._respond_max_tokens = max(256, int(max_tokens))
|
||
|
||
try:
|
||
self._prompts: dict[str, Any] = (
|
||
get_prompt_manager().load_prompts(
|
||
module_name="chat",
|
||
agent_name="agentic_chat",
|
||
language=self.language,
|
||
)
|
||
or {}
|
||
)
|
||
except Exception as exc:
|
||
logger.warning("Failed to load agentic_chat prompts: %s", exc)
|
||
self._prompts = {}
|
||
self._prompt_assembler = ChatPromptAssembler(
|
||
prompts=self._prompts,
|
||
language=self.language,
|
||
)
|
||
self._client_config = LLMClientConfig(
|
||
binding=self.binding,
|
||
model=self.model,
|
||
api_key=self.api_key,
|
||
base_url=self.base_url,
|
||
api_version=self.api_version,
|
||
extra_headers=self.extra_headers or None,
|
||
reasoning_effort=self.reasoning_effort,
|
||
)
|
||
|
||
@property
|
||
def usage(self) -> UsageTracker:
|
||
return self._usage
|
||
|
||
@property
|
||
def max_rounds(self) -> int:
|
||
return max(1, self._max_rounds)
|
||
|
||
def effective_max_rounds(self, context: UnifiedContext) -> int:
|
||
"""Round budget for this turn, lifted to satisfy any capability minimum.
|
||
|
||
A capability that needs guaranteed loop headroom — the subagent
|
||
capability, which must allow its full consult budget plus a finishing
|
||
round — sets ``context.metadata["_min_loop_rounds"]``; the loop honours
|
||
the larger of that and the configured budget. A generic seam (like
|
||
solve's ``solve_max_replans``) so the loop stays capability-agnostic.
|
||
"""
|
||
try:
|
||
floor = int(context.metadata.get("_min_loop_rounds") or 0)
|
||
except (TypeError, ValueError):
|
||
floor = 0
|
||
return max(self.max_rounds, floor)
|
||
|
||
@property
|
||
def exploring_max_tokens(self) -> int:
|
||
return max(128, self._exploring_max_tokens)
|
||
|
||
@property
|
||
def respond_max_tokens(self) -> int:
|
||
return max(256, self._respond_max_tokens)
|
||
|
||
@property
|
||
def loop_max_tokens(self) -> int:
|
||
"""Single per-round token budget for the merged loop.
|
||
|
||
The loop has no separate exploring/respond split, so every round —
|
||
including the round that writes the final answer — uses one budget.
|
||
It must be large enough for a full answer; the responding budget is
|
||
that ceiling (tool-only rounds rarely approach it).
|
||
"""
|
||
return self.respond_max_tokens
|
||
|
||
async def run(self, context: UnifiedContext, stream: StreamBus) -> None:
|
||
await self._prepare_deferred_tools(context)
|
||
self._exec_enabled = await self._exec_allowed(context)
|
||
enabled_tools = self._compose_enabled_tools(context)
|
||
use_native_tools = bool(enabled_tools) and self._can_use_native_tool_calling()
|
||
tool_schemas = (
|
||
self._build_llm_tool_schemas(enabled_tools, context) if use_native_tools else None
|
||
)
|
||
if tool_schemas is not None and self._deferred_loader is not None:
|
||
tool_schemas.extend(self._deferred_loader.initial_schemas())
|
||
self._deferred_loader.bind_live_schemas(tool_schemas)
|
||
|
||
loop = AgentLoop(
|
||
pipeline=self,
|
||
context=context,
|
||
stream=stream,
|
||
client=self._build_openai_client(),
|
||
enabled_tools=enabled_tools if use_native_tools else [],
|
||
tool_schemas=tool_schemas,
|
||
)
|
||
await loop.run()
|
||
|
||
# ---- prompt assembly -------------------------------------------------
|
||
|
||
def _build_system_prompt(
|
||
self,
|
||
enabled_tools: list[str],
|
||
context: UnifiedContext,
|
||
*,
|
||
include_tool_manifest: bool = True,
|
||
) -> str:
|
||
return self._prompt_assembler.system_prompt(
|
||
context=context,
|
||
tool_manifest=self._tool_manifest(enabled_tools),
|
||
kb_note=self._kb_system_note(context),
|
||
deferred_tools_manifest=(
|
||
self._deferred_tools_manifest() if include_tool_manifest else ""
|
||
),
|
||
notebook_manifest=self._build_notebook_manifest(),
|
||
workspace_note=self._workspace_system_note(context),
|
||
capability_blocks=self._capability_system_blocks(context),
|
||
include_tool_manifest=include_tool_manifest,
|
||
)
|
||
|
||
def _build_loop_messages(
|
||
self,
|
||
*,
|
||
context: UnifiedContext,
|
||
enabled_tools: list[str],
|
||
kb_seed: str = "",
|
||
include_tool_manifest: bool = True,
|
||
) -> list[dict[str, Any]]:
|
||
"""Build the turn's ONE conversation.
|
||
|
||
The loop appends each round (assistant + ``role=tool`` results) to
|
||
this list, so the system prompt stays byte-stable for the whole turn
|
||
and the KB cache prefix is preserved. The KB seed rides inside the
|
||
trailing user message, not the system prompt.
|
||
"""
|
||
system_prompt = self._build_system_prompt(
|
||
enabled_tools,
|
||
context,
|
||
include_tool_manifest=include_tool_manifest,
|
||
)
|
||
user_content = self._prompt_assembler.user_message(
|
||
context=context,
|
||
kb_seed=kb_seed,
|
||
)
|
||
messages: list[dict[str, Any]] = [{"role": "system", "content": system_prompt}]
|
||
for item in context.conversation_history:
|
||
role = item.get("role")
|
||
content = item.get("content")
|
||
if role in {"user", "assistant"} and isinstance(content, (str, list)):
|
||
messages.append({"role": role, "content": content})
|
||
elif role == "system" and isinstance(content, str) and content.strip():
|
||
# ContextBuilder emits the compressed-history summary as a
|
||
# leading system message; deliver it right after the system
|
||
# prompt so compacted turns stay visible to the model.
|
||
header = _prompt_text(
|
||
self._prompts,
|
||
("notices", "conversation_summary_header"),
|
||
"[Conversation summary]",
|
||
)
|
||
messages.append({"role": "system", "content": f"{header}\n{content}"})
|
||
messages.append({"role": "user", "content": user_content})
|
||
return self._prepare_messages_with_attachments(messages, context)
|
||
|
||
def _finish_exhausted_instruction(self) -> str:
|
||
return self._prompt_assembler.finish_exhausted_instruction()
|
||
|
||
def _tool_manifest(self, enabled_tools: list[str]) -> str:
|
||
names = list(enabled_tools)
|
||
if self._deferred_loader is not None:
|
||
for name in sorted(self._deferred_loader.loaded_names):
|
||
if name not in names:
|
||
names.append(name)
|
||
try:
|
||
return self.registry.build_prompt_text(
|
||
names,
|
||
format="list_with_usage",
|
||
language=self.language,
|
||
)
|
||
except TypeError:
|
||
return self.registry.build_prompt_text(names)
|
||
except Exception:
|
||
logger.warning("failed to build tool prompt text", exc_info=True)
|
||
return ""
|
||
|
||
def _tool_result_snip_marker(self) -> str:
|
||
return self._t(
|
||
"notices.tool_result_snipped",
|
||
default=(
|
||
"[earlier tool result snipped to stay within context window; "
|
||
"call the same tool again if the content is still needed]"
|
||
),
|
||
)
|
||
|
||
def _prepare_messages_with_attachments(
|
||
self,
|
||
messages: list[dict[str, Any]],
|
||
context: UnifiedContext,
|
||
) -> list[dict[str, Any]]:
|
||
return prepare_multimodal_messages(
|
||
messages,
|
||
context.attachments,
|
||
binding=self.binding,
|
||
model=self.model,
|
||
).messages
|
||
|
||
# ---- deferred tools / tool composition ------------------------------
|
||
|
||
@staticmethod
|
||
def _is_partner_turn(context: UnifiedContext) -> bool:
|
||
"""Whether this turn runs under a partner's synthetic scope.
|
||
|
||
A partner turn executes as a synthetic non-admin user but acts as the
|
||
admin owner's extension. Authorization for these turns travels through
|
||
context metadata (the owner-scoped ``mcp_tools_filter`` / exec gate),
|
||
not the synthetic user's grant file — so callers must bypass real-user
|
||
grant resolution and defer to that metadata whitelist instead.
|
||
"""
|
||
return str((context.metadata or {}).get("source") or "") == "partner"
|
||
|
||
async def _prepare_deferred_tools(self, context: UnifiedContext) -> None:
|
||
try:
|
||
from deeptutor.services.mcp import get_mcp_manager, load_loaded_tools
|
||
|
||
await get_mcp_manager().ensure_started()
|
||
# Caller-scoped whitelist (e.g. a partner's configured MCP tools)
|
||
# intersected with the current user's grant. ``None`` means
|
||
# unrestricted; a set narrows the deferred tools. Real non-admin
|
||
# users fail closed when no MCP grant is present, while partner
|
||
# turns defer to their owner-scoped metadata whitelist as the
|
||
# authority (see ``_is_partner_turn``).
|
||
from deeptutor.multi_user.tool_access import allowed_mcp_tools, combine_whitelists
|
||
|
||
raw_filter = context.metadata.get("mcp_tools_filter")
|
||
caller_allowed = (
|
||
{str(name) for name in raw_filter} if isinstance(raw_filter, list) else None
|
||
)
|
||
user_allowed = None if self._is_partner_turn(context) else allowed_mcp_tools()
|
||
allowed: set[str] | None = combine_whitelists(caller_allowed, user_allowed)
|
||
pool = self.registry.deferred_tools()
|
||
if allowed is not None:
|
||
pool = [t for t in pool if t.get_definition().name in allowed]
|
||
self._deferred_pool = pool
|
||
if not pool:
|
||
self._deferred_loader = None
|
||
return
|
||
self._deferred_loader = DeferredToolLoader(
|
||
registry=self.registry,
|
||
session_id=context.session_id,
|
||
loaded=load_loaded_tools(context.session_id),
|
||
allowed=allowed,
|
||
)
|
||
except Exception:
|
||
logger.warning("deferred-tool preparation failed", exc_info=True)
|
||
self._deferred_loader = None
|
||
|
||
def _deferred_tools_manifest(self) -> str:
|
||
if self._deferred_loader is None:
|
||
return ""
|
||
return render_deferred_tools_manifest(
|
||
getattr(self, "_deferred_pool", None) or self.registry.deferred_tools(),
|
||
language=self.language,
|
||
)
|
||
|
||
async def _exec_allowed(self, context: UnifiedContext) -> bool:
|
||
try:
|
||
from deeptutor.services.sandbox import IsolationLevel, get_sandbox_service
|
||
|
||
# A partner turn runs as a synthetic non-admin user but IS the admin
|
||
# owner's extension (partners are anchored to the admin workspace), so
|
||
# exec follows the owner's authority — not the partner's "user" role.
|
||
# The owner still gates exec per-partner via the builtin-tool whitelist.
|
||
is_partner = self._is_partner_turn(context)
|
||
|
||
level = await get_sandbox_service().isolation_level()
|
||
if level is IsolationLevel.SYSTEM:
|
||
# Admin can switch exec off per user (grant v2). ``None``
|
||
# follows the policy: SYSTEM isolation serves everyone.
|
||
from deeptutor.multi_user.tool_access import exec_override
|
||
|
||
return exec_override() is not False
|
||
if level is IsolationLevel.APPLICATION:
|
||
if is_partner:
|
||
return True
|
||
try:
|
||
from deeptutor.multi_user.context import get_current_user
|
||
|
||
return bool(get_current_user().is_admin)
|
||
except Exception:
|
||
# Single-user local runtime: APPLICATION isolation is the
|
||
# same explicit opt-in posture TutorBot uses for local dev.
|
||
return True
|
||
return False
|
||
except Exception:
|
||
logger.warning("exec policy gate failed; disabling exec", exc_info=True)
|
||
return False
|
||
|
||
def _compose_enabled_tools(self, context: UnifiedContext) -> list[str]:
|
||
is_partner = self._is_partner_turn(context)
|
||
composed = compose_enabled_tools(
|
||
registry=self.registry,
|
||
requested_tools=context.enabled_tools,
|
||
optional_whitelist=CHAT_OPTIONAL_TOOLS,
|
||
mount_flags=ToolMountFlags(
|
||
has_kb=bool(self._selected_kbs(context)),
|
||
# read_source is owned by the explore_context pre-pass (it runs
|
||
# the investigation over attached sources), not the answer loop.
|
||
# Keep it off the answer surface even when sources are present.
|
||
has_sources=False,
|
||
has_memory=user_has_memory(),
|
||
has_notebooks=user_has_notebooks(),
|
||
has_skills=bool(context.skills_manifest),
|
||
has_deferred_tools=getattr(self, "_deferred_loader", None) is not None,
|
||
has_exec=getattr(self, "_exec_enabled", False),
|
||
has_code=getattr(self, "_exec_enabled", False),
|
||
),
|
||
capability_owned=self._capability_owned_tools(context),
|
||
exclusive=self._exclusive_capability_active(context),
|
||
builtin_whitelist=(
|
||
set(context.allowed_builtin_tools)
|
||
if context.allowed_builtin_tools is not None
|
||
else None
|
||
),
|
||
# Partners get the partner_* memory/history tools force-mounted and
|
||
# chat's read_memory/write_memory suppressed — the split-memory model
|
||
# (own workspace writable, owner's memory read-only) lives in those
|
||
# tools, not in chat's.
|
||
forced=PARTNER_BUILTIN_TOOL_NAMES if is_partner else (),
|
||
suppressed=_PARTNER_SUPPRESSED_TOOLS if is_partner else (),
|
||
)
|
||
return _drop_unconfigured_generation_tools(composed)
|
||
|
||
def _active_loop_capabilities(self, context: UnifiedContext) -> tuple[LoopCapability, ...]:
|
||
return active_loop_capabilities(context)
|
||
|
||
@staticmethod
|
||
def _exclusive_capability_active(context: UnifiedContext) -> bool:
|
||
"""True when a knowledge capability owns the turn (replaces the surface).
|
||
|
||
Suppresses rag scaffolding (KB seed / kb note) too — rag isn't mounted,
|
||
so seeding or advertising it would be wrong.
|
||
"""
|
||
return any_exclusive_capability_active(context)
|
||
|
||
def _capability_owned_tools(self, context: UnifiedContext) -> tuple[str, ...]:
|
||
"""The active capabilities' own tools — added on top of chat's full surface."""
|
||
names: list[str] = []
|
||
for cap in self._active_loop_capabilities(context):
|
||
names.extend(cap.owned_tools)
|
||
return tuple(names)
|
||
|
||
def _capability_system_blocks(self, context: UnifiedContext):
|
||
blocks = []
|
||
for cap in self._active_loop_capabilities(context):
|
||
block = cap.system_block(
|
||
context,
|
||
language=self.language,
|
||
prompts=self._prompts,
|
||
)
|
||
if block is not None:
|
||
blocks.append(block)
|
||
return blocks
|
||
|
||
def _capability_pre_loop_seed(self, context: UnifiedContext) -> str:
|
||
seeds = [
|
||
seed.strip()
|
||
for cap in self._active_loop_capabilities(context)
|
||
if (seed := cap.pre_loop_seed(context))
|
||
]
|
||
return "\n\n".join(seed for seed in seeds if seed)
|
||
|
||
async def _capability_pre_loop_briefings(
|
||
self,
|
||
context: UnifiedContext,
|
||
stream: StreamBus,
|
||
) -> str:
|
||
"""Run each active capability's optional async ``pre_loop`` hook and
|
||
join their returned blocks into one seed fragment.
|
||
|
||
The hook is optional (read via ``getattr`` so plain capabilities are
|
||
unaffected) and runs once before the answer loop's first LLM call —
|
||
see the ``pre_loop`` note on :class:`LoopCapability`. Failures are
|
||
swallowed: a pre-pass is best-effort grounding and must never sink the
|
||
turn.
|
||
"""
|
||
blocks: list[str] = []
|
||
for cap in self._active_loop_capabilities(context):
|
||
hook = getattr(cap, "pre_loop", None)
|
||
if not callable(hook):
|
||
continue
|
||
try:
|
||
block = await hook(context, stream, usage=self._usage)
|
||
except Exception:
|
||
logger.warning(
|
||
"pre_loop hook failed for capability %s",
|
||
getattr(cap, "name", "?"),
|
||
exc_info=True,
|
||
)
|
||
continue
|
||
content = (getattr(block, "content", "") or "").strip()
|
||
if content:
|
||
blocks.append(content)
|
||
return "\n\n".join(blocks)
|
||
|
||
def _build_llm_tool_schemas(
|
||
self,
|
||
enabled_tools: list[str],
|
||
context: UnifiedContext,
|
||
) -> list[dict[str, Any]]:
|
||
schemas = self.registry.build_openai_schemas(enabled_tools)
|
||
kb_choices = self._selected_kbs(context)
|
||
notebook_choices = self._notebook_choices()
|
||
for schema in schemas:
|
||
function = schema.get("function") if isinstance(schema, dict) else None
|
||
if not isinstance(function, dict):
|
||
continue
|
||
parameters = function.get("parameters")
|
||
if not isinstance(parameters, dict):
|
||
continue
|
||
properties = parameters.get("properties") or {}
|
||
if function.get("name") == "rag" and isinstance(properties, dict):
|
||
if isinstance(properties.get("query"), dict):
|
||
properties["query"].setdefault("minLength", 1)
|
||
if isinstance(properties.get("kb_name"), dict):
|
||
properties["kb_name"]["enum"] = kb_choices
|
||
if function.get("name") == "geogebra_analysis" and isinstance(properties, dict):
|
||
properties.pop("image_base64", None)
|
||
required = parameters.get("required")
|
||
if isinstance(required, list):
|
||
parameters["required"] = [n for n in required if n != "image_base64"]
|
||
if (
|
||
function.get("name") in {"list_notebook", "write_note"}
|
||
and isinstance(properties, dict)
|
||
and notebook_choices
|
||
and isinstance(properties.get("notebook_id"), dict)
|
||
):
|
||
nb_schema = properties["notebook_id"]
|
||
nb_schema["enum"] = [choice["id"] for choice in notebook_choices]
|
||
rendered = "; ".join(f"{c['id']} = {c['name']}" for c in notebook_choices)
|
||
nb_schema["description"] = (
|
||
f"{nb_schema.get('description', '').rstrip(' .')}. Available: {rendered}."
|
||
)
|
||
parameters["additionalProperties"] = False
|
||
return schemas
|
||
|
||
# ---- notebook / context helpers -------------------------------------
|
||
|
||
def _build_notebook_manifest(self) -> str:
|
||
choices = self._notebook_choices_full()
|
||
if not choices:
|
||
return ""
|
||
capped = choices[:30]
|
||
lines = ["[用户的笔记本列表]" if self.language == "zh" else "[User's notebooks]"]
|
||
for entry in capped:
|
||
nid = entry.get("id", "")
|
||
name = entry.get("name", nid)
|
||
count = entry.get("record_count", 0)
|
||
lines.append(f"- `{nid}` - {name} ({count} records)")
|
||
if len(choices) > len(capped):
|
||
lines.append(
|
||
f"... (+{len(choices) - len(capped)} more; call `list_notebook` to see the rest)"
|
||
)
|
||
return "\n".join(lines)
|
||
|
||
@staticmethod
|
||
def _notebook_choices_full() -> list[dict[str, Any]]:
|
||
try:
|
||
from deeptutor.services.notebook import get_notebook_manager
|
||
|
||
notebooks = get_notebook_manager().list_notebooks() or []
|
||
except Exception:
|
||
return []
|
||
rows: list[dict[str, Any]] = []
|
||
for nb in notebooks:
|
||
nid = str(nb.get("id") or "").strip()
|
||
if not nid:
|
||
continue
|
||
name = str(nb.get("name") or nb.get("title") or nid).strip() or nid
|
||
try:
|
||
count = int(nb.get("record_count") or 0)
|
||
except (TypeError, ValueError):
|
||
count = 0
|
||
rows.append({"id": nid, "name": name, "record_count": count})
|
||
return rows
|
||
|
||
@staticmethod
|
||
def _notebook_choices() -> list[dict[str, str]]:
|
||
return [
|
||
{"id": str(row["id"]), "name": str(row["name"])}
|
||
for row in AgenticChatPipeline._notebook_choices_full()
|
||
]
|
||
|
||
# ---- tool execution --------------------------------------------------
|
||
|
||
async def _execute_tool_call(
|
||
self,
|
||
tool_name: str,
|
||
tool_args: dict[str, Any],
|
||
*,
|
||
stream: StreamBus | None = None,
|
||
retrieve_meta: dict[str, Any] | None = None,
|
||
) -> dict[str, Any]:
|
||
from deeptutor.core.agentic import execute_tool_call
|
||
|
||
stream = stream or StreamBus()
|
||
return await execute_tool_call(
|
||
registry=self.registry,
|
||
tool_name=tool_name,
|
||
tool_args=tool_args,
|
||
stream=stream,
|
||
source="chat",
|
||
stage="responding",
|
||
retrieve_meta=retrieve_meta,
|
||
empty_tool_result_message=self._t("notices.empty_tool_result"),
|
||
start_retrieval_message=self._t(
|
||
"notices.start_retrieval", default="Starting retrieval"
|
||
),
|
||
retrieve_label=self._t("labels.retrieve", default="Retrieve"),
|
||
unknown_error_message_factory=lambda tn: self._t(
|
||
"notices.tool_unknown_error",
|
||
tool=tn,
|
||
default=f"An unknown error occurred while executing {tn}.",
|
||
),
|
||
)
|
||
|
||
async def _dispatch_tool_calls(
|
||
self,
|
||
*,
|
||
tool_calls: list[dict[str, Any]],
|
||
context: UnifiedContext,
|
||
stream: StreamBus,
|
||
iteration_index: int,
|
||
stage: str = "exploring",
|
||
) -> DispatchOutcome:
|
||
too_many = None
|
||
if len(tool_calls) > MAX_PARALLEL_TOOL_CALLS:
|
||
too_many = self._t(
|
||
"notices.too_many_tool_calls",
|
||
requested=len(tool_calls),
|
||
limit=MAX_PARALLEL_TOOL_CALLS,
|
||
)
|
||
return await dispatch_tool_calls(
|
||
tool_calls=tool_calls,
|
||
context=context,
|
||
stream=stream,
|
||
source="chat",
|
||
stage=stage,
|
||
iteration_index=iteration_index,
|
||
registry=self.registry,
|
||
kwarg_augmenter=self._augment_tool_kwargs,
|
||
retrieve_meta_factory=lambda meta, tn, ta: self._retrieve_trace_metadata(
|
||
meta, context=context, tool_name=tn, tool_args=ta
|
||
),
|
||
tool_call_label=self._t("labels.tool_call", default="Tool call"),
|
||
retrieve_label=self._t("labels.retrieve", default="Retrieve"),
|
||
empty_tool_result_message=self._t("notices.empty_tool_result"),
|
||
start_retrieval_message=self._t(
|
||
"notices.start_retrieval", default="Starting retrieval"
|
||
),
|
||
too_many_tool_calls_message=too_many,
|
||
unknown_error_message_factory=lambda tn: self._t(
|
||
"notices.tool_unknown_error",
|
||
tool=tn,
|
||
default=f"An unknown error occurred while executing {tn}.",
|
||
),
|
||
trace_id_prefix="chat-loop",
|
||
)
|
||
|
||
async def _await_user_reply_and_resolve(
|
||
self,
|
||
*,
|
||
context: UnifiedContext,
|
||
stream: StreamBus,
|
||
dispatch: DispatchOutcome,
|
||
) -> bool:
|
||
ask_user = (dispatch.pause_payload or {}).get("ask_user") or {}
|
||
waiter = context.metadata.get("wait_for_user_reply")
|
||
if not callable(waiter):
|
||
await self._emit_terminator_final_response(
|
||
stream,
|
||
{
|
||
"tool_name": (dispatch.pause_payload or {}).get("tool_name", "ask_user"),
|
||
"content": _flatten_ask_user_summary(ask_user),
|
||
"metadata": {"ask_user": ask_user},
|
||
},
|
||
)
|
||
return False
|
||
|
||
raw_reply = await waiter()
|
||
if raw_reply is None:
|
||
return False
|
||
reply_text, answers = _normalise_user_reply(raw_reply)
|
||
body_text = _format_user_reply_body(
|
||
reply_text,
|
||
answers,
|
||
ask_user,
|
||
prompts=self._prompts,
|
||
)
|
||
continue_directive = self._t(
|
||
"notices.ask_user_resolved_directive",
|
||
default=(
|
||
"[ask_user resolved. Continue the user's original request using these answers. "
|
||
"Do not stop with an acknowledgement.]"
|
||
),
|
||
)
|
||
directive = f"{body_text}\n\n{continue_directive}"
|
||
for tm in dispatch.tool_messages:
|
||
if tm.get("tool_call_id") == dispatch.pause_tool_call_id:
|
||
tm["content"] = directive
|
||
break
|
||
meta: dict[str, Any] = {
|
||
"trace_kind": "user_reply",
|
||
"ask_user_resolved": True,
|
||
"ask_user_tool_call_id": dispatch.pause_tool_call_id,
|
||
"reply_preview": (reply_text or "")[:200],
|
||
}
|
||
if answers:
|
||
meta["answers"] = list(answers)
|
||
await stream.progress("", source="chat", stage="responding", metadata=meta)
|
||
return True
|
||
|
||
def _augment_tool_kwargs(
|
||
self,
|
||
tool_name: str,
|
||
args: dict[str, Any],
|
||
context: UnifiedContext,
|
||
) -> dict[str, Any]:
|
||
from deeptutor.services.path_service import get_path_service
|
||
|
||
kwargs = dict(args)
|
||
turn_id = str(context.metadata.get("turn_id", "") or "").strip()
|
||
workspace_key = self._workspace_key(context)
|
||
task_dir = (
|
||
get_path_service().get_task_workspace("chat", workspace_key) if workspace_key else None
|
||
)
|
||
exec_dir = task_dir / "exec" if task_dir is not None else None
|
||
if tool_name == "rag":
|
||
kwargs.setdefault("mode", "hybrid")
|
||
elif tool_name == "load_tools":
|
||
kwargs["_tool_loader"] = self._deferred_loader
|
||
elif tool_name == "exec":
|
||
from deeptutor.services.sandbox import Mount
|
||
|
||
kwargs["_sandbox_user_id"] = self._current_user_id()
|
||
if exec_dir is not None:
|
||
exec_dir.mkdir(parents=True, exist_ok=True)
|
||
kwargs["_sandbox_workdir"] = str(exec_dir)
|
||
kwargs["_sandbox_mounts"] = (
|
||
Mount(host_path=str(exec_dir), sandbox_path=str(exec_dir), read_only=False),
|
||
)
|
||
elif tool_name == "code_execution":
|
||
from deeptutor.services.sandbox import Mount
|
||
|
||
kwargs["_sandbox_user_id"] = self._current_user_id()
|
||
code_dir = task_dir / "code_runs" if task_dir is not None else None
|
||
if code_dir is not None:
|
||
code_dir.mkdir(parents=True, exist_ok=True)
|
||
kwargs["_sandbox_workdir"] = str(code_dir)
|
||
kwargs["_sandbox_mounts"] = (
|
||
Mount(host_path=str(code_dir), sandbox_path=str(code_dir), read_only=False),
|
||
)
|
||
elif tool_name in ("imagegen", "videogen"):
|
||
# Generated media lands in the turn's public workspace so it
|
||
# surfaces as a download card via /api/outputs (same convention as
|
||
# exec/code_execution artifacts).
|
||
media_dir = task_dir / "media" if task_dir is not None else None
|
||
if media_dir is not None:
|
||
media_dir.mkdir(parents=True, exist_ok=True)
|
||
kwargs["_workspace_dir"] = str(media_dir)
|
||
elif tool_name == "cron":
|
||
# Owner routing is supplied server-side — the model never picks
|
||
# where a scheduled task's output lands.
|
||
meta = context.metadata or {}
|
||
cron_job_id = str(meta.get("cron_job_id") or meta.get("_cron_job_id") or "")
|
||
kwargs["_cron_in_context"] = bool(
|
||
cron_job_id or str(meta.get("source") or "") == "cron"
|
||
)
|
||
if self._is_partner_turn(context):
|
||
channel_meta = meta.get("channel_metadata")
|
||
kwargs["_cron_owner"] = {
|
||
"kind": "partner",
|
||
"partner_id": str(meta.get("partner_id") or ""),
|
||
"channel": str(meta.get("channel") or ""),
|
||
"chat_id": str(meta.get("chat_id") or ""),
|
||
"session_key": str(meta.get("session_key") or ""),
|
||
"channel_meta": dict(channel_meta) if isinstance(channel_meta, dict) else {},
|
||
"language": context.language or "en",
|
||
}
|
||
else:
|
||
from deeptutor.multi_user.context import get_current_user
|
||
|
||
user = get_current_user()
|
||
kwargs["_cron_owner"] = {
|
||
"kind": "chat",
|
||
"user_id": user.id,
|
||
"is_admin": user.is_admin,
|
||
"session_id": context.session_id,
|
||
"language": context.language or "en",
|
||
}
|
||
elif tool_name in {"reason", "brainstorm"}:
|
||
kwargs.setdefault("context", context.user_message)
|
||
elif tool_name == "paper_search":
|
||
kwargs.setdefault("max_results", 3)
|
||
kwargs.setdefault("years_limit", 3)
|
||
kwargs.setdefault("sort_by", "relevance")
|
||
elif tool_name == "web_search":
|
||
kwargs.setdefault("query", context.user_message)
|
||
if task_dir is not None:
|
||
kwargs.setdefault("output_dir", str(task_dir / "web_search"))
|
||
elif tool_name == "write_note":
|
||
kwargs["conversation_history"] = list(context.conversation_history or [])
|
||
kwargs["current_user_message"] = context.user_message or ""
|
||
elif tool_name == "geogebra_analysis":
|
||
first_image = next(
|
||
(
|
||
att
|
||
for att in (context.attachments or [])
|
||
if getattr(att, "type", "") == "image" and getattr(att, "base64", "")
|
||
),
|
||
None,
|
||
)
|
||
if first_image is not None:
|
||
raw_b64 = first_image.base64
|
||
if raw_b64.startswith("data:"):
|
||
kwargs["image_base64"] = raw_b64
|
||
else:
|
||
mime = getattr(first_image, "mime_type", "") or "image/png"
|
||
kwargs["image_base64"] = f"data:{mime};base64,{raw_b64}"
|
||
kwargs["language"] = context.language or "zh"
|
||
for cap in self._active_loop_capabilities(context):
|
||
kwargs = cap.augment_kwargs(tool_name, kwargs, context)
|
||
return kwargs
|
||
|
||
def _retrieve_trace_metadata(
|
||
self,
|
||
tool_meta: dict[str, Any],
|
||
*,
|
||
context: UnifiedContext,
|
||
tool_name: str,
|
||
tool_args: dict[str, Any],
|
||
) -> dict[str, Any] | None:
|
||
_ = context
|
||
if tool_name == "rag":
|
||
return derive_trace_metadata(
|
||
tool_meta,
|
||
label=self._t("labels.retrieve", default="Retrieve"),
|
||
call_kind="rag_retrieval",
|
||
trace_role="retrieve",
|
||
trace_group="retrieve",
|
||
query=str(tool_args.get("query", "") or ""),
|
||
)
|
||
# imagegen/videogen are long-running: wiring retrieve_meta gives them an
|
||
# event_sink so their progress (esp. videogen's poll loop) streams to the
|
||
# client, which resets the chat idle-timeout watchdog mid-render.
|
||
if tool_name in ("imagegen", "videogen"):
|
||
return derive_trace_metadata(
|
||
tool_meta,
|
||
label=self._t("labels.tool_call", default="Tool call"),
|
||
call_kind="media_generation",
|
||
query=str(tool_args.get("prompt", "") or ""),
|
||
)
|
||
# consult_subagent drives a live local agent that runs for as long as it
|
||
# needs: wiring retrieve_meta gives it an event_sink so every native
|
||
# output/log streams to the sidebar in real time (and keeps the
|
||
# idle-timeout watchdog fed during a long agent run).
|
||
if tool_name == "consult_subagent":
|
||
return derive_trace_metadata(
|
||
tool_meta,
|
||
label=self._t("labels.consult_subagent", default="Consult agent"),
|
||
call_kind="subagent_consult",
|
||
query=str(tool_args.get("question", "") or ""),
|
||
)
|
||
return None
|
||
|
||
# ---- KB seed ---------------------------------------------------------
|
||
|
||
async def _retrieve_kb_seed_block(
|
||
self,
|
||
context: UnifiedContext,
|
||
stream: StreamBus,
|
||
) -> str:
|
||
if self._exclusive_capability_active(context):
|
||
return ""
|
||
kbs = self._selected_kbs(context)
|
||
query = (context.user_message or "").strip()
|
||
if not kbs or not query:
|
||
return ""
|
||
if len(kbs) > KB_SEED_MAX_KBS:
|
||
kbs = kbs[:KB_SEED_MAX_KBS]
|
||
results = await asyncio.gather(*(self._seed_search_one_kb(kb, query, stream) for kb in kbs))
|
||
sections: list[str] = []
|
||
sources: list[dict[str, Any]] = []
|
||
for kb, result in zip(kbs, results, strict=False):
|
||
if result is None:
|
||
continue
|
||
text, kb_sources = result
|
||
sections.append(f"## {kb}\n{text}")
|
||
sources.extend(kb_sources)
|
||
if not sections:
|
||
return ""
|
||
if sources:
|
||
await stream.sources(
|
||
sources, source="chat", stage="responding", metadata={"trace_kind": "sources"}
|
||
)
|
||
header = self._t(
|
||
"knowledge_base_seed.header",
|
||
default=(
|
||
"[Knowledge Base Context]\n"
|
||
"Passages retrieved from attached knowledge bases for the current question."
|
||
),
|
||
)
|
||
return header + "\n\n" + "\n\n".join(sections)
|
||
|
||
async def _seed_search_one_kb(
|
||
self,
|
||
kb_name: str,
|
||
query: str,
|
||
stream: StreamBus,
|
||
) -> tuple[str, list[dict[str, Any]]] | None:
|
||
call_id = new_call_id("chat-kb-seed")
|
||
retrieve_meta = build_trace_metadata(
|
||
call_id=call_id,
|
||
phase="responding",
|
||
label=self._t("labels.retrieve", default="Retrieve"),
|
||
call_kind="rag_retrieval",
|
||
trace_id=call_id,
|
||
trace_role="retrieve",
|
||
trace_group="retrieve",
|
||
query=query,
|
||
)
|
||
result = await self._execute_tool_call(
|
||
"rag",
|
||
{"query": query, "kb_name": kb_name, "mode": "hybrid"},
|
||
stream=stream,
|
||
retrieve_meta=retrieve_meta,
|
||
)
|
||
if not result.get("success"):
|
||
return None
|
||
metadata = result.get("metadata") or {}
|
||
if metadata.get("error_type") or metadata.get("needs_reindex"):
|
||
return None
|
||
text = str(metadata.get("content") or metadata.get("answer") or "").strip()
|
||
if not text:
|
||
return None
|
||
if len(text) > KB_SEED_CHARS_PER_KB:
|
||
text = text[:KB_SEED_CHARS_PER_KB].rstrip() + "\n...[truncated]"
|
||
return text, list(result.get("sources") or [])
|
||
|
||
# ---- emissions / context guard --------------------------------------
|
||
|
||
async def _emit_final_text(
|
||
self,
|
||
stream: StreamBus,
|
||
text: str,
|
||
final_meta: dict[str, Any],
|
||
) -> None:
|
||
if not text:
|
||
return
|
||
await stream.content(
|
||
text,
|
||
source="chat",
|
||
stage="responding",
|
||
metadata=merge_trace_metadata(final_meta, {"trace_kind": "llm_output"}),
|
||
)
|
||
|
||
async def _emit_protocol_fallback_final_response(
|
||
self,
|
||
stream: StreamBus,
|
||
content: str,
|
||
) -> None:
|
||
final_meta = build_trace_metadata(
|
||
call_id=new_call_id("chat-final-response"),
|
||
phase="responding",
|
||
label=self._t("labels.final_response", default="Final response"),
|
||
call_kind="llm_final_response",
|
||
trace_id="chat-final-response",
|
||
trace_role="response",
|
||
trace_group="stage",
|
||
fallback=True,
|
||
)
|
||
await self._emit_final_text(stream, content, final_meta)
|
||
|
||
async def _emit_terminator_final_response(
|
||
self,
|
||
stream: StreamBus,
|
||
payload: dict[str, Any] | None,
|
||
) -> None:
|
||
if not payload:
|
||
return
|
||
content = str(payload.get("content") or "").strip()
|
||
if not content:
|
||
return
|
||
final_meta = build_trace_metadata(
|
||
call_id=new_call_id("chat-final-response"),
|
||
phase="responding",
|
||
label=self._t("labels.final_response", default="Final response"),
|
||
call_kind="llm_final_response",
|
||
trace_id="chat-final-response",
|
||
trace_role="response",
|
||
trace_group="stage",
|
||
terminator_tool=str(payload.get("tool_name") or ""),
|
||
)
|
||
merged: dict[str, Any] = {"trace_kind": "llm_output"}
|
||
tool_metadata = payload.get("metadata") or {}
|
||
if isinstance(tool_metadata, dict) and tool_metadata:
|
||
merged["tool_metadata"] = dict(tool_metadata)
|
||
await stream.content(
|
||
content,
|
||
source="chat",
|
||
stage="responding",
|
||
metadata=merge_trace_metadata(final_meta, merged),
|
||
)
|
||
|
||
async def _guard_context_window(
|
||
self,
|
||
messages: list[dict[str, Any]],
|
||
stream: StreamBus,
|
||
) -> None:
|
||
try:
|
||
window = resolve_effective_context_window(
|
||
context_window=getattr(self.llm_config, "context_window", None),
|
||
model=str(self.model or ""),
|
||
max_tokens=getattr(self.llm_config, "max_tokens", None),
|
||
)
|
||
except Exception:
|
||
return
|
||
if not window or window <= 0:
|
||
return
|
||
budget = int(window * CONTEXT_WINDOW_GUARD_RATIO)
|
||
if self._estimate_messages_tokens(messages) <= budget:
|
||
return
|
||
snipped = False
|
||
for msg in messages:
|
||
if msg.get("role") != "tool":
|
||
continue
|
||
marker = self._tool_result_snip_marker()
|
||
if msg.get("content") == marker:
|
||
continue
|
||
msg["content"] = marker
|
||
snipped = True
|
||
if self._estimate_messages_tokens(messages) <= budget:
|
||
break
|
||
if snipped:
|
||
await stream.progress(
|
||
self._t("notices.context_window_guard"),
|
||
source="chat",
|
||
stage="responding",
|
||
metadata={"trace_kind": "warning"},
|
||
)
|
||
|
||
@staticmethod
|
||
def _estimate_messages_tokens(messages: list[dict[str, Any]]) -> int:
|
||
from deeptutor.services.session.context_builder import count_tokens
|
||
|
||
total = 0
|
||
for msg in messages:
|
||
content = msg.get("content")
|
||
if isinstance(content, str):
|
||
total += count_tokens(content)
|
||
elif isinstance(content, list):
|
||
for part in content:
|
||
if isinstance(part, dict) and part.get("type") == "text":
|
||
total += count_tokens(str(part.get("text") or ""))
|
||
return total
|
||
|
||
# ---- LLM client ------------------------------------------------------
|
||
|
||
def _build_openai_client(self):
|
||
return build_openai_client(self._client_config)
|
||
|
||
def _completion_kwargs(self, max_tokens: int) -> dict[str, Any]:
|
||
return build_completion_kwargs(
|
||
temperature=self._chat_temperature,
|
||
model=self.model,
|
||
max_tokens=max_tokens,
|
||
binding=self.binding,
|
||
reasoning_effort=self.reasoning_effort,
|
||
)
|
||
|
||
def _can_use_native_tool_calling(self) -> bool:
|
||
return can_use_native_tool_calling(binding=self.binding, model=self.model)
|
||
|
||
# ---- small helpers ---------------------------------------------------
|
||
|
||
@staticmethod
|
||
def _current_user_id() -> str:
|
||
try:
|
||
from deeptutor.multi_user.context import get_current_user
|
||
|
||
return str(get_current_user().id or "anonymous")
|
||
except Exception:
|
||
return "anonymous"
|
||
|
||
@staticmethod
|
||
def _selected_kbs(context: UnifiedContext) -> list[str]:
|
||
return [str(kb).strip() for kb in context.knowledge_bases if str(kb).strip()]
|
||
|
||
@staticmethod
|
||
def _workspace_key(context: UnifiedContext) -> str:
|
||
raw = str(
|
||
context.metadata.get("turn_id")
|
||
or context.session_id
|
||
or context.metadata.get("message_id")
|
||
or "direct"
|
||
)
|
||
cleaned = "".join(ch if ch.isalnum() or ch in {"-", "_"} else "_" for ch in raw)
|
||
return cleaned.strip("_") or "direct"
|
||
|
||
def _kb_system_note(self, context: UnifiedContext) -> str:
|
||
if self._exclusive_capability_active(context):
|
||
return ""
|
||
kbs = self._selected_kbs(context)
|
||
if not kbs:
|
||
return ""
|
||
joined = ", ".join(kbs)
|
||
if self.language == "zh":
|
||
return f"用户已挂载知识库：{joined}。调用 rag 时，kb_name 必须从其中选一个。"
|
||
return f"Attached knowledge bases: {joined}. When calling rag, kb_name must be one of these names."
|
||
|
||
def _workspace_system_note(self, context: UnifiedContext) -> str:
|
||
if not getattr(self, "_exec_enabled", False):
|
||
return ""
|
||
try:
|
||
from deeptutor.services.path_service import get_path_service
|
||
|
||
exec_dir = (
|
||
get_path_service().get_task_workspace(
|
||
"chat",
|
||
self._workspace_key(context),
|
||
)
|
||
/ "exec"
|
||
)
|
||
except Exception:
|
||
return ""
|
||
if self.language == "zh":
|
||
return (
|
||
"[本轮工作区]\n"
|
||
f"脚本和临时文件应写入：{exec_dir}\n"
|
||
"相对路径会解析到这个目录。需要创建 PDF、图片、表格或其他下载文件时，"
|
||
"直接通过 exec 写入并运行脚本（如 heredoc：python - <<'PY' … PY，"
|
||
"或 cat > gen.py <<'EOF' … EOF 后再运行）。生成的文件会自动以可下载"
|
||
"卡片呈现给用户——在回答里描述你做了什么即可，不要粘贴原始 URL。"
|
||
)
|
||
return (
|
||
"[Turn workspace]\n"
|
||
f"Scripts and temporary files should be written under: {exec_dir}\n"
|
||
"Relative paths resolve to this directory. When creating PDFs, images, "
|
||
"spreadsheets, or other downloadable files, write and run scripts directly "
|
||
"through exec (e.g. a heredoc: python - <<'PY' … PY, or cat > gen.py <<'EOF' "
|
||
"… EOF then run it). Generated files are shown to the user automatically as "
|
||
"downloadable cards — describe what you made, do not paste raw URLs."
|
||
)
|
||
|
||
def _t(self, key: str, default: str = "", **kwargs: Any) -> str:
|
||
value: Any = self._prompts
|
||
for part in key.split("."):
|
||
if not isinstance(value, dict) or part not in value:
|
||
value = default
|
||
break
|
||
value = value[part]
|
||
if not isinstance(value, str):
|
||
value = default
|
||
if kwargs:
|
||
try:
|
||
return value.format(**kwargs)
|
||
except (KeyError, IndexError, ValueError):
|
||
return value
|
||
return value
|
||
|
||
|
||
__all__ = [
|
||
"AgenticChatPipeline",
|
||
"CHAT_OPTIONAL_TOOLS",
|
||
"KB_SEED_CHARS_PER_KB",
|
||
"KB_SEED_MAX_KBS",
|
||
"_DispatchOutcome",
|
||
"_read_int",
|
||
]
|