feat: 添加旧版会话管理器的兼容实现,支持会话数据的存储、获取和删除功能

feat: 在内存客户端中添加精确获取记忆项的功能
feat: 在事件消息中添加字段约束验证,确保各阶段的字段符合要求
feat: 在 SupervisorRuntime 中注册 handler 时处理冲突并输出警告
This commit is contained in:
whatevertogo
2026-03-13 01:45:17 +08:00
parent 0159b008c3
commit 280e32bfae
8 changed files with 475 additions and 19 deletions

View File

@@ -4,6 +4,7 @@
- 2026-03-12: Legacy `handshake` payloads only contain `event_type` / `handler_full_name` metadata and do not preserve v4 command/message trigger details such as command names, aliases, keywords, or regex. Any legacy-to-v4 handshake translation must approximate handlers as coarse event subscriptions and keep the raw handshake payload in metadata for lossless fallback.
- 2026-03-12: `src/astrbot_sdk/tests/start_client.py` and `benchmark_8_plugins_resource_usage.py` still reference legacy `astrbot_sdk.runtime.galaxy`, but `src-new/astrbot_sdk/runtime/galaxy.py` no longer exists. Treat `tests_v4/test_script_migrations.py` as the maintained replacement instead of reviving the old Galaxy path.
- 2026-03-12: Legacy `src/astrbot_sdk/api/event/filter.py` exported a much larger decorator surface than `src-new/astrbot_sdk/api/event/filter.py`. Current compat coverage is enough for `command` / `regex` / `permission` and the exercised migration tests, but it is not a full drop-in replacement for every historical filter helper.
- 2026-03-13: Transport-pair startup tests for `SupervisorRuntime` must start a real peer on the opposite transport and provide an `initialize` response. Wiring only the supervisor side drops or captures the outgoing initialize message without replying, and `Peer.initialize()` then waits forever.
# 开发命令
@@ -19,10 +20,12 @@ ruff check . --fix # 使用 ruff 检查并自动修复问题
## 测试
如果修改了内容可能影响现有功能,请运行测试以确保没有引入错误:
如果修改了bug或者更改了功能需要添加新的测试
```bash
python run_tests.py # 运行所有测试
python run_tests.py -v # 详细输出
python run_tests.py -k "test_peer" # 运行匹配模式的测试
python run_tests.py --cov # 运行测试并生成覆盖率报告
```

View File

@@ -4,6 +4,7 @@
- 2026-03-12: Legacy `handshake` payloads only contain `event_type` / `handler_full_name` metadata and do not preserve v4 command/message trigger details such as command names, aliases, keywords, or regex. Any legacy-to-v4 handshake translation must approximate handlers as coarse event subscriptions and keep the raw handshake payload in metadata for lossless fallback.
- 2026-03-12: `src/astrbot_sdk/tests/start_client.py` and `benchmark_8_plugins_resource_usage.py` still reference legacy `astrbot_sdk.runtime.galaxy`, but `src-new/astrbot_sdk/runtime/galaxy.py` no longer exists. Treat `tests_v4/test_script_migrations.py` as the maintained replacement instead of reviving the old Galaxy path.
- 2026-03-12: Legacy `src/astrbot_sdk/api/event/filter.py` exported a much larger decorator surface than `src-new/astrbot_sdk/api/event/filter.py`. Current compat coverage is enough for `command` / `regex` / `permission` and the exercised migration tests, but it is not a full drop-in replacement for every historical filter helper.
- 2026-03-13: Transport-pair startup tests for `SupervisorRuntime` must start a real peer on the opposite transport and provide an `initialize` response. Wiring only the supervisor side drops or captures the outgoing initialize message without replying, and `Peer.initialize()` then waits forever.
# 开发命令
@@ -19,10 +20,12 @@ ruff check . --fix # 使用 ruff 检查并自动修复问题
## 测试
如果修改了内容可能影响现有功能,请运行测试以确保没有引入错误:
如果修改了bug或者更改了功能需要添加新的测试
```bash
python run_tests.py # 运行所有测试
python run_tests.py -v # 详细输出
python run_tests.py -k "test_peer" # 运行匹配模式的测试
python run_tests.py --cov # 运行测试并生成覆盖率报告
```

View File

