Coverage for merco/core/pipeline.py: 72%

1"""结果处理管道 — 链式变换工具结果的可扩展中间层 

2 

3设计原则： 

4- 每个 Processor 只做一件事，通过 use() 组合 

5- 管线顺序即执行顺序，先注册的先执行 

6- Processor 返回 True = 停止管线，后续不执行 

7- context.side_effects 存储需额外注入的消息（如 skill 内容作为 user message） 

8- 零动态开销：简单列表遍历，无反射/装饰器 

9""" 

10 

11from __future__ import annotations 

12 

13import logging 

14from abc import ABC, abstractmethod 

15from dataclasses import dataclass, field 

16 

17logger = logging.getLogger("merco.pipeline") 

18 

19# ── 上下文 ─────────────────────────────────────────────── 

20 

21@dataclass 

22class ProcessContext: 

23 """管线上下文，在处理器链之间传递和变换""" 

24 

25 tool_name: str 

26 arguments: dict 

27 result: dict # 当前结果，处理器可原地修改 

28 tool_schema: dict | None = None 

29 tool_call_id: str = "" 

30 # 额外要添加到上下文的 side-effect 消息（如 skill 内容注入） 

31 extra_messages: list[dict] = field(default_factory=list) 

32 # 处理器间共享的临时数据 

33 metadata: dict = field(default_factory=dict) 

34 

35 

36# ── 处理器 ─────────────────────────────────────────────── 

37 

38class Processor(ABC): 

39 """处理器基类。每个处理器只做一件事。""" 

40 

41 name: str = "" 

42 

43 @abstractmethod 

44 async def process(self, ctx: ProcessContext) -> bool: 

45 """处理结果。返回 True 以停止管线（后续处理器跳过）。""" 

46 ... 

47 

48 

49# ── 管线 ───────────────────────────────────────────────── 

50 

51class ResultPipeline: 

52 """结果处理管线。用 use() 注册处理器，process() 链式执行。""" 

53 

54 def __init__(self): 

55 self._processors: list[Processor] = [] 

56 self._by_name: dict[str, Processor] = {} 

57 self._disabled: set[str] = set() 

58 

59 def use(self, processor: Processor) -> "ResultPipeline": 

60 """注册处理器。按注册顺序执行。""" 

61 self._processors.append(processor) 

62 self._by_name[processor.name] = processor 

63 return self 

64 

65 def disable(self, name: str) -> None: 

66 """临时禁用指定处理器""" 

67 self._disabled.add(name) 

68 

69 def enable(self, name: str) -> None: 

70 """重新启用指定处理器""" 

71 self._disabled.discard(name) 

72 

73 async def process(self, ctx: ProcessContext) -> None: 

74 """链式执行所有启用的处理器""" 

75 for p in self._processors: 

76 if p.name in self._disabled: 

77 continue 

78 try: 

79 stop = await p.process(ctx) 

80 if stop: 

81 break 

82 except Exception: 

83 logger.warning("管线处理器 '%s' 异常", p.name, exc_info=True) 

84 

85 

86 

87def _walk_truncate(obj, max_per_value: int, filepath: str, pagination: dict | None = None): 

88 """递归截断字典/列表中所有超长字符串，附加分页信息。 

89 

90 兼容所有工具返回格式——bash.stdout、read_file.content、web_fetch.content 等。 

91 非字符串值原样保留，嵌套结构递归处理。 

92 """ 

93 if isinstance(obj, str): 

94 if len(obj) <= max_per_value: 

95 return obj 

96 preview = obj[:max_per_value] 

97 page_info = "" 

98 if pagination: 

99 # 用 offset/limit 翻页（不再用已删除的 char_offset/char_limit） 

100 # page_size 是字符数，按 ~60 字符/行估算行数 

