Coverage for src / agent / skills / context_provider.py: 82%

1# Copyright 2025-2026 Microsoft Corporation 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14 

15"""Skill context provider for dynamic instruction injection. 

16 

17This module provides progressive skill documentation using Agent Framework's 

18ContextProvider pattern. Skill documentation is injected on-demand based on 

19user queries and trigger matching, avoiding constant token overhead. 

20""" 

21 

22import logging 

23import re 

24from collections.abc import MutableSequence 

25from typing import Any 

26 

27from agent_framework import ChatMessage, Context, ContextProvider 

28 

29from agent.skills.documentation_index import SkillDocumentationIndex 

30 

31logger = logging.getLogger(__name__) 

32 

33 

34class SkillContextProvider(ContextProvider): 

35 """Progressive skill documentation with on-demand registry. 

36 

37 Implements three-tier progressive disclosure: 

38 1. Minimal breadcrumb (~10 tokens) when skills exist but don't match 

39 2. Full registry (10-15 tokens/skill) when user asks about capabilities 

40 3. Full documentation (hundreds of tokens) when triggers match 

41 

42 Example: 

43 >>> skill_docs = SkillDocumentationIndex() 

44 >>> provider = SkillContextProvider(skill_docs, max_skills=3) 

45 >>> agent = chat_client.create_agent( 

46 ... name="Agent", 

47 ... instructions="...", 

48 ... context_providers=[provider] 

49 ... ) 

50 """ 

51 

52 def __init__( 

53 self, 

54 skill_docs: SkillDocumentationIndex, 

55 memory_manager: Any | None = None, 

56 max_skills: int = 3, 

57 max_all_skills: int = 10, 

58 ): 

59 """Initialize skill context provider. 

60 

61 Args: 

62 skill_docs: SkillDocumentationIndex with loaded skill metadata 

63 memory_manager: Optional memory manager for conversation context (unused) 

64 max_skills: Maximum number of skills to inject when matched (default: 3) 

65 max_all_skills: Cap for "show all skills" to prevent overflow (default: 10) 

66 """ 

67 self.skill_docs = skill_docs 

68 self.memory_manager = memory_manager # For conversation context (future use) 

69 self.max_skills = max_skills 

70 self.max_all_skills = max_all_skills 

71 

72 async def invoking( 

73 self, messages: ChatMessage | MutableSequence[ChatMessage], **kwargs: Any 

74 ) -> Context: 

75 """Inject skill documentation based on request relevance. 

76 

77 Args: 

78 messages: Current conversation messages 

79 **kwargs: Additional context 

80 

81 Returns: 

82 Context with appropriate skill documentation 

83 """ 

84 # 1. Extract current user message 

85 current_message = self._get_latest_user_message(messages) 

86 if not current_message: 

87 return Context() 

88 

89 # 2. Check if user is asking about capabilities 

90 if self._wants_skill_info(current_message): 

91 return self._inject_skill_registry() 

92 

93 # 3. Check for "show all skills" escape hatch 

94 if self._wants_all_skills(current_message): 

95 return self._inject_all_skills_capped() 

96 

97 # 4. Match skills based on current message 

98 relevant_skills = self._match_skills_safely(current_message.lower()) 

99 

100 # 5. Build response based on matches 

101 if relevant_skills: 

102 # Inject full documentation for matched skills 

103 docs = self._build_skill_documentation(relevant_skills[: self.max_skills]) 

104 logger.debug( 

105 f"Injecting {len(relevant_skills[:self.max_skills])} skill(s) documentation" 

106 ) 

107 return Context(instructions=docs) 

108 elif self.skill_docs.has_skills(): 

109 # Hybrid Tier-1 approach: 

110 # - If any skill has explicit triggers (beyond auto-added name) → minimal breadcrumb 

111 # - If no skills have explicit triggers → full registry (better discovery) 

112 if self._any_skill_has_explicit_triggers(): 

113 # Minimal breadcrumb: skill count only 

114 breadcrumb = f"[{self.skill_docs.count()} skills available]" 

115 logger.debug( 

116 f"No skill match - injecting minimal breadcrumb ({self.skill_docs.count()} skills with triggers)" 

117 ) 

118 return Context(instructions=breadcrumb) 

119 else: 

120 # No triggers defined yet → use registry for LLM discovery 

