Coverage for integrations / expert_agents / __init__.py: 32.3%
96 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-12 04:49 +0000
1"""
2Expert Agents Integration - Dream Fulfillment Network for Autogen
4This module integrates the 96-agent expert network with the Autogen framework.
5It bridges expert agents with AgentSkillRegistry, SmartLedger, and human-in-the-loop.
7Usage:
8 from integrations.expert_agents import get_expert_for_task, register_all_experts
10 # Register all 96 expert agents with the skill registry
11 register_all_experts(skill_registry)
13 # Get best expert for a task (with human confirmation)
14 expert_agent = get_expert_for_task("I want to build a mobile app", skill_registry)
15"""
17from typing import Dict, List, Optional, Any
18from .registry import ExpertAgentRegistry, ExpertAgent, AgentCategory
19import logging
21logger = logging.getLogger(__name__)
23_experts_registered = False
24_registered_agents: Dict[str, "ExpertAgent"] = {}
27def register_all_experts(skill_registry) -> Dict[str, ExpertAgent]:
28 """
29 Register all 96 expert agents with the AgentSkillRegistry.
31 This makes expert agents discoverable via the existing skill registry system.
32 Idempotent — safe to call multiple times (only registers once).
34 Args:
35 skill_registry: Instance of AgentSkillRegistry from internal_comm
37 Returns:
38 Dictionary of expert_id -> ExpertAgent
39 """
40 global _experts_registered, _registered_agents
41 if _experts_registered:
42 return _registered_agents
44 logger.info("Registering 96 expert agents with skill registry...")
46 # Load expert registry
47 expert_registry = ExpertAgentRegistry()
49 # Register each expert with the skill registry
50 for agent_id, expert in expert_registry.agents.items():
51 # Convert expert capabilities to skill registry format
52 skills = []
53 for capability in expert.capabilities:
54 skills.append({
55 'name': capability.name,
56 'description': capability.description,
57 'proficiency': expert.reliability, # Use expert reliability as proficiency
58 'metadata': {
59 'example_use': capability.example_use,
60 'category': expert.category.value,
61 'model_type': expert.model_type,
62 'endpoint': expert.endpoint,
63 'cost_per_call': expert.cost_per_call,
64 'avg_latency_ms': expert.avg_latency_ms
65 }
66 })
68 # Register with skill registry
69 skill_registry.register_agent(agent_id, skills)
71 logger.info(f"Successfully registered {len(expert_registry.agents)} expert agents")
72 _experts_registered = True
73 _registered_agents = expert_registry.agents
74 return expert_registry.agents
77def get_expert_for_task(task_description: str, skill_registry,
78 category: Optional[AgentCategory] = None,
79 require_human_approval: bool = True) -> Optional[str]:
80 """
81 Find the best expert agent for a given task with human-in-the-loop.
83 This is the key function for dream fulfillment - it intelligently matches
84 tasks to the best expert agent and allows human oversight.
86 Args:
87 task_description: What the user wants to achieve
88 skill_registry: Instance of AgentSkillRegistry
89 category: Optional category filter
90 require_human_approval: If True, ask human to confirm expert selection
92 Returns:
93 agent_id of selected expert, or None if no match
94 """
95 logger.info(f"Finding expert for task: {task_description}")
97 # Load expert registry
98 expert_registry = ExpertAgentRegistry()
100 # Search for matching experts
101 if category:
102 # Filter by category first
103 candidates = expert_registry.get_agents_by_category(category)
104 # Then search within category
105 matched = expert_registry.search_agents(task_description, category.value)
106 else:
107 # Search across all experts
108 matched = expert_registry.search_agents(task_description)
110 if not matched:
111 logger.warning(f"No expert found for task: {task_description}")
112 return None
114 # Get top 3 candidates
115 top_candidates = matched[:3]
117 if require_human_approval:
118 # Present options to human
119 print("\n" + "="*80)
120 print(f"TASK: {task_description}")
121 print("="*80)
122 print("\nRecommended Expert Agents:\n")
124 for i, expert in enumerate(top_candidates, 1):
125 print(f"{i}. {expert.name} ({expert.category.value})")
126 print(f" Description: {expert.description}")
127 print(f" Capabilities: {', '.join([c.name for c in expert.capabilities[:3]])}")
128 print(f" Reliability: {expert.reliability*100:.0f}%")
129 print()
131 # Ask for human confirmation
132 while True:
133 choice = input(f"Select expert (1-{len(top_candidates)}) or 's' to skip: ").strip().lower()
134 if choice == 's':
135 logger.info("Human skipped expert selection")
136 return None
137 try:
138 idx = int(choice) - 1
139 if 0 <= idx < len(top_candidates):
140 selected = top_candidates[idx]
141 logger.info(f"Human selected expert: {selected.name} ({selected.agent_id})")
142 return selected.agent_id
143 except ValueError:
144 pass
145 print("Invalid choice. Please try again.")
146 else:
147 # Auto-select best match
148 selected = top_candidates[0]
149 logger.info(f"Auto-selected expert: {selected.name} ({selected.agent_id})")
150 return selected.agent_id
153def get_expert_info(agent_id: str) -> Optional[Dict[str, Any]]:
154 """
155 Get detailed information about an expert agent.
157 Args:
158 agent_id: Expert agent identifier
160 Returns:
161 Dictionary with expert details, or None if not found
162 """
163 expert_registry = ExpertAgentRegistry()
164 expert = expert_registry.get_agent(agent_id)
166 if not expert:
167 return None
169 return {
170 'agent_id': expert.agent_id,
171 'name': expert.name,
172 'category': expert.category.value,
173 'description': expert.description,
174 'capabilities': [
175 {
176 'name': cap.name,
177 'description': cap.description,
178 'example_use': cap.example_use
179 }
180 for cap in expert.capabilities
181 ],
182 'endpoint': expert.endpoint,
183 'model_type': expert.model_type,
184 'cost_per_call': expert.cost_per_call,
185 'avg_latency_ms': expert.avg_latency_ms,
186 'reliability': expert.reliability
187 }
190def create_autogen_expert_wrapper(agent_id: str, config_list: List[Dict],
191 skill_registry) -> Optional[Any]:
192 """
193 Create an Autogen ConversableAgent wrapper for an expert agent.
195 This allows expert agents to be used directly in Autogen conversations.
197 Args:
198 agent_id: Expert agent identifier
199 config_list: Autogen config list for LLM
200 skill_registry: Instance of AgentSkillRegistry
202 Returns:
203 Autogen ConversableAgent instance, or None if expert not found
204 """
205 from autogen import ConversableAgent
207 expert_registry = ExpertAgentRegistry()
208 expert = expert_registry.get_agent(agent_id)
210 if not expert:
211 logger.error(f"Expert {agent_id} not found")
212 return None
214 # Create system message from expert profile
215 system_message = f"""You are {expert.name}, a specialized AI agent.
217{expert.description}
219Your capabilities include:
220{chr(10).join(f'- {cap.name}: {cap.description}' for cap in expert.capabilities)}
222You are part of a dream fulfillment network helping users achieve their goals.
223Focus on your area of expertise and provide practical, actionable guidance."""
225 # Create Autogen agent
226 agent = ConversableAgent(
227 name=agent_id,
228 system_message=system_message,
229 llm_config={
230 "config_list": config_list,
231 "temperature": 0.7,
232 "max_tokens": 1000
233 },
234 human_input_mode="NEVER"
235 )
237 # Register agent's skills with skill registry
238 skills = []
239 for capability in expert.capabilities:
240 skills.append({
241 'name': capability.name,
242 'description': capability.description,
243 'proficiency': expert.reliability,
244 'metadata': {
245 'example_use': capability.example_use,
246 'category': expert.category.value
247 }
248 })
249 skill_registry.register_agent(agent_id, skills)
251 logger.info(f"Created Autogen wrapper for expert: {expert.name}")
252 return agent
255def match_expert_for_context(task_description: str, top_k: int = 1,
256 min_score: int = 4) -> Optional[Dict[str, Any]]:
257 """Match an expert agent and return a prompt enhancement block.
259 Non-interactive alternative to get_expert_for_task(). Returns structured
260 data for prompt injection, following the build_personality_prompt() pattern.
262 Args:
263 task_description: Description of the task or action.
264 top_k: Number of top matches to consider.
265 min_score: Minimum score threshold (higher = stricter matching).
267 Returns:
268 Dict with keys: agent_id, name, description, prompt_block, capabilities.
269 None if no match exceeds min_score.
270 """
271 if not task_description or not task_description.strip():
272 return None
274 try:
275 registry = ExpertAgentRegistry()
276 scored = registry.score_match(task_description)
278 if not scored or scored[0][1] < min_score:
279 return None
281 agent, score = scored[0]
283 # Build prompt block (same pattern as build_personality_prompt)
284 cap_lines = '\n'.join(
285 f'- {c.name}: {c.description}' for c in agent.capabilities[:3]
286 )
287 prompt_block = (
288 f"\n[Expert Guidance: {agent.name}]\n"
289 f"Domain: {agent.description}\n"
290 f"Relevant capabilities:\n{cap_lines}\n"
291 )
293 return {
294 'agent_id': agent.agent_id,
295 'name': agent.name,
296 'description': agent.description,
297 'prompt_block': prompt_block,
298 'capabilities': [
299 {'name': c.name, 'description': c.description}
300 for c in agent.capabilities
301 ],
302 'score': score,
303 }
304 except Exception as e:
305 logger.debug(f"Expert matching failed: {e}")
306 return None
309def recommend_experts_for_dream(dream_statement: str, top_k: int = 5) -> List[ExpertAgent]:
310 """
311 Recommend expert agents for achieving a dream.
313 This is the high-level function for dream fulfillment.
315 Args:
316 dream_statement: User's dream in natural language
317 top_k: Number of experts to recommend
319 Returns:
320 List of recommended ExpertAgent instances
321 """
322 expert_registry = ExpertAgentRegistry()
324 # Use the registry's recommendation engine
325 # Use "general" as default dream_category for broad search
326 recommended = expert_registry.recommend_agents(dream_statement, "general")
328 return recommended[:top_k]
331# Export main functions
332__all__ = [
333 'ExpertAgentRegistry',
334 'ExpertAgent',
335 'AgentCategory',
336 'register_all_experts',
337 'get_expert_for_task',
338 'get_expert_info',
339 'create_autogen_expert_wrapper',
340 'recommend_experts_for_dream',
341 'match_expert_for_context',
342]