Losita/mai-bot
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat；私聊回滚wait，添加记忆总结模型配置

Browse Source
This commit is contained in:
SengokuCola
2026-05-08 02:21:27 +08:00
parent 197351d469
commit 7bdbdec157
13 changed files with 371 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
prompts/zh-CN/maisaka_timing_gate.prompt
View File

@@ -8,10 +8,12 @@
在当前场景中，不同的人正在互动（{bot_name} 也是一位参与的用户），用户也可能正在连续发送消息或彼此互动。
你的任务不是生成对别人可见的发言，也不是直接使用查询类工具，而是判断当前是否应该：
- continue：立刻进入下一轮完整思考、搜集信息、回复与其他工具执行
{timing_gate_wait_rule}
- no_reply：本轮不继续发言，等待新的消息；也用于用户可能还没说完、需要先把发言权交还给用户的场景
节奏控制规则：
1. 如果 {bot_name} 已经回复，但用户暂时没有新的回复，且没有新信息需要搜集，使用 no_reply 进行等待。
1. 如果 {bot_name} 已经回复，但用户暂时没有新的回复，且没有新信息需要搜集，进行等待。
2. 如果用户有新发言，但是你评估用户还有后续发言尚未发送，可以适当等待让用户说完。
3. 你需要先评估是用户之间在互动还是和{bot_name}在互动，不要盲目插话，弄错回复对象
4. 你需要评估哪些话是对{bot_name}的发言，哪些是用户之间的交流或者自言自语，根据情况适当发言。

76
pytests/A_memorix_test/test_summary_importer_model_config.py
View File