@@ -9,6 +9,7 @@ from .clients.llm import LLMResponse
from .context import Context as NewContext
from .star import Star
# TODO-迁移文档要写,我好烦烦烦你烦烦烦你
MIGRATION_DOC_URL = "https://docs.astrbot.app/migration/v3"
_warned_methods: set[str] = set()
@@ -26,13 +27,33 @@ def _warn_once(old_name: str, replacement: str) -> None:
class LegacyConversationManager:
"""旧版会话管理器的兼容实现。
使用 db 存储会话数据key 为 `__compat_conversations__`。
注意:此实现不提供持久化保证,会话数据仅在当前运行时有效。
"""
def __init__(self, parent: "LegacyContext") -> None:
self._parent = parent
self._counters: defaultdict[str, int] = defaultdict(int)
# 记录每个 unified_msg_origin 的当前会话 ID
self._current_conversations: dict[str, str] = {}
def _ctx(self) -> NewContext:
return self._parent.require_runtime_context()
async def _get_stored(self) -> dict[str, dict[str, Any]]:
"""获取存储的所有会话数据。"""
ctx = self._ctx()
stored = await ctx.db.get("__compat_conversations__")
return stored if isinstance(stored, dict) else {}
async def _set_stored(self, stored: dict[str, dict[str, Any]]) -> None:
"""保存会话数据。"""
ctx = self._ctx()
await ctx.db.set("__compat_conversations__", stored)
async def new_conversation(
self,
unified_msg_origin: str,
@@ -41,10 +62,11 @@ class LegacyConversationManager:
title: str | None = None,
persona_id: str | None = None,
) -> str:
"""创建新会话并返回会话 ID。"""
ctx = self._ctx()
self._counters[unified_msg_origin] += 1
conversation_id = f"{ctx.plugin_id}-conv-{self._counters[unified_msg_origin]}"
stored = await ctx.db.get("__compat_conversations__") or {}
stored = await self._get_stored()
stored[conversation_id] = {
"unified_msg_origin": unified_msg_origin,
"platform_id": platform_id,
@@ -52,9 +74,277 @@ class LegacyConversationManager:
"title": title,
"persona_id": persona_id,
}
await ctx.db.set("__compat_conversations__", stored)
await self._set_stored(stored)
# 设置为当前会话
self._current_conversations[unified_msg_origin] = conversation_id
return conversation_id
async def switch_conversation(
self, unified_msg_origin: str, conversation_id: str
) -> None:
"""切换到指定会话。
Args:
unified_msg_origin: 统一消息来源
conversation_id: 要切换到的会话 ID
"""
stored = await self._get_stored()
if conversation_id not in stored:
return
# 验证会话属于该 unified_msg_origin
conv_data = stored[conversation_id]
if conv_data.get("unified_msg_origin") != unified_msg_origin:
return
self._current_conversations[unified_msg_origin] = conversation_id
async def delete_conversation(
self,
unified_msg_origin: str,
conversation_id: str | None = None,
) -> None:
"""删除指定会话。
当 conversation_id 为 None 时,删除当前会话。
Args:
unified_msg_origin: 统一消息来源
conversation_id: 要删除的会话 ID为 None 时删除当前会话
"""
# 如果 conversation_id 为 None使用当前会话
if conversation_id is None:
conversation_id = self._current_conversations.get(unified_msg_origin)
if conversation_id is None:
return
stored = await self._get_stored()
if conversation_id not in stored:
return
conv_data = stored[conversation_id]
if conv_data.get("unified_msg_origin") != unified_msg_origin:
return
del stored[conversation_id]
await self._set_stored(stored)
# 如果删除的是当前会话,清除当前会话记录
if self._current_conversations.get(unified_msg_origin) == conversation_id:
del self._current_conversations[unified_msg_origin]
async def get_curr_conversation_id(self, unified_msg_origin: str) -> str | None:
"""获取当前会话 ID。
Args:
unified_msg_origin: 统一消息来源
Returns:
当前会话 ID若无则返回 None
"""
return self._current_conversations.get(unified_msg_origin)
async def get_conversation(
self,
unified_msg_origin: str,
conversation_id: str,
create_if_not_exists: bool = False,
) -> dict[str, Any] | None:
"""获取指定会话的数据。
Args:
unified_msg_origin: 统一消息来源
conversation_id: 会话 ID
create_if_not_exists: 如果会话不存在,是否创建新会话
Returns:
会话数据字典,不存在则返回 None
"""
stored = await self._get_stored()
conv = stored.get(conversation_id)
if conv is None and create_if_not_exists:
# 创建新会话
conv = {
"unified_msg_origin": unified_msg_origin,
"platform_id": None,
"content": [],
"title": None,
"persona_id": None,
}
stored[conversation_id] = conv
await self._set_stored(stored)
self._current_conversations[unified_msg_origin] = conversation_id
return conv
async def get_conversations(
self,
unified_msg_origin: str | None = None,
platform_id: str | None = None,
) -> list[dict[str, Any]]:
"""获取会话列表。
Args:
unified_msg_origin: 统一消息来源,可选
platform_id: 平台 ID可选
Returns:
会话列表,每个元素包含 conversation_id 和会话数据
"""
stored = await self._get_stored()
result = []
for conv_id, conv_data in stored.items():
# 按 unified_msg_origin 过滤
if unified_msg_origin is not None:
if conv_data.get("unified_msg_origin") != unified_msg_origin:
continue
# 按 platform_id 过滤
if platform_id is not None:
if conv_data.get("platform_id") != platform_id:
continue
result.append({"conversation_id": conv_id, **conv_data})
return result
async def update_conversation(
self,
unified_msg_origin: str,
conversation_id: str | None = None,
history: list[dict] | None = None,
title: str | None = None,
persona_id: str | None = None,
) -> None:
"""更新会话数据。
Args:
unified_msg_origin: 统一消息来源
conversation_id: 会话 ID为 None 时更新当前会话
history: 对话历史记录
title: 会话标题
persona_id: Persona ID
"""
# 如果 conversation_id 为 None使用当前会话
if conversation_id is None:
conversation_id = self._current_conversations.get(unified_msg_origin)
if conversation_id is None:
return
stored = await self._get_stored()
if conversation_id not in stored:
return
updates: dict[str, Any] = {}
if history is not None:
updates["content"] = history
if title is not None:
updates["title"] = title
if persona_id is not None:
updates["persona_id"] = persona_id
stored[conversation_id].update(updates)
await self._set_stored(stored)
async def delete_conversations_by_user_id(self, unified_msg_origin: str) -> None:
"""删除指定用户的所有会话。
Args:
unified_msg_origin: 统一消息来源
"""
stored = await self._get_stored()
to_delete = [
conv_id
for conv_id, conv_data in stored.items()
if conv_data.get("unified_msg_origin") == unified_msg_origin
]
for conv_id in to_delete:
del stored[conv_id]
await self._set_stored(stored)
# 清除当前会话记录
if unified_msg_origin in self._current_conversations:
del self._current_conversations[unified_msg_origin]
async def add_message_pair(
self,
cid: str,
user_message: str | dict,
assistant_message: str | dict,
) -> None:
"""向会话添加消息对。
Args:
cid: 会话 ID
user_message: 用户消息
assistant_message: 助手消息
"""
stored = await self._get_stored()
if cid not in stored:
return
content = stored[cid].get("content", [])
# 处理消息格式
user_msg = (
user_message
if isinstance(user_message, dict)
else {"role": "user", "content": user_message}
)
assistant_msg = (
assistant_message
if isinstance(assistant_message, dict)
else {"role": "assistant", "content": assistant_message}
)
content.append(user_msg)
content.append(assistant_msg)
stored[cid]["content"] = content
await self._set_stored(stored)
async def update_conversation_title(
self,
unified_msg_origin: str,
title: str,
conversation_id: str | None = None,
) -> None:
"""更新会话标题。
Args:
unified_msg_origin: 统一消息来源
title: 会话标题
conversation_id: 会话 ID为 None 时更新当前会话
Deprecated:
请使用 update_conversation() 的 title 参数。
"""
await self.update_conversation(
unified_msg_origin, conversation_id, title=title
)
async def update_conversation_persona_id(
self,
unified_msg_origin: str,
persona_id: str,
conversation_id: str | None = None,
) -> None:
"""更新会话 Persona ID。
Args:
unified_msg_origin: 统一消息来源
persona_id: Persona ID
conversation_id: 会话 ID为 None 时更新当前会话
Deprecated:
请使用 update_conversation() 的 persona_id 参数。
"""
await self.update_conversation(
unified_msg_origin, conversation_id, persona_id=persona_id
)
async def get_filtered_conversations(self, *args: Any, **kwargs: Any) -> Any:
"""已弃用v4 不支持此方法。"""
raise NotImplementedError(
"get_filtered_conversations() 在 v4 中不再支持。\n"
f"请使用 ctx.db.query(...) 自行实现过滤逻辑。\n"
f"迁移文档:{MIGRATION_DOC_URL}"
)
async def get_human_readable_context(self, *args: Any, **kwargs: Any) -> Any:
"""已弃用v4 不支持此方法。"""
raise NotImplementedError(
"get_human_readable_context() 在 v4 中不再支持。\n"
f"请自行遍历会话 content 字段格式化输出。\n"
f"迁移文档:{MIGRATION_DOC_URL}"
)
class LegacyContext:
def __init__(self, plugin_id: str) -> None:
@@ -129,9 +419,13 @@ class LegacyContext:
text = str(message_chain)
await ctx.platform.send(session, text)
# TODO:迁移文档中说明已废弃 add_llm_tools(),但仍保留接口以避免核心依赖问题。后续版本将移除此接口。
async def add_llm_tools(self, *tools: Any) -> None:
_warn_once("context.add_llm_tools()", "ctx.llm.chat_raw(..., tools=...)")
return None
raise NotImplementedError(
"context.add_llm_tools() 在 v4 中不再支持。\n"
"请使用 ctx.llm.chat_raw(..., tools=[...]) 直接传递工具。\n"
f"迁移文档:{MIGRATION_DOC_URL}"
)
async def put_kv_data(self, key: str, value: dict[str, Any]) -> None:
_warn_once("context.put_kv_data()", "ctx.db.set(key, value)")