121 logger.debug( 

122 "No skill match - injecting full registry (no skills have structured triggers)" 

123 ) 

124 return self._inject_skill_registry() 

125 else: 

126 # No skills installed - inject nothing 

127 return Context() 

128 

129 def _any_skill_has_explicit_triggers(self) -> bool: 

130 """Check if any skill has explicit structured triggers (beyond skill name). 

131 

132 Returns: 

133 True if at least one skill has more than just its name as a keyword, 

134 or has verbs or patterns defined. 

135 """ 

136 for skill in self.skill_docs.get_all_metadata(): 

137 triggers = skill.get("triggers", {}) 

138 keywords = triggers.get("keywords", []) 

139 # The first keyword is always the skill name (auto-added by model_post_init) 

140 # Explicit triggers exist if there are additional keywords, verbs, or patterns 

141 if len(keywords) > 1 or triggers.get("verbs") or triggers.get("patterns"): 

142 return True 

143 return False 

144 

145 def _inject_skill_registry(self) -> Context: 

146 """Inject skill registry with brief descriptions for LLM-driven discovery.""" 

147 lines = ["## Available Skills\n"] 

148 for skill in self.skill_docs.get_all_metadata(): 

149 # Include brief description for discoverability (10-15 tokens per skill) 

150 brief = skill["brief_description"][:80] # Slightly longer for clarity 

151 lines.append(f"- **{skill['name']}**: {brief}") 

152 

153 lines.append( 

154 "\nWhen you need current information, specialized data, or capabilities " 

155 "beyond your knowledge, consider whether one of these skills could help. " 

156 "Skill scripts are available via the script_run tool." 

157 ) 

158 registry_text = "\n".join(lines) 

159 logger.debug(f"Injecting skill registry with {self.skill_docs.count()} skills") 

160 return Context(instructions=registry_text) 

161 

162 def _wants_skill_info(self, message: str) -> bool: 

163 """Check if user is asking about capabilities.""" 

164 info_patterns = [ 

165 r"\bwhat.*(?:can|could).*(?:you|u).*do\b", 

166 r"\b(?:show|list).*capabilities\b", 

167 r"\bwhat.*skills?\b", 

168 ] 

169 message_lower = message.lower() 

170 return any(re.search(pattern, message_lower) for pattern in info_patterns) 

171 

172 def _wants_all_skills(self, message: str) -> bool: 

173 """Check if user wants to see all skill documentation.""" 

174 all_patterns = [ 

175 r"\bshow.*all.*skills?\b", 

176 r"\blist.*all.*skills?\b", 

177 r"\ball.*skill.*(?:documentation|docs)\b", 

178 ] 

179 message_lower = message.lower() 

180 return any(re.search(pattern, message_lower) for pattern in all_patterns) 

181 

182 def _match_skills_safely(self, context: str) -> list[dict]: 

183 """Match skills with word boundaries and error handling. 

184 

185 Args: 

186 context: User message text (lowercase) 

187 

188 Returns: 

189 List of matched skill metadata dictionaries 

190 """ 

191 matched = [] 

192 seen = set() 

193 

194 for skill in self.skill_docs.get_all_metadata(): 

195 skill_id = skill["name"] 

196 if skill_id in seen: 

197 continue 

198 

199 skill_name_lower = skill["name"].lower() 

200 triggers = skill.get("triggers", {}) 

201 

202 # Check if triggers dict is effectively empty (no keywords, verbs, or patterns) 

203 has_triggers = bool( 

204 triggers.get("keywords") or triggers.get("verbs") or triggers.get("patterns") 

205 ) 

206 

207 # If no triggers defined, only match by skill name (restrictive fallback) 

208 if not has_triggers: 

209 try: 

210 if re.search(rf"\b{re.escape(skill_name_lower)}\b", context): 

211 matched.append(skill) 

212 seen.add(skill_id) 

213 except re.error as e: 

214 logger.warning(f"Regex error matching skill name '{skill_name_lower}': {e}") 

215 continue 

216 

217 # Strategy 1: Skill name mentioned (word boundary) 

218 try: 

219 if re.search(rf"\b{re.escape(skill_name_lower)}\b", context): 

220 matched.append(skill) 

221 seen.add(skill_id) 

222 continue 

223 except re.error as e: 

224 logger.warning(f"Regex error matching skill name '{skill_name_lower}': {e}") 

