From 222378b768d7237b67986db19732a463550906aa Mon Sep 17 00:00:00 2001 From: "astrbot-docs-agent[bot]" Date: Thu, 19 Feb 2026 10:08:34 +0000 Subject: [PATCH] docs: update for AstrBotDevs/AstrBot#5208 --- en/dev/star/guides/send-message.md | 45 ++ zh/dev/star/guides/listen-message-event.md | 8 + zh/dev/star/guides/send-message.md | 45 ++ zh/dev/star/plugin.md | 826 +-------------------- 4 files changed, 120 insertions(+), 804 deletions(-) diff --git a/en/dev/star/guides/send-message.md b/en/dev/star/guides/send-message.md index 417b60e..fe0f5b0 100644 --- a/en/dev/star/guides/send-message.md +++ b/en/dev/star/guides/send-message.md @@ -129,3 +129,48 @@ async def test(self, event: AstrMessageEvent): ``` ![Sending group forward messages](https://files.astrbot.app/docs/source/images/plugin/image-4.png) + +## Other Message Types + +> Some platforms may not support all of the following message types. + +**JSON Card** + +```python +Comp.Json(data={...}) +``` + +**Share** + +```python +Comp.Share(url="...", title="...", content="...", image="...") +``` + +**Music Share** + +```python +# Method 1: Specify platform and ID +Comp.Music(type="qq", id="...") +# Method 2: Custom +Comp.Music(type="custom", url="...", audio="...", title="...", content="...", image="...") +``` + +**Location** + +```python +Comp.Location(lat=..., lon=..., title="...", content="...") +``` + +**Contact (Friend/Group)** + +```python +Comp.Contact(type="qq", id="...") +``` + +**Magic Emoticons/Actions** + +```python +Comp.Shake() # Window shake/Poke +Comp.Dice() # Roll a dice +Comp.RPS() # Rock-paper-scissors +``` diff --git a/zh/dev/star/guides/listen-message-event.md b/zh/dev/star/guides/listen-message-event.md index 9fe083c..ec69bf5 100644 --- a/zh/dev/star/guides/listen-message-event.md +++ b/zh/dev/star/guides/listen-message-event.md @@ -54,6 +54,7 @@ class AstrBotMessage: - `Record`：语音消息段 - `Video`：视频消息段 - `File`：文件消息段 +- `Music`：音乐分享消息段 大多数消息平台都支持上面的消息段类型。 @@ -63,6 +64,13 @@ class AstrBotMessage: - `Node`：合并转发消息中的一个节点 - `Nodes`：合并转发消息中的多个节点 - `Poke`：戳一戳消息段 +- `Json`：JSON 卡片消息段 +- `Share`：分享消息段 +- `Location`：位置消息段 +- `Contact`：推荐好友/群消息段 +- `Shake`：窗口抖动/戳一戳消息段 +- `Dice`：掷骰子消息段 +- `RPS`：猜拳消息段 在 AstrBot 中，消息链表示为 `List[BaseMessageComponent]` 类型的列表。 diff --git a/zh/dev/star/guides/send-message.md b/zh/dev/star/guides/send-message.md index 84eaf8e..99438c2 100644 --- a/zh/dev/star/guides/send-message.md +++ b/zh/dev/star/guides/send-message.md @@ -129,3 +129,48 @@ async def test(self, event: AstrMessageEvent): ``` ![发送群合并转发消息](https://files.astrbot.app/docs/source/images/plugin/image-4.png) + +## 其他消息类型 + +> 部分平台可能不支持以下所有消息类型。 + +**JSON 卡片** + +```python +Comp.Json(data={...}) +``` + +**分享 Share** + +```python +Comp.Share(url="...", title="...", content="...", image="...") +``` + +**音乐分享 Music** + +```python +# 方式 1：指定平台和 ID +Comp.Music(type="qq", id="...") +# 方式 2：自定义 +Comp.Music(type="custom", url="...", audio="...", title="...", content="...", image="...") +``` + +**位置 Location** + +```python +Comp.Location(lat=..., lon=..., title="...", content="...") +``` + +**推荐好友/群 Contact** + +```python +Comp.Contact(type="qq", id="...") +``` + +**魔法表情/动作** + +```python +Comp.Shake() # 窗口抖动/戳一戳 +Comp.Dice() # 掷骰子 +Comp.RPS() # 猜拳 +``` diff --git a/zh/dev/star/plugin.md b/zh/dev/star/plugin.md index cda2cd2..3eb2924 100644 --- a/zh/dev/star/plugin.md +++ b/zh/dev/star/plugin.md @@ -160,6 +160,13 @@ ComponentTypes = { "node": Node, # 转发消息中的节点 "nodes": Nodes, # Node 的列表，用于支持一个转发消息中的多个节点 "poke": Poke, # 戳一戳 + "json": Json, # JSON 卡片 + "share": Share, # 分享 + "location": Location, # 位置 + "contact": Contact, # 推荐好友/群 + "shake": Shake, # 窗口抖动/戳一戳 + "dice": Dice, # 掷骰子 + "rps": RPS, # 猜拳 } ``` @@ -543,7 +550,7 @@ async def helloworld(self, event: AstrMessageEvent): > [!TIP] > 关于 unified_msg_origin。 -> unified_msg_origin 是一个字符串，记录了一个会话的唯一 ID，AstrBot 能够据此找到属于哪个消息平台的哪个会话。这样就能够实现在 `send_message` 的时候，发送消息到正确的会话。有关 MessageChain，请参见接下来的一节。 +> unified_msg_origin 是 a 字符串，记录了一个会话的唯一 ID，AstrBot 能够据此找到属于哪个消息平台的哪个会话。这样就能够实现在 `send_message` 的时候，发送消息到正确的会话。有关 MessageChain，请参见接下来的一节。 #### 富媒体消息 @@ -779,814 +786,25 @@ async def on_aiocqhttp(self, event: AstrMessageEvent, text: str): ``` -![image](https://files.astrbot.app/docs/source/images/plugin/image-3.png) +![image](https://files.astrbot.app/docs/source/images/plugin/image-8.png) -#### 自定义(基于 HTML) +#### 进阶 -如果你觉得上面渲染出来的图片不够美观，你可以使用自定义的 HTML 模板来渲染图片。 - -AstrBot 支持使用 `HTML + Jinja2` 的方式来渲染文转图模板。 - -```py{7} -# 自定义的 Jinja2 模板，支持 CSS -TMPL = ''' -
-

