From 9db179075fd5435277f8e6713654b2a9b6dbea5b Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Fri, 27 Mar 2026 00:27:09 +0800 Subject: [PATCH] =?UTF-8?q?chore:=20=E6=B7=BB=E5=8A=A0=20MDI=20=E5=AD=97?= =?UTF-8?q?=E4=BD=93=E5=88=B0=20gitignore=20=E5=B9=B6=E6=9B=B4=E6=96=B0=20?= =?UTF-8?q?agent-message=20=E8=A7=84=E8=8C=83?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 4 + openspec/agent-message.md | 2116 ++++++++++++++++++++++--------------- 2 files changed, 1282 insertions(+), 838 deletions(-) diff --git a/.gitignore b/.gitignore index eab420672..ed7714776 100644 --- a/.gitignore +++ b/.gitignore @@ -80,3 +80,7 @@ dist/ # 拓展模块 *.so *.dll + +# MDI font subset (generated by dashboard/scripts/subset-mdi-font.mjs) +dashboard/src/assets/mdi-subset/*.woff +dashboard/src/assets/mdi-subset/*.woff2 diff --git a/openspec/agent-message.md b/openspec/agent-message.md index a2b2215de..287015a9a 100644 --- a/openspec/agent-message.md +++ b/openspec/agent-message.md @@ -110,128 +110,151 @@ AstrBot Agent 采用**双缓冲区 + 流控**的消息处理模型，实现消 ### 1.3 工具调用决策 -```python -class ToolRouter: - """工具路由""" +```rust +pub struct ToolRouter { + internal: ToolSet, + mcp: HashMap>, + skills: HashMap>, +} - def __init__( - self, - internal_toolset: ToolSet, - mcp_clients: dict[str, MCPClient], - skill_executors: dict[str, SkillExecutor], - ): - self.internal = internal_toolset - self.mcp = mcp_clients - self.skills = skill_executors +impl ToolRouter { + pub fn new( + internal: ToolSet, + mcp: HashMap>, + skills: HashMap>, + ) -> Self { + Self { internal, mcp, skills } + } - async def route_tool_call( - self, - tool_name: str, - arguments: dict, - context: AgentContext, - ) -> ToolResult: - """路由工具调用""" + /// 路由工具调用 + pub async fn route_tool_call( + &self, + tool_name: &str, + arguments: Value, + context: &mut AgentContext, + ) -> Result { + // 1. 检查内部工具 + if let Some(internal_tool) = self.internal.get_tool(tool_name) { + return self.call_internal(internal_tool, arguments, context).await; + } - # 1. 检查内部工具 - internal_tool = self.internal.get_tool(tool_name) - if internal_tool: - return await self._call_internal(internal_tool, arguments, context) + // 2. 检查 MCP 工具 + for (_, client) in &self.mcp { + if client.has_tool(tool_name) { + return client.call_tool(tool_name, arguments).await; + } + } - # 2. 检查 MCP 工具 - for client_name, client in self.mcp.items(): - if client.has_tool(tool_name): - return await client.call_tool(tool_name, arguments) + // 3. 检查 Skills + if let Some(skill) = self.skills.get(tool_name) { + return skill.execute(tool_name, arguments, context).await; + } - # 3. 检查 Skills - skill = self.skills.get(tool_name) - if skill: - return await self._execute_skill(skill, arguments, context) - - raise ToolNotFoundError(f"Tool not found: {tool_name}") + Err(ToolError::NotFound(format!("Tool not found: {}", tool_name)).into()) + } +} ``` ### 1.4 Agent 协作（ACP 协议） -```python -class ACPAgentClient: - """ACP Agent 客户端""" +```rust +pub struct ACPAgentClient { + connection: ACPConnection, +} - async def call_agent( - self, - agent_name: str, - action: str, - args: dict, - stream: bool = True, - ) -> AsyncIterator[AgentEvent] | AgentResult: - """调用远程 Agent""" - - request = ACPRequest( - method="agent/call", - params={ +impl ACPAgentClient { + /// 调用远程 Agent + pub async fn call_agent( + &self, + agent_name: &str, + action: &str, + args: Value, + stream: bool, + ) -> Result { + let request = ACPRequest { + method: "agent/call".to_string(), + params: json!({ "agent": agent_name, "action": action, "args": args, - } - ) + }), + }; - if stream: - return self._stream_request(request) - else: - return await self._send_request(request) + if stream { + Ok(AgentResult::Stream(self.connection.stream_request(request).await?)) + } else { + self.send_request(request).await + } + } - async def list_agents(self) -> list[AgentCard]: - """列出可用 Agent""" - response = await self._send_request( - ACPRequest(method="agent/list") - ) - return [AgentCard(**a) for a in response.result["agents"]] + /// 列出可用 Agent + pub async fn list_agents(&self) -> Result> { + let response = self.send_request(ACPRequest { + method: "agent/list".to_string(), + params: json!({}), + }).await?; + + let agents = response.result["agents"] + .as_array() + .ok_or_else(|| ACPError::InvalidResponse("agents".to_string()))?; + + agents.iter() + .map(|a| serde_json::from_value(a.clone()).map_err(|e| e.into())) + .collect() + } +} ``` ### 1.5 Skills 执行 -```python -class SkillExecutor: - """Skill 执行器""" +```rust +pub struct SkillExecutor { + registry: SkillRegistry, +} - def __init__(self, skill_registry: SkillRegistry): - self.registry = skill_registry +impl SkillExecutor { + pub fn new(registry: SkillRegistry) -> Self { + Self { registry } + } - async def execute( - self, - skill_name: str, - input_data: dict, - context: AgentContext, - ) -> SkillResult: - """执行 Skill""" + /// 执行 Skill + pub async fn execute( + &self, + skill_name: &str, + input_data: Value, + context: &mut AgentContext, + ) -> Result { + let skill = self.registry.get(skill_name) + .ok_or_else(|| SkillError::NotFound(format!("Skill not found: {}", skill_name)))?; - skill = self.registry.get(skill_name) - if not skill: - raise SkillNotFoundError(f"Skill not found: {skill_name}") + // Skill 可以包含多个步骤 + let steps = skill.get_steps(); + let mut results = Vec::new(); - # Skill 可以包含多个步骤 - steps = skill.get_steps() + for step in steps { + // 每个步骤可以是工具调用或 Agent 调用 + let result = match step.step_type.as_str() { + "tool" => self.call_tool(&step.tool, &step.args).await, + "agent" => self.call_agent(&step.agent, &step.action, &step.args).await, + "llm" => self.call_llm(&step.prompt, context).await, + _ => Err(SkillError::InvalidStep(step.step_type.clone()).into()), + }?; - results = [] - for step in steps: - # 每个步骤可以是工具调用或 Agent 调用 - if step.type == "tool": - result = await self._call_tool(step.tool, step.args) - elif step.type == "agent": - result = await self._call_agent(step.agent, step.action, step.args) - elif step.type == "llm": - result = await self._call_llm(step.prompt, context) + results.push(result); - results.append(result) + // 检查是否需要停止 + if step.on_result == "stop_if_success" && results.last().map(|r| r.success).unwrap_or(false) { + break; + } + } - # 检查是否需要停止 - if step.on_result == "stop_if_success" and result.success: - break - - return SkillResult( - skill_name=skill_name, - steps=results, - final_output=results[-1] if results else None, - ) + Ok(SkillResult { + skill_name: skill_name.to_string(), + steps: results.clone(), + final_output: results.last().cloned(), + }) + } +} ``` ### 1.6 配置 @@ -525,88 +548,122 @@ consume_interval = 1 / effective_rate ### 3.3 令牌桶实现 -```python -class TokenBucket: - """令牌桶流控""" +```rust +use std::sync::atomic::{AtomicFloat, Ordering}; +use std::sync::Arc; +use std::time::{Duration, Instant}; - def __init__( - self, - rate: float, # 每秒令牌数 - capacity: float, # 桶容量 - burst: float = None, # 突发容量 - ): - self.rate = rate - self.capacity = capacity - self.burst = burst or capacity - self.tokens = capacity - self.last_update = time.monotonic() +pub struct TokenBucket { + rate: f64, // 每秒令牌数 + capacity: f64, // 桶容量 + tokens: AtomicFloat, + last_update: std::sync::Mutex, +} - async def acquire(self, tokens: float = 1.0) -> float: - """获取令牌，返回需要等待的秒数""" - now = time.monotonic() - elapsed = now - self.last_update - self.tokens = min(self.capacity, self.tokens + elapsed * self.rate) - self.last_update = now +impl TokenBucket { + pub fn new(rate: f64, capacity: f64) -> Self { + Self { + rate, + capacity, + tokens: AtomicFloat::new(capacity), + last_update: std::sync::Mutex::new(Instant::now()), + } + } - if self.tokens >= tokens: - self.tokens -= tokens - return 0.0 + /// 获取令牌，返回需要等待的秒数 + pub fn acquire(&self, tokens: f64) -> f64 { + let mut last_update = self.last_update.lock().unwrap(); + let elapsed = last_update.elapsed().as_secs_f64(); + let current_tokens = self.tokens.load(Ordering::SeqCst); - wait_time = (tokens - self.tokens) / self.rate - return wait_time + let new_tokens = (current_tokens + elapsed * self.rate).min(self.capacity); + self.tokens.store(new_tokens, Ordering::SeqCst); + *last_update = Instant::now(); - async def wait_and_acquire(self, tokens: float = 1.0) -> None: - """等待直到获取令牌""" - wait = await self.acquire(tokens) - if wait > 0: - await asyncio.sleep(wait) + if new_tokens >= tokens { + self.tokens.fetch_sub(tokens as f32, Ordering::SeqCst); + 0.0 + } else { + (tokens - new_tokens) / self.rate + } + } + + /// 等待直到获取令牌 + pub async fn wait_and_acquire(&self, tokens: f64) { + let wait = self.acquire(tokens); + if wait > 0.0 { + tokio::time::sleep(Duration::from_secs_f64(wait)).await; + } + } +} ``` ### 3.4 优先级调度 -```python -class PriorityScheduler: - """优先级调度器""" +```rust +use std::collections::HashMap; +use std::sync::Arc; - def __init__(self, buckets: dict[str, TokenBucket]): - self.buckets = buckets # per-user or per-session +pub struct PriorityScheduler { + buckets: HashMap>, + queues: HashMap>>, +} - async def next_message(self) -> InputMessage | None: - """获取下一条待处理消息（按优先级）""" - # 1. 收集所有非空队列 - candidates = [] - for user_id, queue in self.queues.items(): - if not queue.messages: - continue +impl PriorityScheduler { + pub fn new() -> Self { + Self { + buckets: HashMap::new(), + queues: HashMap::new(), + } + } - # 2. 计算该用户的可用速率 - bucket = self.buckets.get(user_id) - if not bucket: - continue + /// 获取下一条待处理消息（按优先级） + pub async fn next_message(&self) -> Option { + let mut candidates = Vec::new(); - # 3. 获取队首消息（peek，不移除） - msg = queue.messages[0] + // 1. 收集所有非空队列 + for (user_id, queue) in &self.queues { + let queue = queue.lock().await; + if queue.is_empty() { + continue; + } - candidates.append((msg, bucket, user_id)) + // 2. 计算该用户的可用速率 + let Some(bucket) = self.buckets.get(user_id) else { + continue; + }; - if not candidates: - return None + // 3. 获取队首消息（peek，不移除） + let msg = queue.messages.front()?.clone(); + candidates.push((msg, Arc::clone(bucket), user_id.clone())); + } - # 4. 按优先级 + 可用性排序 - # 优先级相同时，优先处理令牌充足的 - candidates.sort( - key=lambda x: ( - -x[0].priority, - x[1].tokens / x[1].rate if x[1].rate > 0 else 0 - ) - ) + if candidates.is_empty() { + return None; + } - # 5. 等待最紧急消息的令牌 - msg, bucket, user_id = candidates[0] - await bucket.wait_and_acquire(1.0) + // 4. 按优先级 + 可用性排序 + // 优先级相同时，优先处理令牌充足的 + candidates.sort_by(|a, b| { + let a_priority = a.0.priority; + let b_priority = b.0.priority; + let a_tokens = a.1.tokens.load(Ordering::SeqCst); + let b_tokens = b.1.tokens.load(Ordering::SeqCst); + let a_score = (a_priority as f64, a_tokens); + let b_score = (b_priority as f64, b_tokens); + b_score.partial_cmp(&a_score).unwrap_or(std::cmp::Ordering::Equal) + }); - # 6. 移除并返回 - return queue.messages.popleft() + // 5. 等待最紧急消息的令牌 + let (_msg, bucket, user_id) = &candidates[0]; + bucket.wait_and_acquire(1.0).await; + + // 6. 移除并返回 + let queue = self.queues.get(user_id)?; + let mut queue = queue.lock().await; + queue.pop() + } +} ``` --- @@ -615,79 +672,112 @@ class PriorityScheduler: ### 4.1 上下文管理（Context Manager） -```python -@dataclass -class AgentContext: - """Agent 执行上下文""" - messages: list[Message] # 消息历史 - system_prompt: str # 系统提示 - tools: list[ToolDefinition] # 可用工具 - memory: MemoryBank # 记忆存储 - metadata: dict # 扩展元数据 +```rust +use serde::{Deserialize, Serialize}; +use std::sync::Arc; -class ContextManager: - """上下文管理器""" +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct AgentContext { + /// 消息历史 + pub messages: Vec, + /// 系统提示 + pub system_prompt: String, + /// 可用工具 + pub tools: Vec, + /// 记忆存储 + pub memory: Arc, + /// 扩展元数据 + pub metadata: HashMap, +} - def __init__(self, config: ContextConfig): - self.max_tokens: int = config.max_context_tokens - self.compress_threshold: float = config.compress_threshold - self.keep_recent: int = config.keep_recent_messages +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Message { + pub role: String, + pub content: String, + #[serde(default)] + pub metadata: HashMap, +} - def build_context( - self, - queue: UserMessageQueue, - memory: MemoryBank, - ) -> AgentContext: - """构建 Agent 执行上下文""" +#[derive(Debug, Clone)] +pub struct ContextConfig { + pub max_context_tokens: usize, + pub compress_threshold: f64, + pub keep_recent_messages: usize, +} - # 1. 从队列获取消息 - raw_messages = list(queue.messages) +pub struct ContextManager { + config: ContextConfig, + tool_registry: Arc, +} - # 2. 应用安全过滤 - raw_messages = self.apply_security_filters(raw_messages) +impl ContextManager { + pub fn new(config: ContextConfig, tool_registry: Arc) -> Self { + Self { config, tool_registry } + } - # 3. 构建消息列表 - messages = self.build_message_list(raw_messages) + /// 构建 Agent 执行上下文 + pub async fn build_context( + &self, + queue: &UserMessageQueue, + memory: Arc, + ) -> Result { + // 1. 从队列获取消息 + let raw_messages: Vec = queue.messages.iter().cloned().collect(); - # 4. 检查是否需要压缩 - total_tokens = self.estimate_tokens(messages) + // 2. 应用安全过滤 + let filtered_messages = self.apply_security_filters(raw_messages)?; - if total_tokens > self.max_tokens * self.compress_threshold: - messages = self.compress_context(messages, memory) + // 3. 构建消息列表 + let messages = self.build_message_list(filtered_messages)?; - # 5. 添加系统提示 - system_prompt = self.build_system_prompt() + // 4. 检查是否需要压缩 + let total_tokens = self.estimate_tokens(&messages); + let messages = if total_tokens > self.config.max_context_tokens { + self.compress_context(messages, memory.clone()).await? + } else { + messages + }; - return AgentContext( - messages=messages, - system_prompt=system_prompt, - tools=self.get_available_tools(), - memory=memory, - ) + // 5. 添加系统提示 + let system_prompt = self.build_system_prompt()?; - def compress_context( - self, - messages: list[Message], - memory: MemoryBank, - ) -> list[Message]: - """压缩上下文""" + // 6. 获取可用工具 + let tools = self.tool_registry.list_tools().await?; - # 保留最近 N 条消息 - recent = messages[-self.keep_recent:] + Ok(AgentContext { + messages, + system_prompt, + tools, + memory, + metadata: HashMap::new(), + }) + } - # 提取历史消息进行压缩 - history = messages[:-self.keep_recent] + /// 压缩上下文 + async fn compress_context( + &self, + messages: Vec, + memory: Arc, + ) -> Result, ContextError> { + let keep = self.config.keep_recent_messages; - # 摘要历史消息并存入记忆 - if history: - summary = self.summarize(history) - memory.add(Message( - role="system", - content=f"[历史摘要] {summary}", - metadata={"type": "summary"} - )) + // 保留最近 N 条消息 + let recent: Vec = messages.into_iter().rev().take(keep).collect(); + let history: Vec = messages.into_iter().rev().skip(keep).collect(); - return recent + // 摘要历史消息并存入记忆 + if !history.is_empty() { + let summary = self.summarize(&history)?; + memory.add(Message { + role: "system".into(), + content: format!("[历史摘要] {}", summary), + metadata: HashMap::from([("type".into(), "summary".into())]), + }).await?; + } + + Ok(recent) + } +} ``` ### 4.2 上下文配置 @@ -727,7 +817,7 @@ context: ## 5. 工具调用策略（Tool Calling Strategy） -### 4.1 工具调用最佳实践 +### 5.1 工具调用最佳实践 ```yaml # agent.yaml @@ -757,148 +847,223 @@ tool_calling: stream_intermediate: true ``` -### 4.2 工具调用流程 +### 5.2 工具调用流程 -```python -class ToolCallingPolicy: - """工具调用策略""" +```rust +use async_trait::async_trait; - async def execute_tools( - self, - llm_response: LLMResponse, - context: AgentContext, - ) -> list[ToolResult]: - """执行工具调用""" +#[derive(Debug, Clone)] +pub struct ToolCall { + pub id: String, + pub name: String, + pub arguments: HashMap, +} - # 1. 解析工具调用请求 - tool_calls = llm_response.tool_calls or [] +#[derive(Debug)] +pub struct ToolResult { + pub id: String, + pub name: String, + pub result: Result, +} - if not tool_calls: - return [] +#[derive(Debug, thiserror::Error)] +pub enum ToolError { + #[error("Tool not found: {0}")] + NotFound(String), + #[error("Execution failed: {0}")] + ExecutionFailed(String), + #[error("Timeout")] + Timeout, +} - # 2. 按策略分组 - groups = self._group_by_dependency(tool_calls) +pub struct ToolCallingPolicy { + config: ToolCallingConfig, + tool_executor: Arc, +} - results = [] +impl ToolCallingPolicy { + /// 执行工具调用 + pub async fn execute_tools( + &self, + llm_response: &LLMResponse, + context: &AgentContext, + ) -> Result, ToolError> { + // 1. 解析工具调用请求 + let tool_calls = &llm_response.tool_calls; - # 3. 按组执行 - for group in groups: - if self._can_parallel(group): - # 并行执行 - group_results = await asyncio.gather( - *[self._execute_single(call, context) for call in group], - return_exceptions=True - ) - else: - # 串行执行 - group_results = [] - for call in group: - result = await self._execute_single(call, context) - group_results.append(result) + if tool_calls.is_empty() { + return Ok(Vec::new()); + } - results.extend(group_results) + // 2. 按策略分组 + let groups = self.group_by_dependency(tool_calls); - # 4. 检查是否超过限制 - if len(results) >= self.config.max_calls_per_request: - break + let mut results = Vec::new(); - # 5. 如果需要流式中间结果 - if self.config.stream_intermediate: - yield ToolCallingEvent( - type="intermediate", - results=group_results - ) + // 3. 按组执行 + for group in groups { + letannels = if self.can_parallel(&group) { + // 并行执行 + self.execute_parallel(group, context).await? + } else { + // 串行执行 + self.execute_sequential(group, context).await? + }; - return results + results.extend(group_results); - def _group_by_dependency( - self, - tool_calls: list[ToolCall], - ) -> list[list[ToolCall]]: - """按依赖关系分组""" + // 4. 检查是否超过限制 + if results.len() >= self.config.max_calls_per_request { + break; + } + } - groups = [] - current_group = [] + Ok(results) + } - for call in tool_calls: - # 检查是否依赖前一个工具的结果 - if call.arguments_depends_on_previous and current_group: - # 依赖：将当前调用加入前一个组 - current_group.append(call) - else: - # 不依赖：开启新组 - if current_group: - groups.append(current_group) - current_group = [call] + /// 按依赖关系分组 + fn group_by_dependency(&self, tool_calls: &[ToolCall]) -> Vec> { + let mut groups = Vec::new(); + let mut current_group = Vec::new(); - if current_group: - groups.append(current_group) + for call in tool_calls { + // 检查是否依赖前一个工具的结果 + if !current_group.is_empty() && call.depends_on_previous { + current_group.push(call.clone()); + } else { + if !current_group.is_empty() { + groups.push(current_group); + } + current_group = vec![call.clone()]; + } + } - return groups + if !current_group.is_empty() { + groups.push(current_group); + } + + groups + } + + /// 检查是否可以并行执行 + fn can_parallel(&self, group: &[ToolCall]) -> bool { + self.config.parallel_calls && group.iter().all(|c| !c.depends_on_previous) + } + + /// 并行执行 + async fn execute_parallel( + &self, + calls: Vec, + context: &AgentContext, + ) -> Result, ToolError> { + let futures = calls.into_iter().map(|call| { + self.execute_single(call, context) + }); + + let results = futures::future::join_all(futures).await; + + Ok(results.into_iter().map(|r| r.unwrap()).collect()) + } + + /// 串行执行 + async fn execute_sequential( + &self, + calls: Vec, + context: &AgentContext, + ) -> Result, ToolError> { + let mut results = Vec::new(); + + for call in calls { + let result = self.execute_single(call, context).await?; + results.push(result); + } + + Ok(results) + } +} ``` -### 4.3 工具选择策略 +### 5.3 工具选择策略 -```python -class ToolSelector: - """工具选择策略""" +```rust +pub struct ToolSelector { + max_tools_per_request: usize, + prefer_recent: bool, +} - def __init__(self, config: ToolSelectionConfig): - self.max_tools_per_request = config.max_tools_per_request - self.prefer_recent = config.prefer_recent_tools +impl ToolSelector { + pub fn new(max_tools_per_request: usize, prefer_recent: bool) -> Self { + Self { + max_tools_per_request, + prefer_recent, + } + } - def select_tools( - self, - available_tools: list[Tool], - query: str, - context: AgentContext, - ) -> list[Tool]: - """选择最相关的工具""" + /// 选择最相关的工具 + pub fn select_tools( + &self, + available_tools: &[Tool], + query: &str, + context: &AgentContext, + ) -> Vec { + // 1. 计算工具与查询的相关性 + let mut scored: Vec<(f64, Tool)> = available_tools + .iter() + .map(|tool| (self.calculate_relevance(tool, query, context), tool.clone())) + .collect(); - # 1. 计算工具与查询的相关性 - scored = [] - for tool in available_tools: - score = self._calculate_relevance(tool, query, context) - scored.append((score, tool)) + // 2. 排序并截取 + scored.sort_by(|a, b| b.0.partial_cmp(&a.0).unwrap_or(std::cmp::Ordering::Equal)); + let selected: Vec = scored.into_iter().take(self.max_tools_per_request).map(|(_, t)| t).collect(); - # 2. 排序并截取 - scored.sort(key=lambda x: -x[0]) - selected = scored[:self.max_tools_per_request] + // 3. 如果启用了最近使用优先 + if self.prefer_recent { + self.boost_recent(selected, context) + } else { + selected + } + } - # 3. 如果启用了最近使用优先 - if self.prefer_recent: - selected = self._boost_recent(selected, context) + /// 计算相关性分数 + fn calculate_relevance(&self, tool: &Tool, query: &str, context: &AgentContext) -> f64 { + let mut score = 0.0; - return [t for _, t in selected] + // 工具名称匹配 + if query.to_lowercase().split_whitespace().any(|w| tool.name.to_lowercase().contains(w)) { + score += 0.3; + } - def _calculate_relevance( - self, - tool: Tool, - query: str, - context: AgentContext, - ) -> float: - """计算相关性分数""" + // 工具描述匹配 + if !tool.description.is_empty() { + let query_words: std::collections::HashSet<&str> = + query.to_lowercase().split_whitespace().collect(); + let desc_words: std::collections::HashSet<&str> = + tool.description.to_lowercase().split_whitespace().collect(); + let overlap = query_words.intersection(&desc_words).count(); + score += overlap as f64 * 0.1; + } - base_score = 0.0 + // 最近使用过的工具加权 + if let Some(recent_tools) = context.metadata.get("recent_tools") { + if recent_tools.contains(&tool.name) { + score += 0.2; + } + } - # 工具名称匹配 - if any(word in tool.name.lower() for word in query.lower().split()): - base_score += 0.3 + score + } - # 工具描述匹配 - if tool.description: - # 简单的词重叠计算 - query_words = set(query.lower().split()) - desc_words = set(tool.description.lower().split()) - overlap = len(query_words & desc_words) - base_score += overlap * 0.1 - - # 最近使用过的工具加权 - if context.metadata.get("recent_tools"): - if tool.name in context.metadata["recent_tools"]: - base_score += 0.2 - - return base_score + /// 最近使用优先 + fn boost_recent(&self, mut tools: Vec, context: &AgentContext) -> Vec { + let recent_tools = context.metadata.get("recent_tools"); + tools.sort_by(|a, b| { + let a_recent = recent_tools.map(|r| r.contains(&a.name)).unwrap_or(false); + let b_recent = recent_tools.map(|r| r.contains(&b.name)).unwrap_or(false); + b_recent.cmp(&a_recent) + }); + tools + } +} ``` --- @@ -973,105 +1138,138 @@ security: ### 6.2 安全过滤器实现 -```python -class SecurityFilter: - """安全过滤器""" +```rust +use regex::Regex; +use std::collections::HashMap; - def __init__(self, config: SecurityConfig): - self.config = config - self.compiled_patterns = [ - (p["name"], re.compile(p["regex"]), p["severity"]) - for p in config.injection.patterns - ] +#[derive(Debug, Clone)] +pub struct Detection { + pub name: String, + pub severity: String, + pub matched: Vec, +} - def filter_messages( - self, - messages: list[InputMessage], - ) -> list[InputMessage]: - """过滤输入消息""" +pub struct SecurityFilter { + config: SecurityConfig, + compiled_patterns: Vec<(String, Regex, String)>, +} - filtered = [] +impl SecurityFilter { + pub fn new(config: SecurityConfig) -> Result { + let compiled_patterns = config + .injection + .patterns + .iter() + .map(|p| { + let regex = Regex::new(&p.regex)?; + Ok((p.name.clone(), regex, p.severity.clone())) + }) + .collect::, _>>()?; - for msg in messages: - # 1. 内容过滤 - if self.config.content_filter.enable: - msg.content = self._filter_content(msg.content) + Ok(Self { config, compiled_patterns }) + } - # 2. 注入检测 - if self.config.injection.enable: - detections = self._detect_injection(msg.content) + /// 过滤输入消息 + pub fn filter_messages(&self, messages: Vec) -> Vec { + let mut filtered = Vec::new(); - if detections: - action = self._handle_injection(detections, msg) - if action == "skip": - continue + for mut msg in messages { + // 1. 内容过滤 + if self.config.content_filter.enable { + msg.content = self.filter_content(msg.content); + } - filtered.append(msg) + // 2. 注入检测 + if self.config.injection.enable { + let detections = self.detect_injection(&msg.content); - return filtered + if !detections.is_empty() { + let action = self.handle_injection(&detections, &mut msg); + if action == "skip" { + continue; + } + } + } - def filter_output( - self, - content: str, - context: AgentContext, - ) -> str: - """过滤输出内容""" + filtered.push(msg); + } - # 1. 泄密防护 - 移除敏感信息 - if self.config.leakage_prevention: - content = self._redact_sensitive(content) + filtered + } - return content + /// 过滤输出内容 + pub fn filter_output(&self, content: String, context: &AgentContext) -> String { + // 泄密防护 - 移除敏感信息 + if let Some(ref leakage) = self.config.leakage_prevention { + self.redact_sensitive(content, leakage) + } else { + content + } + } - def _detect_injection(self, content: str) -> list[Detection]: - """检测注入攻击""" - detections = [] + /// 检测注入攻击 + fn detect_injection(&self, content: &str) -> Vec { + let mut detections = Vec::new(); - for name, pattern, severity in self.compiled_patterns: - if pattern.search(content): - detections.append(Detection( - name=name, - severity=severity, - matched=pattern.findall(content), - )) + for (name, pattern, severity) in &self.compiled_patterns { + if let Some(matched) = pattern.find(content) { + detections.push(Detection { + name: name.clone(), + severity: severity.clone(), + matched: pattern.captures_iter(content).map(|c| c[0].to_string()).collect(), + }); + } + } - return detections + detections + } - def _handle_injection( - self, - detections: list[Detection], - message: InputMessage, - ) -> str: - """处理注入检测""" + /// 处理注入检测 + fn handle_injection(&self, detections: &[Detection], message: &mut InputMessage) -> &str { + let high_severity = detections.iter().any(|d| d.severity == "high"); - high_severity = any(d.severity == "high" for d in detections) + if high_severity && self.config.injection.on_detect == "block" { + tracing::warn!("Blocked injection: {:?}", detections); + return "skip"; + } - if high_severity and self.config.injection.on_detect == "block": - # 记录并阻止 - logging.warning(f"Blocked injection: {detections}") - return "skip" + if self.config.injection.on_detect == "sanitize" { + // 消毒处理 + for detection in detections { + message.content = self.filter_content(message.content.clone()); + } + return "sanitize"; + } - elif self.config.injection.on_detect == "sanitize": - # 消毒处理 - for detection in detections: - message.content = message.content.replace( - detection.matched, - self.config.content_filter.replacement, - ) - return "sanitize" + "allow" + } - return "allow" + /// 内容过滤 + fn filter_content(&self, content: String) -> String { + if !self.config.content_filter.enable { + return content; + } - def _filter_content(self, content: str) -> str: - """内容过滤""" + let mut result = content; + for pattern in &self.config.content_filter.blocklist { + if let Ok(regex) = Regex::new(pattern) { + result = regex.replace_all(&result, self.config.content_filter.replacement.as_str()).to_string(); + } + } + result + } - if not self.config.content_filter.enable: - return content - - for pattern in self.config.content_filter.blocklist: - content = re.sub(pattern, self.config.content_filter.replacement, content) - - return content + /// 移除敏感信息 + fn redact_sensitive(&self, content: String, leakage: &LeakagePrevention) -> String { + let mut result = content; + for pattern in &leakage.blocked_output_patterns { + if let Ok(regex) = Regex::new(pattern) { + result = regex.replace_all(&result, leakage.placeholder.as_str()).to_string(); + } + } + result + } +} ``` --- @@ -1475,33 +1673,75 @@ pub struct PermissionAuditLog { ### 8.1 队列结构 -```python -@dataclass -class OutputMessage: - """输出消息单元""" - session_id: str - content: str | AsyncIterator[str] # 支持流式 - format: str = "plain" # plain | markdown | html - strategy: OutputStrategy = OutputStrategy.FULL - metadata: dict = field(default_factory=dict) +```rust +use serde::{Deserialize, Serialize}; +use std::collections::VecDeque; +use tokio::sync::mpsc; - # 流式相关 - stream_start_time: float | None = None - total_sent: int = 0 +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct OutputMessage { + pub session_id: String, + pub content: OutputContent, + pub format: OutputFormat, + pub strategy: OutputStrategy, + #[serde(default)] + pub metadata: HashMap, +} -@dataclass -class ResultQueue: - """结果队列""" - session_id: str - results: deque[OutputMessage] - max_size: int = 100 - allow_streaming: bool = True +#[derive(Debug, Clone, Serialize, Deserialize)] +pub enum OutputContent { + Text(String), + Stream(mpsc::Receiver), +} -class OutputStrategy(Enum): - """输出策略""" - STREAMING = "streaming" # 流式输出 - SEGMENTED = "segmented" # 智能分段 - FULL = "full" # 一次性输出 +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum OutputFormat { + Plain, + Markdown, + Html, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum OutputStrategy { + Streaming, + Segmented, + Full, +} + +pub struct ResultQueue { + pub session_id: String, + results: VecDeque, + max_size: usize, + allow_streaming: bool, +} + +impl ResultQueue { + pub fn new(session_id: String) -> Self { + Self { + session_id, + results: VecDeque::new(), + max_size: 100, + allow_streaming: true, + } + } + + pub fn push(&mut self, msg: OutputMessage) { + if self.results.len() >= self.max_size { + self.results.pop_front(); + } + self.results.push_back(msg); + } + + pub fn pop(&mut self) -> Option { + self.results.pop_front() + } + + pub fn len(&self) -> usize { + self.results.len() + } +} ``` ### 8.2 输出策略 @@ -1575,136 +1815,200 @@ output: ### 8.3 分段器实现 -```python -class SmartSegmenter: - """智能分段器""" +```rust +use regex::Regex; +use std::time::Duration; - def __init__(self, config: SegmentedConfig): - self.config = config +pub struct SmartSegmenter { + config: SegmentedConfig, +} - def segment(self, content: str) -> list[str]: - """将内容分段""" +impl SmartSegmenter { + pub fn new(config: SegmentedConfig) -> Self { + Self { config } + } - if len(content) < self.config.threshold: - return [content] + /// 将内容分段 + pub fn segment(&self, content: &str) -> Vec { + if content.len() < self.config.threshold { + return vec![content.to_string()]; + } - if self.config.mode == "sentence": - return self._split_by_sentence(content) - elif self.config.mode == "word_count": - return self._split_by_word_count(content) - elif self.config.mode == "regex": - return self._split_by_regex(content) + match self.config.mode.as_str() { + "sentence" => self.split_by_sentence(content), + "word_count" => self.split_by_word_count(content), + "regex" => self.split_by_regex(content), + _ => vec![content.to_string()], + } + } - return [content] + /// 按句子分段 + fn split_by_sentence(&self, content: &str) -> Vec { + let regex = Regex::new(&self.config.split_regex).unwrap_or_else(|_| Regex::new("").unwrap()); + let sentences: Vec<&str> = regex.split(content).collect(); - def _split_by_sentence(self, content: str) -> list[str]: - """按句子分段""" - sentences = re.split( - self.config.split_regex, - content, - ) + let mut segments = Vec::new(); + let mut current = Vec::new(); + let mut current_len = 0; - segments = [] - current = [] + for sentence in sentences { + if sentence.trim().is_empty() { + continue; + } - for sentence in sentences: - if not sentence.strip(): - continue + current.push(sentence); + current_len += sentence.len(); - current.append(sentence) - current_text = "".join(current) + if current_len >= self.config.threshold { + let segment = current.join(""); + let segment = if self.config.add_ellipsis && !segments.is_empty() { + format!("...{}", segment.trim_start()) + } else { + segment + }; + segments.push(segment); + current.clear(); + current_len = 0; + } + } - # 如果当前段落达到阈值 - if len(current_text) >= self.config.threshold: - segment = "".join(current) - if self.config.add_ellipsis and len(segments) > 0: - segment = "..." + segment.lstrip() - segments.append(segment) - current = [] + // 处理剩余内容 + if !current.is_empty() { + let remaining = current.join(""); + if !remaining.trim().is_empty() { + let remaining = if self.config.add_ellipsis && !segments.is_empty() { + format!("...{}", remaining.trim_start()) + } else { + remaining + }; + segments.push(remaining); + } + } - # 处理剩余内容 - if current: - remaining = "".join(current) - if remaining.strip(): - if self.config.add_ellipsis and segments: - remaining = "..." + remaining.lstrip() - segments.append(remaining) + segments + } - return segments + /// 按字数分段 + fn split_by_word_count(&self, content: &str) -> Vec { + let chars: Vec = content.chars().collect(); + let mut segments = Vec::new(); + let mut current = String::new(); - async def stream_segments( - self, - content: str, - output: OutputMessage, - sender: Callable[[str], Awaitable[None]], - ) -> None: - """流式发送分段""" + for c in chars { + current.push(c); + if current.len() >= self.config.threshold { + segments.push(current.clone()); + current.clear(); + } + } - segments = self.segment(content) + if !current.is_empty() { + segments.push(current); + } - for i, segment in enumerate(segments): - # 发送当前分段 - await sender(segment) + segments + } - # 添加间隔（随机） - if i < len(segments) - 1: - interval = self._random_interval() - await asyncio.sleep(interval) + /// 按正则分段 + fn split_by_regex(&self, content: &str) -> Vec { + let regex = Regex::new(&self.config.split_regex).unwrap_or_else(|_| Regex::new("").unwrap()); + regex.split(content).map(|s| s.to_string()).collect() + } - def _random_interval(self) -> float: - """生成随机间隔""" - import random - parts = self.config.random_interval.split(",") - return random.uniform(float(parts[0]), float(parts[1])) + /// 生成随机间隔 + fn random_interval(&self) -> Duration { + let parts: Vec = self.config + .random_interval + .split(',') + .filter_map(|s| s.trim().parse().ok()) + .collect(); + + if parts.len() >= 2 { + let min = parts[0]; + let max = parts[1]; + let duration = min + (max - min) * rand::random::(); + Duration::from_secs_f64(duration) + } else { + Duration::from_millis(500) + } + } +} ``` ### 8.4 流式输出器 -```python -class StreamingOutput: - """流式输出器""" +```rust +pub struct StreamingOutput { + config: StreamingConfig, +} - def __init__(self, config: StreamingConfig): - self.config = config +impl StreamingOutput { + pub fn new(config: StreamingConfig) -> Self { + Self { config } + } - async def stream( - self, - content: str, - sender: Callable[[str], Awaitable[None]], - ) -> None: - """流式输出内容""" + /// 流式输出内容 + pub async fn stream(&self, content: &str, mut sender: F) + where + F: FnMut(String) -> Fut, + Fut: Future, + { + let mut start = 0; + let bytes = content.as_bytes(); - start = 0 - while start < len(content): - end = start + self.config.chunk_size - chunk = content[start:end] + while start < bytes.len() { + let end = (start + self.config.chunk_size).min(bytes.len()); + let chunk = String::from_utf8_lossy(&bytes[start..end]).to_string(); - await sender(chunk) + sender(chunk).await; + start = end; - start = end + // 添加短暂间隔 + if start < bytes.len() { + tokio::time::sleep(self.config.chunk_interval).await; + } + } + } - # 添加短暂间隔 - if start < len(content): - await asyncio.sleep(self.config.chunk_interval) + /// 创建流式迭代器 + pub fn create_stream(&self, content: String) -> impl Stream { + struct StreamIter { + content: String, + chunk_size: usize, + chunk_interval: Duration, + current: usize, + } - def create_stream( - self, - content: str, - ) -> AsyncIterator[str]: - """创建流式迭代器""" + impl Stream for StreamIter { + type Item = String; - async def generator(): - start = 0 - while start < len(content): - end = start + self.config.chunk_size - chunk = content[start:end] - yield chunk - start = end + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let this = &mut *self; - if start < len(content): - await asyncio.sleep(self.config.chunk_interval) + if this.current >= this.content.len() { + return Poll::Ready(None); + } - return generator() + let end = (this.current + this.chunk_size).min(this.content.len()); + let chunk = this.content[this.current..end].to_string(); + this.current = end; + + // Schedule next chunk after interval + let interval = this.chunk_interval; + let _ = cx; // suppress unused warning + + Poll::Ready(Some(chunk)) + } + } + + StreamIter { + content, + chunk_size: self.config.chunk_size, + chunk_interval: self.config.chunk_interval, + current: 0, + } + } +} ``` --- @@ -1755,131 +2059,186 @@ memory: ### 9.2 记忆类型 -```python -class MemoryType(Enum): - """记忆类型""" - WORKING = "working" # 工作记忆（当前会话） - EPISODIC = "episodic" # 情景记忆（历史事件） - SEMANTIC = "semantic" # 语义记忆（持久知识） +```rust +#[derive(Debug, Clone, Serialize, Deserialize)] +pub enum MemoryType { + Working, // 工作记忆（当前会话） + Episodic, // 情景记忆（历史事件） + Semantic, // 语义记忆（持久知识） +} -@dataclass -class MemoryEntry: - """记忆条目""" - id: str - type: MemoryType - content: str - embedding: list[float] | None = None - metadata: dict = field(default_factory=dict) - created_at: float - updated_at: float - access_count: int = 0 - importance: float = 0.5 # 0-1 重要性评分 +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct MemoryEntry { + pub id: String, + #[serde(rename = "type")] + pub memory_type: MemoryType, + pub content: String, + pub embedding: Option>, + pub metadata: HashMap, + pub created_at: f64, + pub updated_at: f64, + pub access_count: u32, + pub importance: f32, // 0-1 重要性评分 +} -class MemoryBank: - """记忆库""" +impl MemoryEntry { + pub fn new(memory_type: MemoryType, content: String) -> Self { + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs_f64(); - def __init__(self, config: MemoryConfig): - self.config = config - self.backend = self._create_backend(config) - self._cache: dict[str, MemoryEntry] = {} - self._cache_max_size = 100 + Self { + id: uuid::Uuid::new_v4().to_string(), + memory_type, + content, + embedding: None, + metadata: HashMap::new(), + created_at: now, + updated_at: now, + access_count: 0, + importance: 0.5, + } + } +} - async def add(self, message: Message) -> None: - """添加记忆""" - entry = MemoryEntry( - id=str(uuid.uuid4()), - type=MemoryType.EPISODIC, - content=message.content, - metadata={ - "role": message.role, - "user_id": message.metadata.get("user_id"), - "session_id": message.metadata.get("session_id"), +pub struct MemoryBank { + config: MemoryConfig, + backend: Box, + cache: HashMap>, + cache_max_size: usize, +} + +impl MemoryBank { + pub fn new(config: MemoryConfig, backend: Box) -> Self { + Self { + config, + backend, + cache: HashMap::new(), + cache_max_size: 100, + } + } + + /// 添加记忆 + pub async fn add(&mut self, message: &Message) -> Result<()> { + let entry = MemoryEntry { + id: uuid::Uuid::new_v4().to_string(), + memory_type: MemoryType::Episodic, + content: message.content.clone(), + metadata: { + let mut m = HashMap::new(); + m.insert("role".to_string(), message.role.clone()); + if let Some(user_id) = message.metadata.get("user_id") { + m.insert("user_id".to_string(), user_id.clone()); + } + if let Some(session_id) = message.metadata.get("session_id") { + m.insert("session_id".to_string(), session_id.clone()); + } + m }, - created_at=time.time(), - updated_at=time.time(), - ) + created_at: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs_f64(), + updated_at: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs_f64(), + access_count: 0, + importance: 0.5, + }; - await self.backend.save(entry) + self.backend.save(entry).await + } - async def search( - self, - query: str, - limit: int = 5, - memory_types: list[MemoryType] | None = None, - ) -> list[MemoryEntry]: - """搜索记忆""" + /// 搜索记忆 + pub async fn search( + &mut self, + query: &str, + limit: usize, + memory_types: Option>, + ) -> Result> { + // 1. 如果有缓存，直接返回 + let cache_key = format!("{}:{}", query, limit); + if let Some(cached) = self.cache.get(&cache_key) { + return Ok(cached.clone()); + } - # 1. 如果有缓存，直接返回 - cache_key = f"{query}:{limit}" - if cache_key in self._cache: - return self._cache[cache_key] + // 2. 向量搜索 + let mut results = self.backend.search(query, limit, memory_types).await?; - # 2. 向量搜索 - results = await self.backend.search( - query=query, - limit=limit, - memory_types=memory_types, - ) + // 3. 更新访问计数 + for entry in &mut results { + entry.access_count += 1; + let _ = self.backend.update(entry.clone()).await; + } - # 3. 更新访问计数 - for entry in results: - entry.access_count += 1 - await self.backend.update(entry) + // 4. 缓存 (LRU淘汰) + if self.cache.len() >= self.cache_max_size { + if let Some((key, _)) = self.cache.iter() + .min_by(|(_, a), (_, b)| a[0].access_count.cmp(&b[0].access_count)) + { + self.cache.remove(key); + } + } - # 4. 缓存 - if len(self._cache) >= self._cache_max_size: - # LRU 淘汰 - oldest = min(self._cache.values(), key=lambda x: x.access_count) - del self._cache[oldest.id] + self.cache.insert(cache_key, results.clone()); + Ok(results) + } - self._cache[cache_key] = results + /// 摘要旧记忆 + pub async fn summarize_old(&mut self, before_timestamp: f64) -> Result { + // 1. 获取指定时间前的记忆 + let entries = self.backend.get_before(before_timestamp).await?; - return results + if entries.is_empty() { + return Ok(String::new()); + } - async def summarize_old( - self, - before_timestamp: float, - ) -> str: - """摘要旧记忆""" + // 2. 构建摘要 + let summary_prompt = format!( + "请简洁总结以下对话要点：\n\n{}\n\n保留关键信息：\n- 主要话题或问题\n- 已确定的结论或方案\n- 未完成的任务", + entries.iter() + .map(|e| format!("- {}", e.content)) + .collect::>() + .join("\n") + ); - # 1. 获取指定时间前的记忆 - entries = await self.backend.get_before(before_timestamp) + // 3. 调用 LLM 摘要 + let summary = self.llm_summarize(&summary_prompt).await?; - if not entries: - return "" + // 4. 创建摘要记忆 + let summary_entry = MemoryEntry { + id: uuid::Uuid::new_v4().to_string(), + memory_type: MemoryType::Semantic, + content: summary.clone(), + metadata: { + let mut m = HashMap::new(); + m.insert("original_entries".to_string(), entries.len().to_string()); + m + }, + created_at: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs_f64(), + updated_at: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs_f64(), + access_count: 0, + importance: 0.7, + }; - # 2. 构建摘要 - summary_prompt = f"""请简洁总结以下对话要点： + self.backend.save(summary_entry).await?; -{chr(10).join(f"- {e.content}" for e in entries)} + // 5. 删除原始记忆 + for entry in &entries { + let _ = self.backend.delete(&entry.id).await; + } -保留关键信息： -- 主要话题或问题 -- 已确定的结论或方案 -- 未完成的任务 -""" - - # 3. 调用 LLM 摘要 - summary = await self._llm_summarize(summary_prompt) - - # 4. 创建摘要记忆 - summary_entry = MemoryEntry( - id=str(uuid.uuid4()), - type=MemoryType.SEMANTIC, - content=summary, - metadata={"original_entries": len(entries)}, - created_at=time.time(), - updated_at=time.time(), - importance=0.7, - ) - - await self.backend.save(summary_entry) - - # 5. 删除原始记忆 - for entry in entries: - await self.backend.delete(entry.id) - - return summary + Ok(summary) + } +} ``` --- @@ -1888,91 +2247,138 @@ class MemoryBank: ### 10.1 平台特性 -```python -@dataclass -class PlatformCapabilities: - """平台能力""" - supports_streaming: bool = False - max_message_length: int = 4096 - supports_markdown: bool = True - supports_html: bool = False - supports_images: bool = True - supports_mentions: bool = True - supports_reply: bool = True - rate_limit_rpm: int = 60 - rate_limit_rpd: int = 10000 - -PLATFORM_CAPABILITIES = { - "telegram": PlatformCapabilities( - supports_streaming=False, - max_message_length=4096, - supports_markdown=True, - supports_html=True, - ), - "discord": PlatformCapabilities( - supports_streaming=False, - max_message_length=2000, - supports_markdown=True, - supports_html=False, - supports_reply=True, - ), - "qq": PlatformCapabilities( - supports_streaming=False, - max_message_length=500, - supports_markdown=False, - supports_mentions=True, - ), - "webchat": PlatformCapabilities( - supports_streaming=True, - max_message_length=10000, - supports_markdown=True, - supports_html=True, - ), +```rust +#[derive(Debug, Clone)] +pub struct PlatformCapabilities { + pub supports_streaming: bool, + pub max_message_length: usize, + pub supports_markdown: bool, + pub supports_html: bool, + pub supports_images: bool, + pub supports_mentions: bool, + pub supports_reply: bool, + pub rate_limit_rpm: u32, + pub rate_limit_rpd: u32, } + +impl Default for PlatformCapabilities { + fn default() -> Self { + Self { + supports_streaming: false, + max_message_length: 4096, + supports_markdown: true, + supports_html: false, + supports_images: true, + supports_mentions: true, + supports_reply: true, + rate_limit_rpm: 60, + rate_limit_rpd: 10000, + } + } +} + +pub static PLATFORM_CAPABILITIES: Lazy> = + Lazy::new(|| { + let mut m = HashMap::new(); + m.insert("telegram", PlatformCapabilities { + supports_streaming: false, + max_message_length: 4096, + supports_markdown: true, + supports_html: true, + ..Default::default() + }); + m.insert("discord", PlatformCapabilities { + supports_streaming: false, + max_message_length: 2000, + supports_markdown: true, + supports_html: false, + supports_reply: true, + ..Default::default() + }); + m.insert("qq", PlatformCapabilities { + supports_streaming: false, + max_message_length: 500, + supports_markdown: false, + supports_mentions: true, + ..Default::default() + }); + m.insert("webchat", PlatformCapabilities { + supports_streaming: true, + max_message_length: 10000, + supports_markdown: true, + supports_html: true, + ..Default::default() + }); + m + }); ``` ### 10.2 策略选择器 -```python -class PlatformStrategySelector: - """平台策略选择器""" +```rust +pub struct PlatformStrategySelector { + config: PlatformAdaptationConfig, + capabilities: &'static HashMap<&'static str, PlatformCapabilities>, +} - def __init__(self, config: PlatformAdaptationConfig): - self.config = config - self.capabilities = PLATFORM_CAPABILITIES +impl PlatformStrategySelector { + pub fn new(config: PlatformAdaptationConfig) -> Self { + Self { + config, + capabilities: &PLATFORM_CAPABILITIES, + } + } - def select_strategy( - self, - platform: str, - content_length: int, - user_preference: str | None = None, - ) -> OutputStrategy: - """选择输出策略""" + /// 选择输出策略 + pub fn select_strategy( + &self, + platform: &str, + content_length: usize, + user_preference: Option<&str>, + ) -> OutputStrategy { + let caps = self.capabilities.get(platform); - caps = self.capabilities.get(platform) + // 1. 用户偏好优先 + if let Some(pref) = user_preference { + if self.is_valid_strategy(pref, caps) { + return OutputStrategy::from_str(pref); + } + } - # 1. 用户偏好优先 - if user_preference and self._is_valid_strategy(user_preference, caps): - return OutputStrategy(user_preference) + // 2. 平台能力判断 + let caps = match caps { + Some(c) => c, + None => return OutputStrategy::Full, + }; - # 2. 平台能力判断 - if not caps: - return OutputStrategy.FULL + // 3. 平台配置覆盖 + if let Some(platform_strategy) = self.config.strategy_by_platform.get(platform) { + return OutputStrategy::from_str(platform_strategy); + } - # 3. 平台配置覆盖 - platform_strategy = self.config.strategy_by_platform.get(platform) - if platform_strategy: - return OutputStrategy(platform_strategy) + // 4. 内容长度判断 + if content_length > caps.max_message_length { + return OutputStrategy::Segmented; + } - # 4. 内容长度判断 - if content_length > caps.max_message_length: - return OutputStrategy.SEGMENTED + // 5. 流式支持判断 + if caps.supports_streaming { + return OutputStrategy::Streaming; + } - # 5. 流式支持判断 - if caps.supports_streaming: - return OutputStrategy.STREAMING + OutputStrategy::Full + } - return OutputStrategy.FULL + fn is_valid_strategy(&self, strategy: &str, caps: Option<&PlatformCapabilities>) -> bool { + let strategy = OutputStrategy::from_str(strategy); + match (strategy, caps) { + (OutputStrategy::Streaming, Some(c)) => c.supports_streaming, + (OutputStrategy::Segmented, Some(_)) => true, + (OutputStrategy::Full, _) => true, + _ => false, + } + } +} ``` --- @@ -2074,83 +2480,144 @@ memory: ### 12.1 错误分类 -```python -class ErrorType(Enum): - """错误类型""" - RATE_LIMIT = "rate_limit" # 限流 - TIMEOUT = "timeout" # 超时 - NETWORK = "network" # 网络错误 - API = "api" # API 错误 - TOOL = "tool" # 工具错误 - SECURITY = "security" # 安全错误 - INTERNAL = "internal" # 内部错误 +```rust +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum ErrorType { + RateLimit, // 限流 + Timeout, // 超时 + Network, // 网络错误 + Api, // API 错误 + Tool, // 工具错误 + Security, // 安全错误 + Internal, // 内部错误 +} -@dataclass -class ErrorRecoveryConfig: - """错误恢复配置""" - max_retries: dict[ErrorType, int] = field(default_factory=lambda: { - ErrorType.RATE_LIMIT: 5, - ErrorType.TIMEOUT: 3, - ErrorType.NETWORK: 3, - ErrorType.API: 2, - ErrorType.TOOL: 2, - ErrorType.SECURITY: 0, - ErrorType.INTERNAL: 1, - }) +impl ErrorType { + pub fn as_str(&self) -> &'static str { + match self { + ErrorType::RateLimit => "rate_limit", + ErrorType::Timeout => "timeout", + ErrorType::Network => "network", + ErrorType::Api => "api", + ErrorType::Tool => "tool", + ErrorType::Security => "security", + ErrorType::Internal => "internal", + } + } +} - backoff_multiplier: float = 1.5 - max_backoff: float = 60.0 +#[derive(Debug, Clone)] +pub struct ErrorRecoveryConfig { + pub max_retries: HashMap, + pub backoff_multiplier: f64, + pub max_backoff: f64, +} + +impl Default for ErrorRecoveryConfig { + fn default() -> Self { + let mut max_retries = HashMap::new(); + max_retries.insert(ErrorType::RateLimit, 5); + max_retries.insert(ErrorType::Timeout, 3); + max_retries.insert(ErrorType::Network, 3); + max_retries.insert(ErrorType::Api, 2); + max_retries.insert(ErrorType::Tool, 2); + max_retries.insert(ErrorType::Security, 0); + max_retries.insert(ErrorType::Internal, 1); + + Self { + max_retries, + backoff_multiplier: 1.5, + max_backoff: 60.0, + } + } +} ``` ### 12.2 错误处理策略 -```python -async def handle_error( - error: Exception, - context: AgentContext, - config: ErrorRecoveryConfig, -) -> ErrorAction: - """处理错误并决定下一步行动""" +```rust +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ErrorAction { + Retry, + Fail, + Block, + Fallback, +} - error_type = classify_error(error) - retries = context.metadata.get(f"retry_{error_type.value}", 0) +/// 处理错误并决定下一步行动 +pub async fn handle_error( + error: &dyn std::error::Error, + context: &mut AgentContext, + config: &ErrorRecoveryConfig, + flow_control: &mut Option<&mut FlowControl>, +) -> ErrorAction { + let error_type = classify_error(error); + let retries_key = format!("retry_{}", error_type.as_str()); + let retries = context.metadata.get(&retries_key).and_then(|v| v.parse().ok()).unwrap_or(0); - if retries >= config.max_retries.get(error_type, 0): - return ErrorAction.FAIL + let max_retries = config.max_retries.get(&error_type).copied().unwrap_or(0); - # 指数退避 - if retries > 0: - backoff = min( - config.backoff_multiplier ** retries, - config.max_backoff - ) - await asyncio.sleep(backoff) + if retries >= max_retries { + return ErrorAction::Fail; + } - context.metadata[f"retry_{error_type.value}"] = retries + 1 + // 指数退避 + if retries > 0 { + let backoff = (config.backoff_multiplier.powi(retries as i32)).min(config.max_backoff); + tokio::time::sleep(tokio::time::Duration::from_secs_f64(backoff)).await; + } - if error_type == ErrorType.RATE_LIMIT: - # 更新流控配置 - flow_control.decrease_rate(0.8) - return ErrorAction.RETRY + context.metadata.insert(retries_key, (retries + 1).to_string()); - elif error_type == ErrorType.SECURITY: - # 安全错误不重试 - return ErrorAction.BLOCK + match error_type { + ErrorType::RateLimit => { + // 更新流控配置 + if let Some(fc) = flow_control { + fc.decrease_rate(0.8); + } + ErrorAction::Retry + } + ErrorType::Security => { + // 安全错误不重试 + ErrorAction::Block + } + ErrorType::Api => { + // API 错误，检查是否可恢复 + if is_retryable_api_error(error) { + ErrorAction::Retry + } else { + ErrorAction::Fail + } + } + _ => ErrorAction::Retry, + } +} - elif error_type == ErrorType.API: - # API 错误，检查是否可恢复 - if is_retryable_api_error(error): - return ErrorAction.RETRY - return ErrorAction.FAIL +fn classify_error(error: &dyn std::error::Error) -> ErrorType { + let msg = error.to_string().to_lowercase(); - return ErrorAction.RETRY + if msg.contains("rate limit") || msg.contains("too many requests") { + ErrorType::RateLimit + } else if msg.contains("timeout") { + ErrorType::Timeout + } else if msg.contains("network") || msg.contains("connection") { + ErrorType::Network + } else if msg.contains("api") { + ErrorType::Api + } else if msg.contains("tool") { + ErrorType::Tool + } else if msg.contains("security") || msg.contains("injection") { + ErrorType::Security + } else { + ErrorType::Internal + } +} -class ErrorAction(Enum): - """错误处理动作""" - RETRY = "retry" - FAIL = "fail" - BLOCK = "block" - FALLBACK = "fallback" +fn is_retryable_api_error(error: &dyn std::error::Error) -> bool { + let msg = error.to_string().to_lowercase(); + // 5xx 错误可重试，4xx 通常不行 + msg.contains("500") || msg.contains("502") || msg.contains("503") || msg.contains("504") +} ``` --- @@ -2159,80 +2626,53 @@ class ErrorAction(Enum): ### 13.1 插件扩展点 -```python -# 输入处理扩展 -class InputBufferPlugin(ABC): - """输入缓冲区插件""" +```rust +// 输入处理扩展 +#[async_trait] +pub trait InputBufferPlugin: Send + Sync { + /// 消息添加前拦截，返回 None 表示跳过 + async fn pre_add_message(&self, message: InputMessage) -> Option; - async def pre_add_message( - self, - message: InputMessage, - ) -> InputMessage | None: - """消息添加前拦截，返回 None 表示跳过""" - pass + /// 消息添加后处理 + async fn post_add_message(&self, message: &InputMessage) {} +} - async def post_add_message( - self, - message: InputMessage, - ) -> None: - """消息添加后处理""" - pass +// 输出处理扩展 +#[async_trait] +pub trait OutputBufferPlugin: Send + Sync { + /// 消息发送前拦截 + async fn pre_send_message(&self, message: OutputMessage) -> Option; -# 输出处理扩展 -class OutputBufferPlugin(ABC): - """输出缓冲区插件""" + /// 消息发送后处理 + async fn post_send_message(&self, message: &OutputMessage) {} +} - async def pre_send_message( - self, - message: OutputMessage, - ) -> OutputMessage | None: - """消息发送前拦截""" - pass +// 安全扩展 +#[async_trait] +pub trait SecurityPlugin: Send + Sync { + /// 自定义注入检测 + async fn check_injection(&self, content: &str) -> Vec; - async def post_send_message( - self, - message: OutputMessage, - ) -> None: - """消息发送后处理""" - pass - -# 安全扩展 -class SecurityPlugin(ABC): - """安全插件""" - - async def check_injection( - self, - content: str, - ) -> list[SecurityIssue]: - """自定义注入检测""" - pass - - async def filter_content( - self, - content: str, - ) -> str: - """自定义内容过滤""" - pass + /// 自定义内容过滤 + async fn filter_content(&self, content: &str) -> String { + content.to_string() + } +} ``` ### 13.2 调度器扩展 -```python -# 自定义调度策略 -class CustomScheduler(ABC): - """自定义调度策略""" +```rust +/// 自定义调度策略 +#[async_trait] +pub trait CustomScheduler: Send + Sync { + /// 选择下一条消息 + async fn select_next_message( + &self, + queues: &HashMap, + ) -> Option; - async def select_next_message( - self, - queues: dict[str, UserMessageQueue], - ) -> InputMessage | None: - """选择下一条消息""" - pass - - async def on_queue_empty( - self, - user_id: str, - ) -> None: - """队列为空时的处理""" - pass + /// 队列为空时的处理 + async fn on_queue_empty(&self, user_id: &str) {} +} ```