fz-mirrors/AstrBot

Fork 0

mirror of https://github.com/AstrBotDevs/AstrBot synced 2026-07-15 17:30:13 +08:00

Files

LIghtJUNction 868c81bbc7 chore: commit all changes

2026-04-02 21:23:23 +08:00

13 KiB

Raw Blame History

Agent 消息处理流程规范

概述

本文档定义 Agent 层消息处理的设计规范与契约（语言中立）。目标是提供一套清晰的、可实现的消息流、缓冲、流控、工具调用与安全策略，使不同语言或运行时实现都能遵循统一语义与行为。

注意事项：

本文档只描述设计与接口语义，不绑定任何具体实现语言或特定库。
所有数据结构应为可序列化格式（例如 JSON），便于跨进程或跨语言传输与测试。
实现必须清楚描述并发语义、错误处理、资源释放与边界情况。

目录（概要）

工具、技能与 Agent 协作体系
输入缓冲区（Input Buffer）
流控引擎（Flow Control）
Agent 核心（Context 管理与运行时）
工具调用策略（Tool Calling Strategy）
安全层（Security）
权限模型（Permission）
输出缓冲区（Output Buffer）
记忆管理（Memory）
平台适配（Platform Adaptation）
配置汇总
错误处理与恢复
扩展点与插件接口

1. 工具、技能与 Agent 协作体系

设计要点：

将工具（internal）、外部工具（MCP/remote）、技能（skills）视为统一的“工具集”层级。Agent 在执行时向工具集发起调用，由工具路由器（Tool Router）负责选择与调度。
支持多来源工具合并：本地内建工具、注册的外部 MCP 服务器、可加载的 skills。
提供工具元数据（schema、并发限制、超时、权限要求），以支持运行时决策。

接口示例（语言中立签名）：

ToolRouter.register_internal_tool(name: str, schema: dict) -> None
ToolRouter.register_mcp_server(name: str, endpoint: dict) -> None
ToolRouter.route_tool_call(call: {name: str, arguments: dict}) -> {status: str, result: any}

工具选择策略应考虑：

优先级（internal > skill > mcp 可配置）
并发与速率限制
依赖关系（某些工具需按顺序调用）
权限与安全策略

Agent 协作（ACP）：

定义 Agent-to-Agent 调用契约（例如 RPC/HTTP/消息队列），包含调用接口、超时、重试与鉴权。
上层应能发现已注册 Agent 实例并列出其能力（capabilities）。

2. 输入缓冲区（Input Buffer）

目标与语义：

输入缓冲负责按会话或用户分隔消息队列，支持优先级、分段消息与限流。
缓冲应支持批出队和基于策略的溢出处理。

核心数据模型（语言中立描述）：

InputMessage（示例字段）:
- message_id (string): 全局唯一 ID
- platform (string)
- user_id (string)
- conversation_id (string)
- content (object/string)
- timestamp (ISO8601)
- metadata (dict)
- priority (int)

缓冲配置要点：

max_queue_size: 每用户/会话的最大消息数
max_message_age: 消息超期策略
overflow_strategy: [DropOldest, DropNewest, Block]
per_user_queue / per_conversation_queue 可配置

行为契约：

enqueue_message(msg: InputMessage) -> message_id：如果队列已满，按 overflow_strategy 决定是丢弃、阻塞或返回错误。
dequeue_messages(limit: int) -> list[InputMessage]：批量出队以提高吞吐。
get_queue_depth(session_id: str) -> int
clear_queue(session_id: str) -> None

实现建议：

使用分段锁或细粒度并发结构避免全局锁。
明确队列边界与持久化选项（内存/磁盘/数据库）。

3. 流控引擎（Flow Control）

目标：

在全局与会话层实现速率限制，保护上游 LLM 提供方与下游平台，避免突发流量导致超限或费用高涨。

常见策略：

令牌桶（Token Bucket）与漏桶
按 API-key / 会话 / 平台的分层限流
自动限流（根据错误率或响应延迟自适应）

接口契约：

set_rate_limit(scope: str, requests: int, period_seconds: float) -> None
acquire(scope: str) -> bool（非阻塞）
wait_for_token(scope: str, timeout_seconds: float) -> bool（阻塞/等待）

配置细节：

safety_margin：预留给突发重试的额外容量
min_interval / max_interval：自适应限流的上下界

实现建议：

在高并发场景下使用原子操作或专用速率库实现高性能令牌桶。
将统计和指标暴露用于监控与告警。

