Coverage for agentos/protocols/contracts.py: 0%

1""" 

2AgentOS v0.70 — Agent能力契约与发现协议。 

3基因来源: MCP (Model Context Protocol) + OpenAPI Spec 

4 

5契约系统允许Agent声明自己的能力和限制，其他Agent可以通过 

6能力匹配引擎找到合适的Agent协作。 

7 

8契约格式: 

9- AgentCapability: 单个能力描述（名称、描述、输入输出schema） 

10- AgentContract: Agent的完整契约（身份、能力列表、QoS、限制） 

11- CapabilityMatcher: 能力匹配引擎 

12""" 

13 

14from __future__ import annotations 

15 

16from dataclasses import dataclass, field 

17from enum import StrEnum 

18from typing import Any 

19 

20# ── Capability Types ──────────────────────────── 

21 

22 

23class CapabilityDomain(StrEnum): 

24 """能力域枚举。""" 

25 

26 REASONING = "reasoning" # 推理分析 

27 CODING = "coding" # 代码生成 

28 SEARCH = "search" # 信息检索 

29 EXECUTION = "execution" # 命令执行 

30 CREATIVE = "creative" # 创意生成 

31 ANALYSIS = "analysis" # 数据分析 

32 COORDINATION = "coordination" # 协调调度 

33 

34 

35class QoSLevel(StrEnum): 

36 """服务质量等级。""" 

37 

38 BEST_EFFORT = "best_effort" # 尽力而为 

39 HIGH_AVAILABILITY = "ha" # 高可用 

40 LOW_LATENCY = "low_latency" # 低延迟 

41 HIGH_ACCURACY = "high_accuracy" # 高准确 

42 

43 

44@dataclass 

45class AgentCapability: 

46 """单个能力声明。""" 

47 

48 name: str 

49 description: str 

50 domain: CapabilityDomain = CapabilityDomain.REASONING 

51 input_schema: dict = field(default_factory=dict) 

52 output_schema: dict = field(default_factory=dict) 

53 max_tokens: int = 8192 

54 cost_per_call: float = 0.0 

55 avg_latency_ms: float = 1000.0 

56 tags: list[str] = field(default_factory=list) 

57 confidence: float = 0.9 # 0.0 - 1.0 

58 version: str = "1.0.0" 

59 

60 

61@dataclass 

62class AgentContract: 

63 """Agent完整契约 — 身份 + 能力 + 限制。""" 

64 

65 agent_id: str 

66 agent_name: str 

67 agent_type: str = "general" 

68 description: str = "" 

69 capabilities: list[AgentCapability] = field(default_factory=list) 

70 qos_level: QoSLevel = QoSLevel.BEST_EFFORT 

71 rate_limit_rpm: int = 60 # 每分钟最大请求数 

72 max_context_tokens: int = 128000 

73 supported_languages: list[str] = field(default_factory=list) 

74 endpoints: list[str] = field(default_factory=list) 

75 dependencies: list[str] = field(default_factory=list) 

76 health_check_url: str = "" 

77 metadata: dict[str, Any] = field(default_factory=dict) 

78 version: str = "1.0.0" 

79 created_at: str = "" 

80 updated_at: str = "" 

81 

82 def to_dict(self) -> dict: 

83 return { 

84 "agent_id": self.agent_id, 

85 "agent_name": self.agent_name, 

86 "agent_type": self.agent_type, 

87 "description": self.description, 

88 "capabilities": [ 

89 { 

90 "name": c.name, 

91 "domain": c.domain.value, 

92 "description": c.description, 

93 "tags": c.tags, 

94 } 

95 for c in self.capabilities 

96 ], 

97 "qos_level": self.qos_level.value, 

98 "rate_limit_rpm": self.rate_limit_rpm, 

99 "max_context_tokens": self.max_context_tokens, 

100 } 

101 

102 def has_capability(self, name: str) -> bool: 

103 return any(c.name == name for c in self.capabilities) 

104 

105 def has_domain(self, domain: CapabilityDomain) -> bool: 

106 return any(c.domain == domain for c in self.capabilities) 

107 

108 

109# ── Capability Matcher ────────────────────────── 

110 

111 