View File

@@ -26,5 +26,18 @@ class MemoryClient:
payload.update(extra)
await self._proxy.call("memory.save", {"key": key, "value": payload})
async def get(self, key: str) -> dict[str, Any] | None:
"""精确获取:通过唯一键获取单个记忆项。
Args:
key: 记忆项的唯一键
Returns:
记忆项内容,若不存在则返回 None
"""
output = await self._proxy.call("memory.get", {"key": key})
value = output.get("value")
return value if isinstance(value, dict) else None
async def delete(self, key: str) -> None:
await self._proxy.call("memory.delete", {"key": key})

View File

@@ -13,7 +13,6 @@ from .messages import (
InitializeMessage,
InvokeMessage,
PeerInfo,
ProtocolMessage,
ResultMessage,
)

View File

@@ -3,7 +3,7 @@ from __future__ import annotations
import json
from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, Field
from pydantic import BaseModel, ConfigDict, Field, model_validator
from .descriptors import CapabilityDescriptor, HandlerDescriptor
@@ -65,6 +65,36 @@ class EventMessage(_MessageBase):
output: dict[str, Any] = Field(default_factory=dict)
error: ErrorPayload | None = None
@model_validator(mode="after")
def validate_phase_constraints(self) -> "EventMessage":
"""验证各 phase 的字段约束。
- started: 所有字段必须为空
- delta: 必须有 dataoutput/error 必须为空
- completed: 必须有 outputdata/error 必须为空
- failed: 必须有 errordata/output 必须为空
"""
phase = self.phase
if phase == "started":
if self.data or self.output or self.error:
raise ValueError("started phase 必须所有字段为空")
elif phase == "delta":
if not self.data:
raise ValueError("delta phase 需要 data")
if self.output or self.error:
raise ValueError("delta phase 的 output/error 必须为空")
elif phase == "completed":
if not self.output:
raise ValueError("completed phase 需要 output")
if self.data or self.error:
raise ValueError("completed phase 的 data/error 必须为空")
elif phase == "failed":
if self.error is None:
raise ValueError("failed phase 需要 error")
if self.data or self.output:
raise ValueError("failed phase 的 data/output 必须为空")
return self
class CancelMessage(_MessageBase):
type: Literal["cancel"] = "cancel"