4. Agent 核心（Agent Core）

职责：

管理用户会话上下文（context）、系统提示、工具与内存（memory）。
执行 Agent 的消息处理循环：接收输入 → 权限检查 → 安全过滤 → 工具调用决策 → LLM 调用 → 输出分发。

核心概念：

AgentContext:
- messages: 历史消息列表（按时间排序）
- system_prompt: 系统级提示
- tools: 可用工具清单与元数据
- memory: 与会话关联的记忆项
- metadata: 额外上下文信息

行为契约：

ContextManager.build_context(agent_id, recent_messages, limit) -> context_payload
支持上下文压缩（compress）策略以控制 tokens/长度（例如 summarize、truncate_by_turns 等）
明确并文档化 max_context_length 的单位（tokens / characters / turns）

实现建议：

提供异步构建上下文接口（以避免阻塞事件循环）
将上下文管理与存储分离（缓存 + 后端存储）

5. 工具调用策略（Tool Calling Strategy）

目标：

定义工具调用的执行策略（并行/顺序/失败重试/超时/回退）。
管理工具依赖、分组与并行度。

配置要点：

strategy: e.g., "sequential", "parallel", "dependency-aware"
max_calls_per_request
timeout（每个工具调用）
max_retries, retry_backoff
parallel_calls, max_parallel_calls

行为契约：

execute_tools(request_id: str, calls: list[ToolCall]) -> list[ToolResult]
工具调用结果应包含：id, name, status（success/fail/timeout）, result, error（可序列化）

工具选择策略示例：

优先最近使用（recency boost）
按相关度和能力（schema matching）选择 best-fit 工具
避免超过并发与配额限制

实现建议：

提供工具调用审计日志，支持回放与重试。
在并行执行时以安全方式聚合结果并保证顺序语义（如需要）。

6. 安全层（Security Layer）

目标：

在输入与输出路径实现注入检测、内容过滤、敏感信息屏蔽与泄露防护。

策略：

使用可配置的检测规则集合（正则、关键字、模式、ML-based classifier）
在发现注入时根据策略选择动作：block, warn, sanitize, redact
对输出执行隐私检测（例如屏蔽 secrets、标识个人信息）

接口契约：

SecurityFilter.filter_messages(messages: list, mode: str) -> {filtered_messages, detections}
SecurityFilter.filter_output(output: str) -> {safe_output, detections}

实现建议：

将规则编译为高效匹配结构，避免在高吞吐下成为瓶颈。
为检测提供事件与审计日志，便于事后分析。

7. 权限模型（Permission Model）

设计目标：

提供基于角色的权限模型（RBAC），支持命令级权限、会话级白名单/黑名单与资源配额。

要点：

角色（Owner, Admin, Member, Guest, Blocked）
权限项（capabilities）可细粒度到 API/命令/工具
会话级别覆盖全局策略

接口契约：

RoleManager.get_role(user_id, conversation_id) -> Role
PermissionMiddleware.check_message(event, context) -> PermissionResult {allowed: bool, reason: str}

实现建议：

提供默认策略并允许动态配置、继承与角色映射。
在权限拒绝处返回可展示的用户提示语以改善用户体验。

8. 输出缓冲区（Output Buffer）

职责：

管理对外发送的消息队列与分发策略（streaming, segmented, full）。
提供按会话的结果队列、流式发送以及平台适配。

数据模型：

OutputMessage:
- session_id, content, format (plain/markdown/html), strategy (streaming/segmented/full), metadata

接口契约：

enqueue_result(session_id, result) -> result_id
dequeue_result(session_id) -> OutputMessage | None
set_dispatch_strategy(session_id, strategy)

输出策略：

Streaming：分片推送，适合实时交互
Segmented：根据语言/句子边界分段发送
Full：一次性发送完整响应

实现建议：

分段器（Segmenter）应支持语言与平台特定规则。
提供 backpressure 机制避免下游拥堵。

9. 记忆管理（Memory Management）

目标：

支持短期工作记忆（working）与长期记忆（semantic）存储与检索。
提供记忆压缩、摘要与定期清理策略。

配置要点：

backend：sqlite/redis/remote vector DB 等
retention、auto_summary_threshold、working_memory_days

接口契约：

MemoryBank.add(entry) -> entry_id
MemoryBank.search(query, top_k) -> list[MemoryEntry]
MemoryBank.summarize_old() -> None

实现建议：