112@dataclass 

113class MatchScore: 

114 """匹配评分结果。""" 

115 

116 contract: AgentContract 

117 capability: AgentCapability 

118 score: float # 0.0 - 1.0 

119 match_details: dict[str, float] = field(default_factory=dict) 

120 rank: int = 0 

121 

122 

123class CapabilityMatcher: 

124 """ 

125 能力匹配引擎 — 根据查询找到最合适的Agent。 

126 支持: 语义匹配、标签匹配、领域匹配、QoS权重。 

127 """ 

128 

129 def __init__(self, contracts: list[AgentContract] | None = None): 

130 self._contracts: dict[str, AgentContract] = {} 

131 if contracts: 

132 for c in contracts: 

133 self.register(c) 

134 

135 def register(self, contract: AgentContract): 

136 self._contracts[contract.agent_id] = contract 

137 

138 def unregister(self, agent_id: str): 

139 self._contracts.pop(agent_id, None) 

140 

141 def find( 

142 self, 

143 query: str, 

144 domain: CapabilityDomain | None = None, 

145 min_score: float = 0.3, 

146 top_k: int = 5, 

147 ) -> list[MatchScore]: 

148 """ 

149 根据自然语言查询找到最匹配的Agent。 

150 """ 

151 results: list[MatchScore] = [] 

152 

153 for contract in self._contracts.values(): 

154 for cap in contract.capabilities: 

155 if domain and cap.domain != domain: 

156 continue 

157 

158 score = self._compute_match(query, cap, contract) 

159 if score >= min_score: 

160 results.append( 

161 MatchScore( 

162 contract=contract, 

163 capability=cap, 

164 score=score, 

165 match_details={ 

166 "text_similarity": self._text_sim(query, cap), 

167 "domain_match": 1.0 if domain and cap.domain == domain else 0.5, 

168 "tag_overlap": self._tag_overlap(query, cap.tags), 

169 }, 

170 ) 

171 ) 

172 

173 # Sort by score desc 

174 results.sort(key=lambda x: x.score, reverse=True) 

175 

176 # Assign ranks 

177 for i, r in enumerate(results[:top_k]): 

178 r.rank = i + 1 

179 

180 return results[:top_k] 

181 

182 def find_by_domain(self, domain: CapabilityDomain) -> list[AgentContract]: 

183 """按领域查找所有Agent。""" 

184 return [c for c in self._contracts.values() if c.has_domain(domain)] 

185 

186 def find_by_tag(self, tag: str) -> list[AgentContract]: 

187 """按标签查找。""" 

188 return [ 

189 c for c in self._contracts.values() if any(tag in cap.tags for cap in c.capabilities) 

190 ] 

191 

192 def recommend_for_task(self, task_description: str) -> list[MatchScore]: 

193 """根据任务描述推荐Agent。""" 

194 return self.find(query=task_description, top_k=3) 

195 

196 # ── Internal scoring ───────────────────────── 

197 

198 def _compute_match( 

199 self, 

200 query: str, 

201 capability: AgentCapability, 

202 contract: AgentContract, 

203 ) -> float: 

204 """综合匹配评分。""" 

205 scores = [] 

206 

207 # Text similarity (keyword overlap) 

208 scores.append(self._text_sim(query, capability) * 0.35) 

209 

210 # Domain match (if query mentions domain keywords) 

211 domain_hint = self._detect_domain(query) 

212 if domain_hint and domain_hint == capability.domain: 

213 scores.append(0.3) 

214 else: 

215 scores.append(0.1) 

216 

217 # Tag overlap 

218 scores.append(self._tag_overlap(query, capability.tags) * 0.15) 

219 

220 # QoS bonus 

221 qos_bonus = { 

222 QoSLevel.HIGH_ACCURACY: 0.1, 

223 QoSLevel.LOW_LATENCY: 0.08, 

224 QoSLevel.HIGH_AVAILABILITY: 0.05, 

225 QoSLevel.BEST_EFFORT: 0.0, 

226 }.get(contract.qos_level, 0.0) 

227 scores.append(qos_bonus) 

228 

229 # Confidence 

230 scores.append(capability.confidence * 0.1) 