View File

@@ -220,6 +220,7 @@ class SupervisorRuntime:
self.peer.set_cancel_handler(self._handle_upstream_cancel)
self.worker_sessions: dict[str, WorkerSession] = {}
self.handler_to_worker: dict[str, WorkerSession] = {}
self._handler_sources: dict[str, str] = {} # handler_id -> plugin_name
self.active_requests: dict[str, WorkerSession] = {}
self.loaded_plugins: list[str] = []
self.skipped_plugins: dict[str, str] = {}
@@ -249,6 +250,28 @@ class SupervisorRuntime:
exposed=False,
)
def _register_handler(
self, handler, session: WorkerSession, plugin_name: str
) -> None:
"""注册 handler处理冲突时输出警告。
Args:
handler: Handler 描述符
session: Worker 会话
plugin_name: 插件名称
"""
handler_id = handler.id
existing_plugin = self._handler_sources.get(handler_id)
if existing_plugin is not None:
logger.warning(
f"Handler ID 冲突:'{handler_id}' 已被插件 '{existing_plugin}' 注册,"
f"现在被插件 '{plugin_name}' 覆盖。"
)
self.handler_to_worker[handler_id] = session
self._handler_sources[handler_id] = plugin_name
async def start(self) -> None:
discovery = discover_plugins(self.plugins_dir)
self.skipped_plugins = dict(discovery.skipped_plugins)
@@ -270,7 +293,7 @@ class SupervisorRuntime:
self.worker_sessions[plugin.name] = session
self.loaded_plugins.append(plugin.name)
for handler in session.handlers:
self.handler_to_worker[handler.id] = session
self._register_handler(handler, session, plugin.name)
aggregated_handlers = list(self.handler_to_worker.keys())
logger.info(
@@ -299,9 +322,12 @@ class SupervisorRuntime:
session = self.worker_sessions.pop(plugin_name, None)
if session is None:
return
# 从 handler_to_worker 中移除该 worker 的所有 handlers
# 从 handler_to_worker 中移除该插件注册的 handlers仅当来源仍为此插件时
for handler in session.handlers:
self.handler_to_worker.pop(handler.id, None)
source_plugin = self._handler_sources.get(handler.id)
if source_plugin == plugin_name:
self.handler_to_worker.pop(handler.id, None)
self._handler_sources.pop(handler.id, None)
# 从 loaded_plugins 中移除
if plugin_name in self.loaded_plugins:
self.loaded_plugins.remove(plugin_name)

View File

@@ -2,7 +2,8 @@ from __future__ import annotations
import asyncio
import inspect
from typing import Any
import typing
from typing import Any, get_type_hints
from ..context import CancelToken, Context
from ..events import MessageEvent, PlainTextResult
@@ -80,24 +81,111 @@ class HandlerDispatcher:
ctx: Context,
legacy_args: dict[str, Any] | None = None,
) -> list[Any]:
"""构建 handler 参数列表。
注入优先级:
1. 按类型注解注入(支持 Optional[Type]
2. 按参数名注入(兼容无类型注解的情况)
3. 从 legacy_args 注入命令参数、regex 捕获组等)
Args:
handler: Handler 可调用对象
event: 消息事件
ctx: 运行时上下文
legacy_args: 旧版参数字典
Returns:
参数列表
"""
from loguru import logger
signature = inspect.signature(handler)
args: list[Any] = []
legacy_args = legacy_args or {}
# 尝试获取类型注解
type_hints: dict[str, Any] = {}
try:
type_hints = get_type_hints(handler)
except Exception:
pass
for parameter in signature.parameters.values():
if parameter.kind not in (
inspect.Parameter.POSITIONAL_ONLY,
inspect.Parameter.POSITIONAL_OR_KEYWORD,
):
continue
if parameter.name == "event":
args.append(event)
elif parameter.name in {"ctx", "context"}:
args.append(ctx)
elif parameter.name in legacy_args:
# 支持从 legacy args 中注入参数如命令参数、regex 捕获组等)
args.append(legacy_args[parameter.name])
injected = None
# 1. 优先按类型注解注入
param_type = type_hints.get(parameter.name)
if param_type is not None:
injected = self._inject_by_type(param_type, event, ctx)
# 2. Fallback 按名字注入
if injected is None:
if parameter.name == "event":
injected = event
elif parameter.name in {"ctx", "context"}:
injected = ctx
elif parameter.name in legacy_args:
injected = legacy_args[parameter.name]
# 3. 检查是否有默认值
if injected is None:
if parameter.default is not parameter.empty:
# 有默认值,跳过注入
continue
# 无默认值且无法注入,警告并传 None
logger.warning(
f"Handler '{handler.__name__}': 参数 '{parameter.name}' "
f"无法注入(类型: {param_type or '未知'}),将传入 None"
)
args.append(None)
else:
args.append(injected)
return args
def _inject_by_type(
self, param_type: Any, event: MessageEvent, ctx: Context
) -> Any:
"""根据类型注解注入参数。
支持 Optional[Type] 类型。
Args:
param_type: 参数类型注解
event: 消息事件
ctx: 运行时上下文
Returns:
注入的值,若无法注入则返回 None
"""
# 处理 Optional[Type] 情况
origin = typing.get_origin(param_type)
if origin is typing.Union:
args = typing.get_args(param_type)
non_none_types = [a for a in args if a is not type(None)]
if len(non_none_types) == 1:
param_type = non_none_types[0]
# 注入 MessageEvent 及其子类
if param_type is MessageEvent or (
isinstance(param_type, type) and issubclass(param_type, MessageEvent)
):
return event
# 注入 Context 及其子类
if param_type is Context or (
isinstance(param_type, type) and issubclass(param_type, Context)
):
return ctx
return None
async def _consume_legacy_result(self, item: Any, event: MessageEvent) -> None:
if isinstance(item, PlainTextResult):
await event.reply(item.text)