@@ -12,12 +12,24 @@ from src.services import llm_service as llm_api
def _fake_available_models() -> dict[str, TaskConfig]:
return {
"memory": TaskConfig(
model_list=["memory-model"],
max_tokens=512,
temperature=0.4,
selection_strategy="random",
),
"utils": TaskConfig(
model_list=["utils-model"],
max_tokens=256,
temperature=0.5,
selection_strategy="random",
),
"replyer": TaskConfig(
model_list=["test-model"],
model_list=["replyer-model"],
max_tokens=128,
temperature=0.7,
selection_strategy="priority",
)
selection_strategy="random",
),
}
@@ -35,7 +47,63 @@ def test_resolve_summary_model_config_uses_auto_list_when_summarization_missing(
resolved = importer._resolve_summary_model_config()
assert resolved is not None
assert resolved.model_list == ["test-model"]
assert resolved.model_list == ["memory-model"]
def test_resolve_summary_model_config_auto_falls_back_to_utils_then_planner(monkeypatch):
importer = SummaryImporter(
vector_store=None,
graph_store=None,
metadata_store=None,
embedding_manager=None,
plugin_config={},
)
monkeypatch.setattr(
llm_api,
"get_available_models",
lambda: {
"utils": TaskConfig(model_list=["utils-model"]),
"planner": TaskConfig(model_list=["planner-model"]),
"replyer": TaskConfig(model_list=["replyer-model"]),
},
)
resolved = importer._resolve_summary_model_config()
assert resolved is not None
assert resolved.model_list == ["utils-model"]
monkeypatch.setattr(
llm_api,
"get_available_models",
lambda: {
"planner": TaskConfig(model_list=["planner-model"]),
"replyer": TaskConfig(model_list=["replyer-model"]),
},
)
resolved = importer._resolve_summary_model_config()
assert resolved is not None
assert resolved.model_list == ["planner-model"]
def test_resolve_summary_model_config_auto_does_not_fallback_to_replyer(monkeypatch):
monkeypatch.setattr(
llm_api,
"get_available_models",
lambda: {
"replyer": TaskConfig(model_list=["replyer-model"]),
"embedding": TaskConfig(model_list=["embedding-model"]),
},
)
importer = SummaryImporter(
vector_store=None,
graph_store=None,
metadata_store=None,
embedding_manager=None,
plugin_config={},
)
assert importer._resolve_summary_model_config() is None
def test_resolve_summary_model_config_rejects_legacy_string_selector(monkeypatch):

81
pytests/test_maisaka_timing_gate.py
View File

@@ -4,7 +4,7 @@ from types import SimpleNamespace
import asyncio
import pytest
from src.core.tooling import ToolExecutionResult, ToolInvocation
from src.core.tooling import ToolAvailabilityContext, ToolExecutionResult, ToolInvocation
from src.llm_models.payload_content.tool_option import ToolCall
from src.maisaka.builtin_tool import get_timing_tools
from src.maisaka.chat_loop_service import ChatResponse, MaisakaChatLoopService
@@ -33,20 +33,42 @@ def _build_chat_response(tool_calls: list[ToolCall]) -> ChatResponse:
)
def test_timing_gate_tools_only_expose_continue_and_no_reply() -> None:
tool_names = {tool_definition["name"] for tool_definition in get_timing_tools()}
def _build_runtime_stub(*, is_group_chat: bool) -> SimpleNamespace:
return SimpleNamespace(
_force_next_timing_continue=False,
_chat_history=[],
session_id="test-session",
chat_stream=SimpleNamespace(
session_id="test-session",
stream_id="test-stream",
is_group_session=is_group_chat,
group_id="group-1" if is_group_chat else "",
user_id="user-1",
platform="qq",
),
_chat_loop_service=SimpleNamespace(build_prompt_template_context=lambda: {}),
log_prefix="[test]",
stopped=False,
)
assert tool_names == {"continue", "no_reply"}
def test_timing_gate_tools_expose_wait_only_in_private_chat() -> None:
private_tool_names = {
tool_definition["name"]
for tool_definition in get_timing_tools(ToolAvailabilityContext(is_group_chat=False))
}
group_tool_names = {
tool_definition["name"]
for tool_definition in get_timing_tools(ToolAvailabilityContext(is_group_chat=True))
}
assert private_tool_names == {"continue", "no_reply", "wait"}
assert group_tool_names == {"continue", "no_reply"}
@pytest.mark.asyncio
async def test_timing_gate_invalid_tool_defaults_to_no_reply(monkeypatch: pytest.MonkeyPatch) -> None:
runtime = SimpleNamespace(
_force_next_timing_continue=False,
_chat_history=[],
log_prefix="[test]",
stopped=False,
)
runtime = _build_runtime_stub(is_group_chat=True)
def _enter_stop_state() -> None:
runtime.stopped = True
@@ -90,12 +112,7 @@ async def test_timing_gate_invalid_tool_defaults_to_no_reply(monkeypatch: pytest
@pytest.mark.asyncio
async def test_timing_gate_invalid_tool_retries_until_valid(monkeypatch: pytest.MonkeyPatch) -> None:
runtime = SimpleNamespace(
_force_next_timing_continue=False,
_chat_history=[],
log_prefix="[test]",
stopped=False,
)
runtime = _build_runtime_stub(is_group_chat=True)
def _enter_stop_state() -> None:
runtime.stopped = True
@@ -148,6 +165,37 @@ async def test_timing_gate_invalid_tool_retries_until_valid(monkeypatch: pytest.
assert tool_monitor_results[0]["tool_name"] == "continue"
@pytest.mark.asyncio
async def test_timing_gate_group_chat_treats_wait_as_invalid(monkeypatch: pytest.MonkeyPatch) -> None:
runtime = _build_runtime_stub(is_group_chat=True)
def _enter_stop_state() -> None:
runtime.stopped = True
runtime._enter_stop_state = _enter_stop_state
engine = MaisakaReasoningEngine(runtime) # type: ignore[arg-type]
async def _fake_timing_gate_sub_agent(**kwargs: object) -> ChatResponse:
tool_definitions = kwargs["tool_definitions"]
assert {tool_definition["name"] for tool_definition in tool_definitions} == {"continue", "no_reply"}
return _build_chat_response([
ToolCall(call_id="disabled-wait", func_name="wait", args={"seconds": 3}),
])
async def _fail_invoke_tool_call(*args: object, **kwargs: object) -> None:
del args, kwargs
raise AssertionError("群聊中禁用的 wait 不应被执行")
monkeypatch.setattr(engine, "_run_timing_gate_sub_agent", _fake_timing_gate_sub_agent)
monkeypatch.setattr(engine, "_invoke_tool_call", _fail_invoke_tool_call)
action, _, tool_results, _ = await engine._run_timing_gate(object()) # type: ignore[arg-type]
assert action == "no_reply"
assert runtime.stopped is True
assert tool_results[-1] == "- no_reply [非法 Timing 工具]: 返回了 wait，已停止本轮并等待新消息"
def test_timing_gate_invalid_tool_hint_keeps_only_latest() -> None:
old_hint = SimpleNamespace(source=TIMING_GATE_INVALID_TOOL_HINT_SOURCE)
runtime = SimpleNamespace(_chat_history=[old_hint])
@@ -184,6 +232,7 @@ def test_timing_gate_invalid_tool_hint_only_visible_to_timing_gate() -> None:
def test_forced_timing_trigger_bypasses_message_frequency_threshold() -> None:
runtime = SimpleNamespace(
_STATE_WAIT="wait",
_agent_state="stop",
_message_turn_scheduled=False,
_internal_turn_queue=asyncio.Queue(),

22
src/A_memorix/core/utils/summary_importer.py
View File

@@ -164,22 +164,14 @@ class SummaryImporter:
def _pick_default_summary_task(self, available_tasks: Dict[str, TaskConfig]) -> Tuple[Optional[str], Optional[TaskConfig]]:
"""
选择总结默认任务，避免错误落到 embedding 任务。
优先级：replyer > utils > planner > tool_use > 其他非 embedding。
优先级：memory > utils > planner；不再顺延到 replyer 或其他任务。
"""
preferred = ("replyer", "utils", "planner", "tool_use")
preferred = ("memory", "utils", "planner")
for name in preferred:
cfg = available_tasks.get(name)
if cfg and cfg.model_list:
return name, cfg
for name, cfg in available_tasks.items():
if name != "embedding" and cfg.model_list:
return name, cfg
for name, cfg in available_tasks.items():
if cfg.model_list:
return name, cfg
return None, None
def _resolve_summary_model_config(self) -> Optional[TaskConfig]:
@@ -187,6 +179,7 @@ class SummaryImporter:
解析 summarization.model_name 为 TaskConfig。
支持：
- "auto"
- "memory"（任务名）
- "replyer"（任务名）
- "some-model-name"（具体模型名）
- ["utils:model1", "utils:model2", "replyer"]（数组混合语法）
@@ -199,7 +192,7 @@ class SummaryImporter:
# 避免默认值本身触发类型校验异常。
raw_cfg = self.plugin_config.get("summarization", {}).get("model_name", ["auto"])
selectors = self._normalize_summary_model_selectors(raw_cfg)
default_task_name, default_task_cfg = self._pick_default_summary_task(available_tasks)
_default_task_name, default_task_cfg = self._pick_default_summary_task(available_tasks)
selected_models: List[str] = []
base_cfg: Optional[TaskConfig] = None
@@ -262,16 +255,11 @@ class SummaryImporter:
_append_models(default_task_cfg.model_list)
if base_cfg is None:
base_cfg = default_task_cfg
else:
first_cfg = next(iter(available_tasks.values()))
_append_models(first_cfg.model_list)
if base_cfg is None:
base_cfg = first_cfg
if not selected_models:
return None
template_cfg = base_cfg or default_task_cfg or next(iter(available_tasks.values()))
template_cfg = base_cfg or default_task_cfg or TaskConfig()
return TaskConfig(
model_list=selected_models,
max_tokens=template_cfg.max_tokens,

2
src/config/config.py
View File

@@ -58,7 +58,7 @@ LEGACY_ENV_PATH: Path = (PROJECT_ROOT / ".env").resolve().absolute()
A_MEMORIX_LEGACY_CONFIG_PATH: Path = (CONFIG_DIR / "a_memorix.toml").resolve().absolute()
MMC_VERSION: str = "1.0.0-pre.14"
CONFIG_VERSION: str = "8.10.15"
MODEL_CONFIG_VERSION: str = "1.16.0"
MODEL_CONFIG_VERSION: str = "1.16.1"
logger = get_logger("config")

7
src/config/default_model_config.py
View File

@@ -25,6 +25,13 @@ DEFAULT_TASK_CONFIG_TEMPLATES: dict[str, dict[str, Any]] = {
"slow_threshold": 15.0,
"selection_strategy": "random",
},
"memory": {
"model_list": [],
"max_tokens": 4096,
"temperature": 0.5,
"slow_threshold": 30.0,
"selection_strategy": "random",
},
"replyer": {
"model_list": ["deepseek-v4-pro-think", "deepseek-v4-pro-nonthink"],
"max_tokens": 4096,

10
src/config/model_configs.py
View File

@@ -440,6 +440,16 @@ class ModelTaskConfig(ConfigBase):
)
"""规划模型配置"""
memory: TaskConfig = Field(
default_factory=TaskConfig,
json_schema_extra={
"x-widget": "custom",
"x-icon": "brain",
"advanced": True,
},
)
"""记忆模型配置，用于长期记忆总结、抽取、写回等高质量记忆任务；留空时由调用方按需回退"""
utils: TaskConfig = Field(
default_factory=TaskConfig,
json_schema_extra={

7
src/maisaka/builtin_tool/__init__.py
View File

@@ -30,6 +30,8 @@ from .tool_search import get_tool_spec as get_tool_search_tool_spec
from .tool_search import handle_tool as handle_tool_search_tool
from .view_complex_message import get_tool_spec as get_view_complex_message_tool_spec
from .view_complex_message import handle_tool as handle_view_complex_message_tool
from .wait import get_tool_spec as get_wait_tool_spec
from .wait import handle_tool as handle_wait_tool
BuiltinToolHandler = Callable[[ToolInvocation, Optional[ToolExecutionContext]], Awaitable[ToolExecutionResult]]
BuiltinToolRawHandler = Callable[
@@ -70,6 +72,7 @@ def _get_query_memory_tool_spec() -> ToolSpec:
BUILTIN_TOOL_ENTRIES: List[BuiltinToolEntry] = [
BuiltinToolEntry("no_reply", get_no_reply_tool_spec, handle_no_reply_tool, stage="timing"),
BuiltinToolEntry("continue", get_continue_tool_spec, handle_continue_tool, stage="timing"),
BuiltinToolEntry("wait", get_wait_tool_spec, handle_wait_tool, stage="timing", chat_scope="private"),
BuiltinToolEntry("finish", get_finish_tool_spec, handle_finish_tool, stage="action"),
BuiltinToolEntry("reply", get_reply_tool_spec, handle_reply_tool, stage="action"),
BuiltinToolEntry(
@@ -145,12 +148,12 @@ def get_all_builtin_tool_specs(context: Optional[ToolAvailabilityContext] = None
return [entry.build_spec() for entry in _get_builtin_tool_entries(context=context)]
def get_timing_tools() -> List[ToolDefinitionInput]:
def get_timing_tools(context: Optional[ToolAvailabilityContext] = None) -> List[ToolDefinitionInput]:
"""获取 Timing Gate 阶段的兼容工具定义。"""
tool_specs = [
entry.build_spec()
for entry in _get_builtin_tool_entries(stage="timing", visibility="visible")
for entry in _get_builtin_tool_entries(stage="timing", visibility="visible", context=context)
]
return [tool_spec.to_llm_definition() for tool_spec in tool_specs if tool_spec.enabled]

51
src/maisaka/builtin_tool/wait.py Normal file
View File

@@ -0,0 +1,51 @@
"""wait 内置工具。"""
from typing import Optional
from src.core.tooling import ToolExecutionContext, ToolExecutionResult, ToolInvocation, ToolSpec
from .context import BuiltinToolRuntimeContext
def get_tool_spec() -> ToolSpec:
"""获取 wait 工具声明。"""
return ToolSpec(
name="wait",
brief_description="暂停当前私聊对话并固定等待一段时间，期间不因新消息提前恢复。",
detailed_description="参数说明：\n- seconds：integer，必填。等待的秒数。等待期间收到的新消息只会暂存，直到超时后再继续处理。",
parameters_schema={
"type": "object",
"properties": {
"seconds": {
"type": "integer",
"description": "等待的秒数。",
},
},
"required": ["seconds"],
},
provider_name="maisaka_builtin",
provider_type="builtin",
)
async def handle_tool(
tool_ctx: BuiltinToolRuntimeContext,
invocation: ToolInvocation,
context: Optional[ToolExecutionContext] = None,
) -> ToolExecutionResult:
"""执行 wait 内置工具。"""
del context
seconds = invocation.arguments.get("seconds", 30)
try:
wait_seconds = int(seconds)
except (TypeError, ValueError):
wait_seconds = 30
wait_seconds = max(0, wait_seconds)
tool_ctx.runtime._enter_wait_state(seconds=wait_seconds, tool_call_id=invocation.call_id)
return tool_ctx.build_success_result(
invocation.tool_name,
f"当前私聊对话循环进入等待状态，将固定等待 {wait_seconds} 秒；期间收到的新消息不会提前打断本次等待。",
metadata={"pause_execution": True},
)

12
src/maisaka/chat_loop_service.py
View File

@@ -40,7 +40,7 @@ from .history_utils import drop_orphan_tool_results, normalize_tool_result_order
from .display.prompt_cli_renderer import PromptCLIVisualizer
from .visual_mode_utils import resolve_enable_visual_planner
TIMING_GATE_TOOL_NAMES = {"continue", "no_reply"}
TIMING_GATE_TOOL_NAMES = {"continue", "no_reply", "wait"}
REQUEST_TYPE_BY_REQUEST_KIND = {
"planner": "maisaka_planner",
"timing_gate": "maisaka_timing_gate",
@@ -362,6 +362,7 @@ class MaisakaChatLoopService:
"file_tools_section": tools_section,
"group_chat_attention_block": self._build_group_chat_attention_block(),
"identity": self._personality_prompt,
"timing_gate_wait_rule": self._build_timing_gate_wait_rule(),
"time_block": self._build_time_block(),
}
@@ -398,6 +399,15 @@ class MaisakaChatLoopService:
return "在该聊天中的注意事项：\n" + "\n\n".join(prompt_lines) + "\n"
def _build_timing_gate_wait_rule(self) -> str:
"""构造 Timing Gate 中 wait 工具的场景说明。"""
if self._is_group_chat is True:
return ""
if self._is_group_chat is False:
return "- wait：固定再等待一段时间，时间到后再重新判断"
return "- wait：仅私聊可用；如果当前是群聊，不要调用"
@staticmethod
def _get_chat_prompt_for_chat(chat_id: str, is_group_chat: Optional[bool]) -> str:
"""根据聊天流 ID 获取匹配的额外提示。"""

130
src/maisaka/reasoning_engine.py
View File

@@ -55,7 +55,7 @@ logger = get_logger("maisaka_reasoning_engine")
TIMING_GATE_CONTEXT_DROP_HEAD_RATIO = 0.7
TIMING_GATE_MAX_ATTEMPTS = 3
TIMING_GATE_TOOL_NAMES = {"continue", "no_reply"}
TIMING_GATE_TOOL_NAMES = {"continue", "no_reply", "wait"}
HISTORY_SILENT_TOOL_NAMES = {"finish"}
@@ -143,30 +143,13 @@ class MaisakaReasoningEngine:
tool_definitions=tool_definitions,
)
@staticmethod
def _build_timing_gate_fallback_prompt() -> str:
"""构造 Timing Gate 子代理的兜底提示词。"""
return (
"你是 Maisaka 的 timing gate 子代理，只负责决定当前会话下一步的节奏控制。\n"
"你必须且只能调用一个工具，不要输出普通文本答案。\n"
"可用工具只有两个：\n"
"1. no_reply: 适合当前不继续本轮发言，等待新的外部消息或让用户继续说完。\n"
"2. continue: 适合现在立刻进入下一轮正常思考、回复、查询和其他工具执行。\n"
"如果需要真正回复消息、查询信息或使用其他工具，应该调用 continue，让主分支继续执行，而不是在这里完成。\n"
"不要连续调用多个工具，也不要输出工具之外的计划。"
)
def _build_timing_gate_system_prompt(self) -> str:
"""构造 Timing Gate 子代理使用的系统提示词。"""
try:
return load_prompt(
"maisaka_timing_gate",
**self._runtime._chat_loop_service.build_prompt_template_context(),
)
except Exception:
return self._build_timing_gate_fallback_prompt()
return load_prompt(
"maisaka_timing_gate",
**self._runtime._chat_loop_service.build_prompt_template_context(),
)
async def _build_action_tool_definitions(self) -> tuple[list[dict[str, Any]], str]:
"""构造 Action Loop 阶段可见的工具定义与 deferred tools 提示。"""
@@ -242,7 +225,7 @@ class MaisakaReasoningEngine:
async def _run_timing_gate(
self,
anchor_message: SessionMessage,
) -> tuple[Literal["continue", "no_reply"], Any, list[str], list[dict[str, Any]]]:
) -> tuple[Literal["continue", "no_reply", "wait"], Any, list[str], list[dict[str, Any]]]:
"""运行 Timing Gate 子代理并返回控制决策。"""
if self._runtime._force_next_timing_continue:
@@ -254,13 +237,19 @@ class MaisakaReasoningEngine:
selected_tool_call: Optional[ToolCall] = None
invalid_tool_text = ""
for attempt_index in range(TIMING_GATE_MAX_ATTEMPTS):
timing_tool_definitions = get_timing_tools(self._build_tool_availability_context())
available_timing_tool_names = {
str(tool_definition.get("name") or "").strip()
for tool_definition in timing_tool_definitions
if str(tool_definition.get("name") or "").strip()
}
response = await self._run_timing_gate_sub_agent(
system_prompt=self._build_timing_gate_system_prompt(),
tool_definitions=get_timing_tools(),
tool_definitions=timing_tool_definitions,
)
selected_tool_call = None
for tool_call in response.tool_calls:
if tool_call.func_name in TIMING_GATE_TOOL_NAMES:
if tool_call.func_name in available_timing_tool_names:
selected_tool_call = tool_call
break
@@ -332,7 +321,12 @@ class MaisakaReasoningEngine:
self._append_timing_gate_execution_result(response, selected_tool_call, result)
timing_action = str(result.metadata.get("timing_action") or selected_tool_call.func_name).strip()
if timing_action not in TIMING_GATE_TOOL_NAMES:
available_timing_action_names = {
str(tool_definition.get("name") or "").strip()
for tool_definition in get_timing_tools(self._build_tool_availability_context())
if str(tool_definition.get("name") or "").strip()
}
if timing_action not in available_timing_action_names:
logger.warning(
f"{self._runtime.log_prefix} Timing Gate 返回未知动作 {timing_action!r}，将按 no_reply 处理"
)
@@ -388,7 +382,7 @@ class MaisakaReasoningEngine:
hint_content = (
"Timing Gate 上一轮选择了非法工具："
f"{normalized_tool_text}。\n"
"Timing Gate 只能调用 continue 或 no_reply 中的一个工具。"
"Timing Gate 只能调用当前可用的 continue、no_reply 或 wait 中的一个工具。"
)
self._runtime._chat_history.append(
SessionBackedMessage(
@@ -419,7 +413,12 @@ class MaisakaReasoningEngine:
try:
while self._runtime._running:
queued_trigger = await self._runtime._internal_turn_queue.get()
message_triggered = self._drain_ready_turn_triggers(queued_trigger)
message_triggered, timeout_triggered = self._drain_ready_turn_triggers(queued_trigger)
if self._runtime._agent_state == self._runtime._STATE_WAIT and not timeout_triggered:
self._runtime._message_turn_scheduled = False
logger.debug(f"{self._runtime.log_prefix} 当前仍处于 wait 状态，忽略消息触发并继续等待超时")
continue
if message_triggered:
await self._runtime._wait_for_message_quiet_period()
@@ -430,17 +429,30 @@ class MaisakaReasoningEngine:
if self._runtime._has_pending_messages()
else []
)
if not cached_messages:
continue
if cached_messages:
self._runtime._agent_state = self._runtime._STATE_RUNNING
self._runtime._update_stage_status(
"消息整理",
f"待处理消息 {len(cached_messages)} 条",
)
asyncio.create_task(self._runtime._trigger_batch_learning(cached_messages))
if timeout_triggered:
self._runtime._chat_history.append(
self._build_wait_completed_message(has_new_messages=True)
)
await self._ingest_messages(cached_messages)
anchor_message = cached_messages[-1]
else:
anchor_message = self._get_timeout_anchor_message()
if anchor_message is None:
logger.warning(f"{self._runtime.log_prefix} wait 超时后没有可复用的锚点消息，跳过本轮")
continue
logger.info(f"{self._runtime.log_prefix} 等待超时后开始新一轮思考")
if self._runtime._pending_wait_tool_call_id:
self._runtime._chat_history.append(
self._build_wait_completed_message(has_new_messages=False)
)
self._runtime._agent_state = self._runtime._STATE_RUNNING
self._runtime._update_stage_status(
"消息整理",
f"待处理消息 {len(cached_messages)} 条",
)
asyncio.create_task(self._runtime._trigger_batch_learning(cached_messages))
await self._ingest_messages(cached_messages)
anchor_message = cached_messages[-1]
try:
timing_gate_required = True
round_index = 0
@@ -492,7 +504,7 @@ class MaisakaReasoningEngine:
timing_tool_monitor_results,
) = await self._run_timing_gate(anchor_message)
timing_elapsed_seconds = time.time() - timing_started_at
if timing_action != "continue":
if timing_action == "no_reply":
await self._runtime._wait_for_timing_gate_non_continue_cooldown(
timing_elapsed_seconds
)
@@ -511,8 +523,12 @@ class MaisakaReasoningEngine:
)
timing_gate_required = self._mark_timing_gate_completed(timing_action)
if timing_action != "continue":
cycle_end_reason = "timing_no_reply"
cycle_end_detail = "Timing Gate 选择 no_reply，本轮不会进入 Planner。"
if timing_action == "wait":
cycle_end_reason = "timing_wait"
cycle_end_detail = "Timing Gate 选择 wait，本轮不会进入 Planner，将在等待结束后继续。"
else:
cycle_end_reason = "timing_no_reply"
cycle_end_detail = "Timing Gate 选择 no_reply，本轮不会进入 Planner。"
logger.debug(
f"{self._runtime.log_prefix} Timing Gate 结束当前回合: "
f"回合={round_index + 1} 动作={timing_action}"
@@ -757,11 +773,12 @@ class MaisakaReasoningEngine:
def _drain_ready_turn_triggers(
self,
queued_trigger: Literal["message"],
) -> bool:
queued_trigger: Literal["message", "timeout"],
) -> tuple[bool, bool]:
"""合并当前已就绪的消息触发信号。"""
message_triggered = queued_trigger == "message"
timeout_triggered = queued_trigger == "timeout"
while True:
try:
@@ -772,8 +789,33 @@ class MaisakaReasoningEngine:
if next_trigger == "message":
message_triggered = True
continue
if next_trigger == "timeout":
timeout_triggered = True
continue
return message_triggered
return message_triggered, timeout_triggered
def _get_timeout_anchor_message(self) -> Optional[SessionMessage]:
"""在 wait 超时后复用最近一条真实用户消息作为锚点。"""
if self._runtime.message_cache:
return self._runtime.message_cache[-1]
return None
def _build_wait_completed_message(self, *, has_new_messages: bool) -> ToolResultMessage:
"""构造 wait 完成后的工具结果消息。"""
tool_call_id = self._runtime._pending_wait_tool_call_id or "wait_timeout"
self._runtime._pending_wait_tool_call_id = None
content = (
"等待已结束，期间收到了新的用户输入。请结合这些新消息继续下一轮思考。"
if has_new_messages
else "等待已超时，期间没有收到新的用户输入。请基于现有上下文继续下一轮思考。"
)
return ToolResultMessage(
content=content,
timestamp=datetime.now(),
tool_call_id=tool_call_id,
tool_name="wait",
)
async def _ingest_messages(self, messages: list[SessionMessage]) -> None:
"""处理传入消息列表，将其转换为历史消息并加入聊天历史缓存。"""

53
src/maisaka/runtime.py
View File

@@ -62,6 +62,7 @@ class MaisakaHeartFlowChatting:
"""会话级别的 Maisaka 运行时。"""
_STATE_RUNNING: Literal["running"] = "running"
_STATE_WAIT: Literal["wait"] = "wait"
_STATE_STOP: Literal["stop"] = "stop"
def __init__(self, session_id: str):
@@ -84,7 +85,7 @@ class MaisakaHeartFlowChatting:
# Keep all original messages for batching and later learning.
self.message_cache: list[SessionMessage] = []
self._last_processed_index = 0
self._internal_turn_queue: asyncio.Queue[Literal["message"]] = asyncio.Queue()
self._internal_turn_queue: asyncio.Queue[Literal["message", "timeout"]] = asyncio.Queue()
self._mcp_manager: Optional[MCPManager] = None
self._mcp_host_bridge: Optional[MCPHostLLMBridge] = None
@@ -102,6 +103,7 @@ class MaisakaHeartFlowChatting:
self._talk_frequency_adjust = 1.0
self._reply_latency_measurement_started_at: Optional[float] = None
self._recent_reply_latencies: deque[tuple[float, float]] = deque()
self._wait_timeout_task: Optional[asyncio.Task[None]] = None
self._max_internal_rounds = MAX_INTERNAL_ROUNDS
configured_context_size = (
global_config.chat.max_context_size
@@ -109,7 +111,8 @@ class MaisakaHeartFlowChatting:
else global_config.chat.max_private_context_size
)
self._max_context_size = max(1, int(configured_context_size))
self._agent_state: Literal["running", "stop"] = self._STATE_STOP
self._agent_state: Literal["running", "wait", "stop"] = self._STATE_STOP
self._pending_wait_tool_call_id: Optional[str] = None
self._force_next_timing_continue = False
self._force_next_timing_message_id = ""
self._force_next_timing_reason = ""
@@ -208,6 +211,7 @@ class MaisakaHeartFlowChatting:
self._message_turn_scheduled = False
self._message_debounce_required = False
self._cancel_deferred_message_turn_task()
self._cancel_wait_timeout_task()
while not self._internal_turn_queue.empty():
_ = self._internal_turn_queue.get_nowait()
@@ -938,6 +942,9 @@ class MaisakaHeartFlowChatting:
def _schedule_message_turn(self) -> None:
"""为当前待处理消息安排一次内部 turn。"""
if self._agent_state == self._STATE_WAIT:
return
if not self._has_pending_messages() or self._message_turn_scheduled:
return
@@ -1023,6 +1030,48 @@ class MaisakaHeartFlowChatting:
def _enter_stop_state(self) -> None:
"""切换到停止状态。"""
self._agent_state = self._STATE_STOP
self._pending_wait_tool_call_id = None
self._cancel_wait_timeout_task()
def _enter_wait_state(self, seconds: Optional[float] = None, tool_call_id: Optional[str] = None) -> None:
"""切换到等待状态。"""
self._agent_state = self._STATE_WAIT
self._pending_wait_tool_call_id = tool_call_id
self._message_turn_scheduled = False
self._cancel_deferred_message_turn_task()
self._cancel_wait_timeout_task()
if seconds is not None:
self._wait_timeout_task = asyncio.create_task(
self._schedule_wait_timeout(seconds=seconds, tool_call_id=tool_call_id)
)
def _cancel_wait_timeout_task(self) -> None:
"""取消当前 wait 对应的超时任务。"""
if self._wait_timeout_task is None:
return
self._wait_timeout_task.cancel()
self._wait_timeout_task = None
async def _schedule_wait_timeout(self, seconds: float, tool_call_id: Optional[str]) -> None:
"""在 wait 到期后向内部循环投递 timeout 触发。"""
try:
if seconds > 0:
await asyncio.sleep(seconds)
if not self._running:
return
if self._agent_state != self._STATE_WAIT:
return
if self._pending_wait_tool_call_id != tool_call_id:
return
logger.debug(f"{self.log_prefix} Maisaka 等待已超时")
self._agent_state = self._STATE_RUNNING
await self._internal_turn_queue.put("timeout")
except asyncio.CancelledError:
return
finally:
if self._wait_timeout_task is not None and self._pending_wait_tool_call_id == tool_call_id:
self._wait_timeout_task = None
async def _trigger_batch_learning(self, messages: list[SessionMessage]) -> None:
"""按同一批消息触发表达方式和黑话学习。"""

8
uv.lock generated
View File

@@ -1511,7 +1511,7 @@ requires-dist = [
{ name = "httpx", extras = ["socks"] },
{ name = "jieba", specifier = ">=0.42.1" },
{ name = "json-repair", specifier = ">=0.47.6" },
{ name = "maibot-dashboard", specifier = ">=1.0.7" },
{ name = "maibot-dashboard", specifier = ">=1.0.8" },
{ name = "maibot-plugin-sdk", specifier = ">=2.4.0" },
{ name = "maim-message", specifier = ">=0.6.2" },
{ name = "matplotlib", specifier = ">=3.10.5" },
@@ -1549,11 +1549,11 @@ dev = [
[[package]]
name = "maibot-dashboard"
version = "1.0.7"
version = "1.0.8"
source = { registry = "https://pypi.tuna.tsinghua.edu.cn/simple" }
sdist = { url = "https://pypi.tuna.tsinghua.edu.cn/packages/20/48/3477af08da9d7fb422a7ecd7800ebbc2200b7dd09685c167a11aed07e773/maibot_dashboard-1.0.7.tar.gz", hash = "sha256:29a0e2f121d05f6b87cd79059c59e77a533ee9cbdaa56888043443d1d4382785", size = 2488200, upload-time = "2026-05-07T05:09:22.843Z" }
sdist = { url = "https://pypi.tuna.tsinghua.edu.cn/packages/13/9f/e59b1a6299cc4f8c9ac16c7c2774581220fdd27227ac9c2fdfb947dfc2f5/maibot_dashboard-1.0.8.tar.gz", hash = "sha256:a47309072d8154905738d02ccad17a543d5159a1e62ca87076ac4dce39e6c922", size = 2496374, upload-time = "2026-05-07T13:58:39.386Z" }
wheels = [
{ url = "https://pypi.tuna.tsinghua.edu.cn/packages/02/91/629d6891a9e5fc12083efe2de712342a1d20fd0ce22cf1685a45d1d2fc91/maibot_dashboard-1.0.7-py3-none-any.whl", hash = "sha256:38e7e4baa921b5b5b75404aa1b9c0dfce264c6ee4783b0e54174a32f23eb4880", size = 2555199, upload-time = "2026-05-07T05:09:21.262Z" },
{ url = "https://pypi.tuna.tsinghua.edu.cn/packages/0f/60/fde671bf332133f1403673096eefcd49f36133141a6b9229e72c2588b221/maibot_dashboard-1.0.8-py3-none-any.whl", hash = "sha256:39da973fed56f1491245109615d81ea79add859467798af92d4ace7d8a5d7557", size = 2563243, upload-time = "2026-05-07T13:58:37.868Z" },
]
[[package]]