Todo List

- -
-''' - -@filter.command("todo") -async def custom_t2i_tmpl(self, event: AstrMessageEvent): - options = {} # 可选择传入渲染选项。 - url = await self.html_render(TMPL, {"items": ["吃饭", "睡觉", "玩原神"]}, options=options) # 第二个参数是 Jinja2 的渲染数据 - yield event.image_result(url) -``` - -返回的结果: - -![image](https://files.astrbot.app/docs/source/images/plugin/fcc2dcb472a91b12899f617477adc5c7.png) - -这只是一个简单的例子。得益于 HTML 和 DOM 渲染器的强大性，你可以进行更复杂和更美观的的设计。除此之外，Jinja2 支持循环、条件等语法以适应列表、字典等数据结构。你可以从网上了解更多关于 Jinja2 的知识。 - -**图片渲染选项(options)**： - -请参考 Playwright 的 [screenshot](https://playwright.dev/python/docs/api/class-page#page-screenshot) API。 - -- `timeout` (float, optional): 截图超时时间. -- `type` (Literal["jpeg", "png"], optional): 截图图片类型. -- `quality` (int, optional): 截图质量，仅适用于 JPEG 格式图片. -- `omit_background` (bool, optional): 是否允许隐藏默认的白色背景，这样就可以截透明图了，仅适用于 PNG 格式 -- `full_page` (bool, optional): 是否截整个页面而不是仅设置的视口大小，默认为 True. -- `clip` (dict, optional): 截图后裁切的区域。参考 Playwright screenshot API。 -- `animations`: (Literal["allow", "disabled"], optional): 是否允许播放 CSS 动画. -- `caret`: (Literal["hide", "initial"], optional): 当设置为 hide 时，截图时将隐藏文本插入符号，默认为 hide. -- `scale`: (Literal["css", "device"], optional): 页面缩放设置. 当设置为 css 时，则将设备分辨率与 CSS 中的像素一一对应，在高分屏上会使得截图变小. 当设置为 device 时，则根据设备的屏幕缩放设置或当前 Playwright 的 Page/Context 中的 device_scale_factor 参数来缩放. -- `mask` (List["Locator"]], optional): 指定截图时的遮罩的 Locator。元素将被一颜色为 #FF00FF 的框覆盖. - -### 会话控制 - -> 大于等于 v3.4.36 - -为什么需要会话控制？考虑一个 成语接龙 插件，某个/群用户需要和机器人进行多次对话，而不是一次性的指令。这时候就需要会话控制。 - -```txt -用户: /成语接龙 -机器人: 请发送一个成语 -用户: 一马当先 -机器人: 先见之明 -用户: 明察秋毫 -... -``` - -AstrBot 提供了开箱即用的会话控制功能： - -导入： - -```py -import astrbot.api.message_components as Comp -from astrbot.core.utils.session_waiter import ( - session_waiter, - SessionController, -) -``` - -handler 内的代码可以如下： +你可以自定义渲染的样式。 ```python -from astrbot.api.event import filter, AstrMessageEvent - -@filter.command("成语接龙") -async def handle_empty_mention(self, event: AstrMessageEvent): - """成语接龙具体实现""" - try: - yield event.plain_result("请发送一个成语~") - - # 具体的会话控制器使用方法 - @session_waiter(timeout=60, record_history_chains=False) # 注册一个会话控制器，设置超时时间为 60 秒，不记录历史消息链 - async def empty_mention_waiter(controller: SessionController, event: AstrMessageEvent): - idiom = event.message_str # 用户发来的成语，假设是 "一马当先" - - if idiom == "退出": # 假设用户想主动退出成语接龙，输入了 "退出" - await event.send(event.plain_result("已退出成语接龙~")) - controller.stop() # 停止会话控制器，会立即结束。 - return - - if len(idiom) != 4: # 假设用户输入的不是4字成语 - await event.send(event.plain_result("成语必须是四个字的呢~")) # 发送回复，不能使用 yield - return - # 退出当前方法，不执行后续逻辑，但此会话并未中断，后续的用户输入仍然会进入当前会话 - - # ... - message_result = event.make_result() - message_result.chain = [Comp.Plain("先见之明")] # import astrbot.api.message_components as Comp - await event.send(message_result) # 发送回复，不能使用 yield - - controller.keep(timeout=60, reset_timeout=True) # 重置超时时间为 60s，如果不重置，则会继续之前的超时时间计时。 - - # controller.stop() # 停止会话控制器，会立即结束。 - # 如果记录了历史消息链，可以通过 controller.get_history_chains() 获取历史消息链 - - try: - await empty_mention_waiter(event) - except TimeoutError as _: # 当超时后，会话控制器会抛出 TimeoutError - yield event.plain_result("你超时了！") - except Exception as e: - yield event.plain_result("发生错误，请联系管理员: " + str(e)) - finally: - event.stop_event() - except Exception as e: - logger.error("handle_empty_mention error: " + str(e)) -``` - -当激活会话控制器后，该发送人之后发送的消息会首先经过上面你定义的 `empty_mention_waiter` 函数处理，直到会话控制器被停止或者超时。 - -#### SessionController - -用于开发者控制这个会话是否应该结束，并且可以拿到历史消息链。 - -- keep(): 保持这个会话 - - timeout (float): 必填。会话超时时间。 - - reset_timeout (bool): 设置为 True 时, 代表重置超时时间, timeout 必须 > 0, 如果 <= 0 则立即结束会话。设置为 False 时, 代表继续维持原来的超时时间, 新 timeout = 原来剩余的 timeout + timeout (可以 < 0) -- stop(): 结束这个会话 -- get_history_chains() -> List[List[Comp.BaseMessageComponent]]: 获取历史消息链 - -#### 自定义会话 ID 算子 - -默认情况下，AstrBot 会话控制器会将基于 `sender_id` （发送人的 ID）作为识别不同会话的标识，如果想将一整个群作为一个会话，则需要自定义会话 ID 算子。 - -```py -import astrbot.api.message_components as Comp -from astrbot.core.utils.session_waiter import ( - session_waiter, - SessionFilter, - SessionController, -) - -# 沿用上面的 handler -# ... -class CustomFilter(SessionFilter): - def filter(self, event: AstrMessageEvent) -> str: - return event.get_group_id() if event.get_group_id() else event.unified_msg_origin - -await empty_mention_waiter(event, session_filter=CustomFilter()) # 这里传入 session_filter -# ... -``` - -这样之后，当群内一个用户发送消息后，会话控制器会将这个群作为一个会话，群内其他用户发送的消息也会被认为是同一个会话。 - -甚至，可以使用这个特性来让群内组队！ - -### AI - -#### 通过提供商调用 LLM - -获取提供商有以下几种方式: - -- 获取当前使用的大语言模型提供商: `self.context.get_using_provider(umo=event.unified_msg_origin)`。 -- 根据 ID 获取大语言模型提供商: `self.context.get_provider_by_id(provider_id="xxxx")`。 -- 获取所有大语言模型提供商: `self.context.get_all_providers()`。 - -```python -from astrbot.api.event import filter, AstrMessageEvent - -@filter.command("test") -async def test(self, event: AstrMessageEvent): - # func_tools_mgr = self.context.get_llm_tool_manager() - prov = self.context.get_using_provider(umo=event.unified_msg_origin) - if prov: - llm_resp = await provider.text_chat( - prompt="Hi!", - context=[ - {"role": "user", "content": "balabala"}, - {"role": "assistant", "content": "response balabala"} - ], - system_prompt="You are a helpful assistant." - ) - print(llm_resp) -``` - -`Provider.text_chat()` 用于请求 LLM。其返回 `LLMResponse` 方法。除了上面的三个参数，其还支持: - -- `func_tool`(ToolSet): 可选。用于传入函数工具。参考 [函数工具](#函数工具)。 -- `image_urls`(List[str]): 可选。用于传入请求中带有的图片 URL 列表。支持文件路径。 -- `model`(str): 可选。用于强制指定使用的模型。默认使用这个提供商默认配置的模型。 -- `tool_calls_result`(dict): 可选。用于传入工具调用的结果。 - -::: details LLMResponse 类型定义 - -```py - -@dataclass -class LLMResponse: - role: str - """角色, assistant, tool, err""" - result_chain: MessageChain = None - """返回的消息链""" - tools_call_args: List[Dict[str, any]] = field(default_factory=list) - """工具调用参数""" - tools_call_name: List[str] = field(default_factory=list) - """工具调用名称""" - tools_call_ids: List[str] = field(default_factory=list) - """工具调用 ID""" - - raw_completion: ChatCompletion = None - _new_record: Dict[str, any] = None - - _completion_text: str = "" - - is_chunk: bool = False - """是否是流式输出的单个 Chunk""" - - def __init__( - self, - role: str, - completion_text: str = "", - result_chain: MessageChain = None, - tools_call_args: List[Dict[str, any]] = None, - tools_call_name: List[str] = None, - tools_call_ids: List[str] = None, - raw_completion: ChatCompletion = None, - _new_record: Dict[str, any] = None, - is_chunk: bool = False, - ): - """初始化 LLMResponse - - Args: - role (str): 角色, assistant, tool, err - completion_text (str, optional): 返回的结果文本，已经过时，推荐使用 result_chain. Defaults to "". - result_chain (MessageChain, optional): 返回的消息链. Defaults to None. - tools_call_args (List[Dict[str, any]], optional): 工具调用参数. Defaults to None. - tools_call_name (List[str], optional): 工具调用名称. Defaults to None. - raw_completion (ChatCompletion, optional): 原始响应, OpenAI 格式. Defaults to None. - """ - if tools_call_args is None: - tools_call_args = [] - if tools_call_name is None: - tools_call_name = [] - if tools_call_ids is None: - tools_call_ids = [] - - self.role = role - self.completion_text = completion_text - self.result_chain = result_chain - self.tools_call_args = tools_call_args - self.tools_call_name = tools_call_name - self.tools_call_ids = tools_call_ids - self.raw_completion = raw_completion - self._new_record = _new_record - self.is_chunk = is_chunk - - @property - def completion_text(self): - if self.result_chain: - return self.result_chain.get_plain_text() - return self._completion_text - - @completion_text.setter - def completion_text(self, value): - if self.result_chain: - self.result_chain.chain = [ - comp - for comp in self.result_chain.chain - if not isinstance(comp, Comp.Plain) - ] # 清空 Plain 组件 - self.result_chain.chain.insert(0, Comp.Plain(value)) - else: - self._completion_text = value - - def to_openai_tool_calls(self) -> List[Dict]: - """将工具调用信息转换为 OpenAI 格式""" - ret = [] - for idx, tool_call_arg in enumerate(self.tools_call_args): - ret.append( - { - "id": self.tools_call_ids[idx], - "function": { - "name": self.tools_call_name[idx], - "arguments": json.dumps(tool_call_arg), - }, - "type": "function", - } - ) - return ret -``` - -::: - -#### 获取其他类型的提供商 - -> 嵌入、重排序 没有 “当前使用”。这两个提供商主要用于知识库。 - -- 获取当前使用的语音识别提供商(STTProvider): `self.context.get_using_stt_provider(umo=event.unified_msg_origin)`。 -- 获取当前使用的语音合成提供商(TTSProvider): `self.context.get_using_tts_provider(umo=event.unified_msg_origin)`。 -- 获取所有语音识别提供商: `self.context.get_all_stt_providers()`。 -- 获取所有语音合成提供商: `self.context.get_all_tts_providers()`。 -- 获取所有嵌入提供商: `self.context.get_all_embedding_providers()`。 - -::: details STTProvider / TTSProvider / EmbeddingProvider 类型定义 - -```py -class TTSProvider(AbstractProvider): - def __init__(self, provider_config: dict, provider_settings: dict) -> None: - super().__init__(provider_config) - self.provider_config = provider_config - self.provider_settings = provider_settings - - @abc.abstractmethod - async def get_audio(self, text: str) -> str: - """获取文本的音频，返回音频文件路径""" - raise NotImplementedError() - - -class EmbeddingProvider(AbstractProvider): - def __init__(self, provider_config: dict, provider_settings: dict) -> None: - super().__init__(provider_config) - self.provider_config = provider_config - self.provider_settings = provider_settings - - @abc.abstractmethod - async def get_embedding(self, text: str) -> list[float]: - """获取文本的向量""" - ... - - @abc.abstractmethod - async def get_embeddings(self, text: list[str]) -> list[list[float]]: - """批量获取文本的向量""" - ... - - @abc.abstractmethod - def get_dim(self) -> int: - """获取向量的维度""" - ... - -class STTProvider(AbstractProvider): - def __init__(self, provider_config: dict, provider_settings: dict) -> None: - super().__init__(provider_config) - self.provider_config = provider_config - self.provider_settings = provider_settings - - @abc.abstractmethod - async def get_text(self, audio_url: str) -> str: - """获取音频的文本""" - raise NotImplementedError() -``` - -::: - -#### 函数工具 - -函数工具给了大语言模型调用外部工具的能力。在 AstrBot 中，函数工具有多种定义方式。 - -##### 以类的形式（推荐） - -推荐在插件目录下新建 `tools` 文件夹，然后在其中编写工具类： - -`tools/search.py`: - -```py -from astrbot.api import FunctionTool -from astrbot.api.event import AstrMessageEvent -from dataclasses import dataclass, field - -@dataclass -class HelloWorldTool(FunctionTool): - name: str = "hello_world" # 工具名称 - description: str = "Say hello to the world." # 工具描述 - parameters: dict = field( - default_factory=lambda: { - "type": "object", - "properties": { - "greeting": { - "type": "string", - "description": "The greeting message.", - }, - }, - "required": ["greeting"], - } - ) # 工具参数定义，见 OpenAI 官网或 https://json-schema.org/understanding-json-schema/ - - async def run( - self, - event: AstrMessageEvent, # 必须包含此 event 参数在前面，用于获取上下文 - greeting: str, # 工具参数，必须与 parameters 中定义的参数名一致 - ): - return f"{greeting}, World!" # 也支持 mcp.types.CallToolResult 类型 -``` - -要将上述工具注册到 AstrBot，可以在插件主文件的 `__init__.py` 中添加以下代码： - -```py -from .tools.search import SearchTool - -class MyPlugin(Star): - def __init__(self, context: Context): - super().__init__(context) - # >= v4.5.1 使用： - self.context.add_llm_tools(HelloWorldTool(), SecondTool(), ...) - - # < v4.5.1 之前使用： - tool_mgr = self.context.provider_manager.llm_tools - tool_mgr.func_list.append(HelloWorldTool()) -``` - -##### 以装饰器的形式 - -这个形式定义的工具函数会被自动加载到 AstrBot Core 中，在 Core 请求大模型时会被自动带上。 - -请务必按照以下格式编写一个工具（包括**函数注释**，AstrBot 会解析该函数注释，请务必将注释格式写对） - -```py{3,4,5,6,7} -@filter.llm_tool(name="get_weather") # 如果 name 不填，将使用函数名 -async def get_weather(self, event: AstrMessageEvent, location: str) -> MessageEventResult: - '''获取天气信息。 - - Args: - location(string): 地点 - ''' - resp = self.get_weather_from_api(location) - yield event.plain_result("天气信息: " + resp) -``` - -在 `location(string): 地点` 中，`location` 是参数名，`string` 是参数类型，`地点` 是参数描述。 - -支持的参数类型有 `string`, `number`, `object`, `boolean`。 - -> [!NOTE] -> 对于装饰器注册的 llm_tool，如果需要调用 Provider.text_chat()，func_tool（ToolSet 类型） 可以通过以下方式获取： -> -> ```py -> func_tool = self.context.get_llm_tool_manager() # 获取 AstrBot 的 LLM Tool Manager，包含了所有插件和 MCP 注册的 Tool -> tool = func_tool.get_func("xxx") -> if tool: -> tool_set = ToolSet() -> tool_set.add_tool(tool) -> ``` - -#### 对话管理器 ConversationManager - -**获取会话当前的 LLM 对话历史** - -```py -from astrbot.core.conversation_mgr import Conversation - -uid = event.unified_msg_origin -conv_mgr = self.context.conversation_manager -curr_cid = await conv_mgr.get_curr_conversation_id(uid) -conversation = await conv_mgr.get_conversation(uid, curr_cid) # Conversation -``` - -::: details Conversation 类型定义 - -```py -@dataclass -class Conversation: - """LLM 对话类 - - 对于 WebChat，history 存储了包括指令、回复、图片等在内的所有消息。 - 对于其他平台的聊天，不存储非 LLM 的回复（因为考虑到已经存储在各自的平台上）。 - - 在 v4.0.0 版本及之后，WebChat 的历史记录被迁移至 `PlatformMessageHistory` 表中， - """ - - platform_id: str - user_id: str - cid: str - """对话 ID, 是 uuid 格式的字符串""" - history: str = "" - """字符串格式的对话列表。""" - title: str | None = "" - persona_id: str | None = "" - """对话当前使用的人格 ID""" - created_at: int = 0 - updated_at: int = 0 -``` - -::: - -**所有方法** - -##### `new_conversation` - -- **Usage** - 在当前会话中新建一条对话，并自动切换为该对话。 -- **Arguments** - - `unified_msg_origin: str` – 形如 `platform_name:message_type:session_id` - - `platform_id: str | None` – 平台标识，默认从 `unified_msg_origin` 解析 - - `content: list[dict] | None` – 初始历史消息 - - `title: str | None` – 对话标题 - - `persona_id: str | None` – 绑定的 persona ID -- **Returns** - `str` – 新生成的 UUID 对话 ID - -##### `switch_conversation` - -- **Usage** - 将会话切换到指定的对话。 -- **Arguments** - - `unified_msg_origin: str` - - `conversation_id: str` -- **Returns** - `None` - -##### `delete_conversation` - -- **Usage** - 删除会话中的某条对话；若 `conversation_id` 为 `None`，则删除当前对话。 -- **Arguments** - - `unified_msg_origin: str` - - `conversation_id: str | None` -- **Returns** - `None` - -##### `get_curr_conversation_id` - -- **Usage** - 获取当前会话正在使用的对话 ID。 -- **Arguments** - - `unified_msg_origin: str` -- **Returns** - `str | None` – 当前对话 ID，不存在时返回 `None` - -##### `get_conversation` - -- **Usage** - 获取指定对话的完整对象；若不存在且 `create_if_not_exists=True` 则自动创建。 -- **Arguments** - - `unified_msg_origin: str` - - `conversation_id: str` - - `create_if_not_exists: bool = False` -- **Returns** - `Conversation | None` - -##### `get_conversations` - -- **Usage** - 拉取用户或平台下的全部对话列表。 -- **Arguments** - - `unified_msg_origin: str | None` – 为 `None` 时不过滤用户 - - `platform_id: str | None` -- **Returns** - `List[Conversation]` - -##### `get_filtered_conversations` - -- **Usage** - 分页 + 关键词搜索对话。 -- **Arguments** - - `page: int = 1` - - `page_size: int = 20` - - `platform_ids: list[str] | None` - - `search_query: str = ""` - - `**kwargs` – 透传其他过滤条件 -- **Returns** - `tuple[list[Conversation], int]` – 对话列表与总数 - -##### `update_conversation` - -- **Usage** - 更新对话的标题、历史记录或 persona_id。 -- **Arguments** - - `unified_msg_origin: str` - - `conversation_id: str | None` – 为 `None` 时使用当前对话 - - `history: list[dict] | None` - - `title: str | None` - - `persona_id: str | None` -- **Returns** - `None` - -##### `get_human_readable_context` - -- **Usage** - 生成分页后的人类可读对话上下文，方便展示或调试。 -- **Arguments** - - `unified_msg_origin: str` - - `conversation_id: str` - - `page: int = 1` - - `page_size: int = 10` -- **Returns** - `tuple[list[str], int]` – 当前页文本列表与总页数 - -```py -import json - -context = json.loads(conversation.history) -``` - -#### 人格设定管理器 PersonaManager - -`PersonaManager` 负责统一加载、缓存并提供所有人格（Persona）的增删改查接口，同时兼容 AstrBot 4.x 之前的旧版人格格式（v3）。 -初始化时会自动从数据库读取全部人格，并生成一份 v3 兼容数据，供旧代码无缝使用。 - -```py -persona_mgr = self.context.persona_manager -``` - -##### `get_persona` - -- **Usage** - 获取根据人格 ID 获取人格数据。 -- **Arguments** - - `persona_id: str` – 人格 ID -- **Returns** - `Persona` – 人格数据，若不存在则返回 None -- **Raises** - `ValueError` – 当不存在时抛出 - -##### `get_all_personas` - -- **Usage** - 一次性获取数据库中所有人格。 -- **Returns** - `list[Persona]` – 人格列表，可能为空 - -##### `create_persona` - -- **Usage** - 新建人格并立即写入数据库，成功后自动刷新本地缓存。 -- **Arguments** - - `persona_id: str` – 新人格 ID（唯一） - - `system_prompt: str` – 系统提示词 - - `begin_dialogs: list[str]` – 可选，开场对话（偶数条，user/assistant 交替） - - `tools: list[str]` – 可选，允许使用的工具列表；`None`=全部工具，`[]`=禁用全部 -- **Returns** - `Persona` – 新建后的人格对象 -- **Raises** - `ValueError` – 若 `persona_id` 已存在 - -##### `update_persona` - -- **Usage** - 更新现有人格的任意字段，并同步到数据库与缓存。 -- **Arguments** - - `persona_id: str` – 待更新的人格 ID - - `system_prompt: str` – 可选，新的系统提示词 - - `begin_dialogs: list[str]` – 可选，新的开场对话 - - `tools: list[str]` – 可选，新的工具列表；语义同 `create_persona` -- **Returns** - `Persona` – 更新后的人格对象 -- **Raises** - `ValueError` – 若 `persona_id` 不存在 - -##### `delete_persona` - -- **Usage** - 删除指定人格，同时清理数据库与缓存。 -- **Arguments** - - `persona_id: str` – 待删除的人格 ID -- **Raises** - `Valueable` – 若 `persona_id` 不存在 - -##### `get_default_persona_v3` - -- **Usage** - 根据当前会话配置，获取应使用的默认人格（v3 格式）。 - 若配置未指定或指定的人格不存在，则回退到 `DEFAULT_PERSONALITY`。 -- **Arguments** - - `umo: str | MessageSession | None` – 会话标识，用于读取用户级配置 -- **Returns** - `Personality` – v3 格式的默认人格对象 - -::: details Persona / Personality 类型定义 - -```py - -class Persona(SQLModel, table=True): - """Persona is a set of instructions for LLMs to follow. - - It can be used to customize the behavior of LLMs. - """ - - __tablename__ = "personas" - - id: int = Field(primary_key=True, sa_column_kwargs={"autoincrement": True}) - persona_id: str = Field(max_length=255, nullable=False) - system_prompt: str = Field(sa_type=Text, nullable=False) - begin_dialogs: Optional[list] = Field(default=None, sa_type=JSON) - """a list of strings, each representing a dialog to start with""" - tools: Optional[list] = Field(default=None, sa_type=JSON) - """None means use ALL tools for default, empty list means no tools, otherwise a list of tool names.""" - created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) - updated_at: datetime = Field( - default_factory=lambda: datetime.now(timezone.utc), - sa_column_kwargs={"onupdate": datetime.now(timezone.utc)}, - ) - - __table_args__ = ( - UniqueConstraint( - "persona_id", - name="uix_persona_id", - ), +@filter.command("image") +async def on_aiocqhttp(self, event: AstrMessageEvent, text: str): + url = await self.text_to_image( + text, + width = 500, # 图片宽度 + font_size = 20, # 字体大小 + bg_color = (255, 255, 255), # 背景颜色 + font_color = (0, 0, 0), # 字体颜色 ) - - -class Personality(TypedDict): - """LLM 人格类。 - - 在 v4.0.0 版本及之后，推荐使用上面的 Persona 类。并且， mood_imitation_dialogs 字段已被废弃。 - """ - - prompt: str - name: str - begin_dialogs: list[str] - mood_imitation_dialogs: list[str] - """情感模拟对话预设。在 v4.0.0 版本及之后，已被废弃。""" - tools: list[str] | None - """工具列表。None 表示使用所有工具，空列表表示不使用任何工具""" -``` - -::: - -### 其他 - -#### 配置文件 - -##### 默认配置文件 - -```py -config = self.context.get_config() -``` - -不建议修改默认配置文件，建议只读取。 - -##### 会话配置文件 - -v4.0.0 后，AstrBot 支持会话粒度的多配置文件。 - -```py -umo = event.unified_msg_origin -config = self.context.get_config(umo=umo) -``` - -#### 获取消息平台实例 - -> v3.4.34 后 - -```python -from astrbot.api.event import filter, AstrMessageEvent - -@filter.command("test") -async def test_(self, event: AstrMessageEvent): - from astrbot.api.platform import AiocqhttpAdapter # 其他平台同理 - platform = self.context.get_platform(filter.PlatformAdapterType.AIOCQHTTP) - assert isinstance(platform, AiocqhttpAdapter) - # platform.get_client().api.call_action() -``` - -#### 调用 QQ 协议端 API - -```py -@filter.command("helloworld") -async def helloworld(self, event: AstrMessageEvent): - if event.get_platform_name() == "aiocqhttp": - # qq - from astrbot.core.platform.sources.aiocqhttp.aiocqhttp_message_event import AiocqhttpMessageEvent - assert isinstance(event, AiocqhttpMessageEvent) - client = event.bot # 得到 client - payloads = { - "message_id": event.message_obj.message_id, - } - ret = await client.api.call_action('delete_msg', **payloads) # 调用 协议端 API - logger.info(f"delete_msg: {ret}") -``` - -关于 CQHTTP API，请参考如下文档： - -Napcat API 文档： - -Lagrange API 文档： - -#### 载入的所有插件 - -```py -plugins = self.context.get_all_stars() # 返回 StarMetadata 包含了插件类实例、配置等等 -``` - -#### 注册一个异步任务 - -直接在 **init**() 中使用 `asyncio.create_task()` 即可。 - -```py -import asyncio - -@register("task", "Soulter", "一个异步任务示例", "1.0.0") -class TaskPlugin(Star): - def __init__(self, context: Context): - super().__init__(context) - asyncio.create_task(self.my_task()) - - async def my_task(self): - await asyncio.sleep(1) - print("Hello") + yield event.image_result(url) ``` -#### 获取加载的所有平台 +### 插件市场发布 -```py -from astrbot.api.platform import Platform -platforms = self.context.platform_manager.get_insts() # List[Platform] -``` +请参考 [插件发布指南](plugin-publish.md)。