225 

226 # Strategy 2: Keyword triggers (word boundary) 

227 for keyword in triggers.get("keywords", []): 

228 try: 

229 if re.search(rf"\b{re.escape(keyword.lower())}\b", context): 

230 matched.append(skill) 

231 seen.add(skill_id) 

232 break 

233 except re.error as e: 

234 logger.warning(f"Regex error matching keyword '{keyword}': {e}") 

235 

236 if skill_id in seen: 

237 continue 

238 

239 # Strategy 3: Verb triggers (word boundary) 

240 for verb in triggers.get("verbs", []): 

241 try: 

242 if re.search(rf"\b{re.escape(verb.lower())}\b", context): 

243 matched.append(skill) 

244 seen.add(skill_id) 

245 break 

246 except re.error as e: 

247 logger.warning(f"Regex error matching verb '{verb}': {e}") 

248 

249 if skill_id in seen: 

250 continue 

251 

252 # Strategy 4: Pattern matching (with error handling) 

253 for pattern in triggers.get("patterns", []): 

254 try: 

255 if re.search(pattern, context, re.IGNORECASE): 

256 matched.append(skill) 

257 seen.add(skill_id) 

258 break 

259 except re.error as e: 

260 logger.warning(f"Invalid regex pattern for {skill_id}: {pattern} - {e}") 

261 

262 return matched 

263 

264 def _build_skill_documentation(self, skills: list[dict]) -> str: 

265 """Build full documentation for matched skills. 

266 

267 Args: 

268 skills: List of skill metadata dictionaries 

269 

270 Returns: 

271 Formatted skill documentation string 

272 """ 

273 docs = ["## Relevant Skill Documentation\n"] 

274 for skill in skills: 

275 docs.append(f"### {skill['name']}\n") 

276 docs.append(skill.get("instructions", "")) 

277 docs.append("") 

278 return "\n".join(docs) 

279 

280 def _inject_all_skills_capped(self) -> Context: 

281 """Inject skill documentation with cap to avoid context overflow.""" 

282 all_skills = self.skill_docs.get_all_metadata() 

283 

284 if len(all_skills) <= self.max_all_skills: 

285 # Show all if under cap 

286 docs = self._build_skill_documentation(all_skills) 

287 else: 

288 # Show capped list with note 

289 docs = self._build_skill_documentation(all_skills[: self.max_all_skills]) 

290 docs += f"\n\n*Showing {self.max_all_skills} of {len(all_skills)} skills. " 

291 docs += "Ask about specific skills for more details.*" 

292 

293 logger.debug(f"Injecting all skills (capped at {self.max_all_skills})") 

294 return Context(instructions=docs) 

295 

296 def _get_latest_user_message(self, messages: ChatMessage | MutableSequence[ChatMessage]) -> str: 

297 """Extract the latest user message. 

298 

299 Args: 

300 messages: Current conversation messages 

301 

302 Returns: 

303 Latest user message text or empty string 

304 """ 

305 msg_list = messages if isinstance(messages, MutableSequence) else [messages] 

306 

307 # Find user message (robust extraction like MemoryContextProvider) 

308 for msg in reversed(msg_list): 

309 role = str(getattr(msg, "role", "")) 

310 if "user" in role.lower(): 

311 return self._extract_message_text(msg) 

312 

313 return "" 

314 

315 def _extract_message_text(self, msg: ChatMessage) -> str: 

316 """Extract text from a ChatMessage (copied from MemoryContextProvider). 

317 

318 Args: 

319 msg: Chat message 

320 

321 Returns: 

322 Message text or empty string 

323 """ 

324 # Try different attributes the message might have 

325 if hasattr(msg, "text"): 

326 return str(msg.text) 

327 elif hasattr(msg, "content"): 

328 content = msg.content 

329 if isinstance(content, str): 

330 return content 

331 elif isinstance(content, list): 

332 # Handle list of content items 

333 texts = [] 

334 for item in content: 

335 if hasattr(item, "text"): 

336 texts.append(str(item.text)) 

337 elif isinstance(item, dict) and "text" in item: 

338 texts.append(str(item["text"])) 

339 return " ".join(texts) if texts else "" 

340 else: 

341 return str(content) 

342 else: 

343 # Fallback to string representation 

344 return str(msg) if msg else ""