231 

232 return min(sum(scores), 1.0) 

233 

234 def _text_sim(self, query: str, capability: AgentCapability) -> float: 

235 """关键词重叠相似度。""" 

236 q_lower = query.lower() 

237 keywords = set(q_lower.split()) | {q_lower} 

238 

239 # Capability text 

240 cap_text = ( 

241 f"{capability.name} {capability.description} " 

242 f"{' '.join(capability.tags)} {capability.domain.value}" 

243 ).lower() 

244 cap_words = set(cap_text.split()) 

245 

246 if not keywords or not cap_words: 

247 return 0.0 

248 

249 overlap = len(keywords & cap_words) 

250 return min(overlap / max(len(keywords), 1), 1.0) 

251 

252 def _tag_overlap(self, query: str, tags: list[str]) -> float: 

253 """标签重叠率。""" 

254 q_lower = query.lower() 

255 if not tags: 

256 return 0.0 

257 hits = sum(1 for tag in tags if tag.lower() in q_lower) 

258 return hits / len(tags) 

259 

260 def _detect_domain(self, query: str) -> CapabilityDomain | None: 

261 """从查询中检测能力域。""" 

262 q = query.lower() 

263 domain_keywords = { 

264 CapabilityDomain.CODING: ["code", "代码", "编程", "开发", "函数", "bug", "debug"], 

265 CapabilityDomain.SEARCH: ["search", "搜索", "查找", "检索", "资料"], 

266 CapabilityDomain.ANALYSIS: ["分析", "数据", "统计", "图表", "analysis", "data"], 

267 CapabilityDomain.CREATIVE: ["写", "创作", "生成", "设计", "创意", "write", "create"], 

268 CapabilityDomain.EXECUTION: ["运行", "执行", "操作", "命令", "exec", "run"], 

269 CapabilityDomain.COORDINATION: ["协调", "编排", "调度", "工作流", "orchestrate"], 

270 } 

271 for domain, keywords in domain_keywords.items(): 

272 if any(kw in q for kw in keywords): 

273 return domain 

274 return None 

275 

276 

277# ── Contract Registry ─────────────────────────── 

278 

279 

280class ContractRegistry: 

281 """ 

282 契约注册中心 — 分布式Agent能力发现。 

283 支持心跳检测、自动过期。 

284 """ 

285 

286 def __init__(self): 

287 self._contracts: dict[str, AgentContract] = {} 

288 self._heartbeats: dict[str, float] = {} 

289 self._matcher = CapabilityMatcher() 

290 

291 def register(self, contract: AgentContract): 

292 import time 

293 

294 self._contracts[contract.agent_id] = contract 

295 self._heartbeats[contract.agent_id] = time.time() 

296 self._matcher.register(contract) 

297 

298 def heartbeat(self, agent_id: str): 

299 import time 

300 

301 if agent_id in self._contracts: 

302 self._heartbeats[agent_id] = time.time() 

303 

304 def unregister(self, agent_id: str): 

305 self._contracts.pop(agent_id, None) 

306 self._heartbeats.pop(agent_id, None) 

307 self._matcher.unregister(agent_id) 

308 

309 def prune_stale(self, max_idle_seconds: float = 300.0): 

310 """移除超时未心跳的Agent。""" 

311 import time 

312 

313 now = time.time() 

314 stale = [aid for aid, ts in self._heartbeats.items() if now - ts > max_idle_seconds] 

315 for aid in stale: 

316 self.unregister(aid) 

317 return stale 

318 

319 def find(self, query: str, **kwargs) -> list[MatchScore]: 

320 return self._matcher.find(query, **kwargs) 

321 

322 @property 

323 def active_count(self) -> int: 

324 return len(self._contracts) 

325 

326 def list_contracts(self) -> list[AgentContract]: 

327 return list(self._contracts.values()) 

328 

329 def summary(self) -> str: 

330 lines = [f"注册Agent: {self.active_count}"] 

331 for c in self._contracts.values(): 

332 caps = ", ".join(cap.domain.value for cap in c.capabilities) 

333 lines.append(f" {c.agent_name} ({c.agent_type}): [{caps}]") 

334 return "\n".join(lines)