支持 embedding 存储与检索（若使用 vector DB）。
在写路径增加去重与重要性评分。

10. 平台适配（Platform Adaptation）

目标：

将抽象输出与行为适配到不同平台（Telegram, Discord, Webchat 等），考虑平台最大长度、支持的格式、mention/quote 行为。

要点：

平台能力描述（supports_streaming, max_message_length, supports_markdown, supports_mentions...）
平台策略选择器负责为特定平台选择最合适的输出策略与格式。

接口契约：

PlatformAdapter.render(output_message, platform_capabilities) -> platform_payload

实现建议：

将平台特性维护为可配置字典，并提供测试矩阵。

11. 配置汇总

重要配置域：

input_buffer, flow_control, context, tool_calling, security, output, memory 等
优先级：环境变量 > secrets > 用户配置 > 默认值

建议：

使用 JSON Schema 或等价方案为每个配置文件定义 schema，便于验证与 UI 生成。

12. 错误处理与恢复

分类：

RateLimit, Timeout, Network, Api, Tool, Security, Internal

策略：

为每个错误类型定义可配置的动作（Retry, Fail, Block, Fallback）
提供统一的错误对象：
- { code: int, name: str, message: str, details: dict }

行为契约：

所有对外接口在失败时返回结构化错误，调用者不得假定隐式成功。
关键路径应提供幂等性保证（或文档化其非幂等行为）。

实现建议：

在工具调用与 LLM 请求路径实现可配置的重试策略与指数退避。
在错误统计与监控系统中上报错误码与聚合指标以便告警。

13. 扩展点（Plugins / Hooks）

支持点：

InputBufferPlugin：pre_add_message / post_add_message
OutputBufferPlugin：pre_send_message / post_send_message
SecurityPlugin：check_injection / filter_content
Scheduler 插件：自定义调度器选择下一个消息

契约：

插件通过稳定的接口注册并被沙箱化执行（若属于同一进程，需明确资源与权限边界）。
插件注册表应保留版本与能力元数据，支持热加载/卸载。

调试、测试与验证建议

提供端到端回环（loopback）transport 与“虚拟工具”以在 CI 中进行集成测试。
为每个模块（缓冲、流控、工具调用、安全）编写单元与压力测试用例，覆盖边界条件（队列溢出、速率突变、工具超时）。
导出丰富的指标（队列深度、工具调用延迟、错误计数、令牌消耗），并在实现中保持可观测性（logs, traces, metrics）。
实现一致的日志格式与结构化事件，便于聚合与分析。

可序列化示例数据模型（JSON Schema 风格，语言中立）

InputMessage（示例字段说明）：

message_id: string
platform: string
user_id: string
conversation_id: string
content: object | string
timestamp: ISO8601 string
metadata: object
priority: integer

ToolCall（请求示例）：

id: string
name: string
arguments: object

ToolResult（返回示例）：

id: string
name: string
status: "success" | "fail" | "timeout"
result: any
error: { code: int, message: string } | null

ErrorObject（统一错误）：

code: int
name: string
message: string
details: object | null

结束语

本规范为语言中立的 Agent 层消息处理设计文档，旨在为实现方（无论使用何种语言/运行时）提供统一的接口、数据模型与行为约定。实现时请务必补充：

具体实现细节（并发原语、线程模型或事件循环策略）
绑定/序列化策略（如何在不同进程或语言之间传输数据）
运行时监控与故障恢复方案

如需我把本规范中的某个模块进一步细化成接口定义、JSON Schema 或示例调用序列（包含错误路径），告诉我具体模块与目标实现语言/环境（例如 Python 异步服务、Go 微服务、或独立进程插件），我会继续输出详细设计。

13 KiB Raw Blame History Unescape Escape

Agent 消息处理流程规范

概述

目录（概要）

1. 工具、技能与 Agent 协作体系

2. 输入缓冲区（Input Buffer）

3. 流控引擎（Flow Control）

4. Agent 核心（Agent Core）

5. 工具调用策略（Tool Calling Strategy）

6. 安全层（Security Layer）

7. 权限模型（Permission Model）

8. 输出缓冲区（Output Buffer）

9. 记忆管理（Memory Management）

10. 平台适配（Platform Adaptation）

11. 配置汇总

12. 错误处理与恢复

13. 扩展点（Plugins / Hooks）

调试、测试与验证建议

可序列化示例数据模型（JSON Schema 风格，语言中立）

13 KiB

Raw Blame History