101 lines_per_page = max(1, pagination["page_size"] // 60) 

102 next_offset = lines_per_page + 1 

103 page_info = ( 

104 f"\n📄 第 1/{pagination['total_pages']} 页" 

105 f"（共 {pagination['total_chars']:,} 字符）\n" 

106 f"➡️ 下一页: read_file {filepath} offset={next_offset} limit={lines_per_page}" 

107 ) 

108 return f"{preview}\n\n⚠️ 已截断{page_info}" 

109 if isinstance(obj, dict): 

110 return {k: _walk_truncate(v, max_per_value, filepath, pagination) for k, v in obj.items()} 

111 if isinstance(obj, list): 

112 return [_walk_truncate(v, max_per_value, filepath, pagination) for v in obj] 

113 return obj 

114 

115 

116def _is_reading_trunc_file(ctx: ProcessContext, trunc_dir: str) -> bool: 

117 """检测是否正在读取截断缓存文件（防套娃）。""" 

118 # read_file(path=...) / write_file(path=...) 等文件工具的 path 参数 

119 path_arg = ctx.arguments.get("path", "") 

120 if isinstance(path_arg, str) and path_arg.startswith(trunc_dir): 

121 return True 

122 # bash 中也可能 cat/grep 截断文件，但结果通常不超限 → 低风险 

123 # 未来可扩展其他工具 

124 return False 

125 

126 

127 

128# ── LLM 调用恢复管线 ────────────────────────────────────── 

129 

130@dataclass 

131class RecoveryContext: 

132 """恢复管线上下文——策略的「诊断面板」。 

133 

134 调用方（Agent）填入诊断信息，策略根据这些信息表达恢复意图。 

135 策略只设置标志位，不执行——由 Agent 在管线返回后统一执行。 

136 """ 

137 

138 # ── 诊断（Agent 填入）──────────────── 

139 error: Exception 

140 status_code: int = 0 # HTTP 状态码（APIStatusError 自动提取） 

141 context_tokens: int = 0 # 当前上下文 token 估算 

142 tool_count: int = 0 # 请求中的工具数量 

143 attempt_count: int = 0 # 本轮已尝试恢复次数（跨策略累计） 

144 model: str = "" # 当前模型名 

145 

146 # ── 意图标志位（策略设置，Agent 执行）── 

147 compress: bool = False # 需要压缩上下文 

148 reduce_tools: bool = False # 需要精简工具列表 

149 switch_model: str | None = None # 切换到指定模型 

150 reduce_history: bool = False # 需要裁剪对话历史 

151 extra_wait: float = 0.0 # 额外等待时间（秒），0 = 不等 

152 

153 # ── 限制 ──────────────────────────── 

154 compress_count: int = 0 # 已执行压缩的次数 

155 max_compress: int = 2 # 允许的最大压缩次数 

156 max_reduce: int = 1 # 允许的最大精简次数 

157 

158 

159class Recovery: 

160 """LLM 调用失败时的恢复策略基类。 

161 

162 子类读取 RecoveryContext 的诊断字段做决策，设置标志位表达意图。 

163 返回 True = 已决策恢复，Agent 执行标志位后重试；False = 无法处理。 

164 

165 框架预留的动态能力（当前未实现，标志位已就绪）: 

166 - reduce_tools: 上下文过大时自动精简工具（如关闭 idle/5xx 的工具集） 

167 - switch_model: 当前模型不可用时降级到备选模型 

168 - reduce_history: 裁剪遥远的对话轮次 

169 """ 

170 

171 name: str = "" 

172 

173 @abstractmethod 

174 async def attempt(self, ctx: RecoveryContext) -> bool: 

175 """根据 ctx 诊断信息做决策，设置标志位。""" 

176 ... 

177 

178 

179class RecoveryPipeline: 

180 """LLM 调用恢复管线。按注册顺序尝试，第一个成功的策略生效。""" 

181 

182 def __init__(self): 

183 self._recoveries: list[Recovery] = [] 

184 self._disabled: set[str] = set() 

185 

186 def use(self, recovery: Recovery) -> "RecoveryPipeline": 

187 self._recoveries.append(recovery) 

188 return self 

189 

190 def disable(self, name: str) -> None: 

191 self._disabled.add(name) 

192 

193 def enable(self, name: str) -> None: 

194 self._disabled.discard(name) 

195 

196 async def attempt(self, ctx: RecoveryContext) -> bool: 

197 """依次尝试恢复策略，任一成功即返回 True""" 

198 for r in self._recoveries: 

199 if r.name in self._disabled: 

200 continue 

201 try: 

202 if await r.attempt(ctx): 

203 ctx.attempt_count += 1 

204 if ctx.compress: 

205 ctx.compress_count += 1 

206 return True 

207 except Exception: 

208 logger.warning("恢复策略 '%s' 异常", r.name, exc_info=True) 

209 return False 

210 

211 

212# ── 空回复处理管线 ───────────────────────────────────────── 

213 

214@dataclass 

215class EmptyResponseContext: 

216 """空回复处理的上下文。管线策略读取诊断字段，设置标志位。""" 

217 

218 reasoning: str = "" # 模型输出的 reasoning 内容 

219 retry_count: int = 0 # 本轮已重试次数 

220 max_retries: int = 2 # 允许的最大重试次数 

221 

222 # ── 意图标志位 ──────────────── 

223 inject_error: str | None = None # 要注入上下文的 user 错误消息 

224 

225 

226class EmptyResponseStrategy(ABC): 

227 """空回复策略：第一个 enabled 的策略生效。""" 

228 

229 name: str = "" 

230 

231 @abstractmethod 

232 async def attempt(self, ctx: EmptyResponseContext) -> bool: 

233 """返回 True = 已处理，Agent 执行标志位后重试；False = 放过""" 

234 ... 

235 

236 

237class EmptyResponsePipeline: 

238 """空回复处理管线。按注册顺序尝试，第一个成功的策略生效。""" 

239 

240 def __init__(self): 

241 self._strategies: list[EmptyResponseStrategy] = [] 

242 self._disabled: set[str] = set() 

243 

244 def use(self, strategy: EmptyResponseStrategy) -> "EmptyResponsePipeline": 

245 self._strategies.append(strategy) 

246 return self 

247 

248 def disable(self, name: str) -> None: 

249 self._disabled.add(name) 

250 

251 def enable(self, name: str) -> None: 

252 self._disabled.discard(name) 

253 

254 async def attempt(self, ctx: EmptyResponseContext) -> bool: 

255 for s in self._strategies: 

256 if s.name in self._disabled: 

257 continue 

258 try: 

259 if await s.attempt(ctx): 

260 return True 

261 except Exception: 

262 logger.warning("空回复策略 '%s' 异常", s.name, exc_info=True) 

263 return False