Spaces:
Sleeping
Sleeping
File size: 11,856 Bytes
7644eac |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 |
"""
Research Agent for autonomous learning
Handles research tasks and knowledge acquisition
"""
from typing import List, Dict, Any, Optional
from datetime import datetime
import json
from .base_agent import BaseAgent
class ResearchAgent(BaseAgent):
"""
Specialized agent for conducting research and acquiring knowledge
"""
def __init__(self, api_key: Optional[str] = None):
super().__init__(api_key)
self.research_topics = []
self.research_history = []
self.current_research_focus = None
def execute_task(self, task: Dict[str, Any]) -> Dict[str, Any]:
"""
Execute a research task
Args:
task: Task description and parameters
Returns:
Research results
"""
task_type = task.get("type", "research")
if task_type == "research":
return self.conduct_research(task)
elif task_type == "update_knowledge":
return self.update_knowledge(task)
elif task_type == "analyze_trends":
return self.analyze_trends(task)
else:
return {
"success": False,
"message": f"Unknown task type: {task_type}"
}
def conduct_research(self, task: Dict[str, Any]) -> Dict[str, Any]:
"""
Conduct research on a specific topic
Args:
task: Research task parameters
Returns:
Research findings
"""
topic = task.get("topic")
depth = task.get("depth", "medium")
context = task.get("context", [])
if not topic:
return {
"success": False,
"message": "Topic is required for research"
}
# Create a more intelligent research prompt that can handle any type of query
context_str = '\n'.join(context) if context else ''
# Enhanced prompt with better instruction and flexibility for ANY topic
prompt = f"""
I want you to act as an expert AI educational assistant with comprehensive knowledge in all fields of study, technologies, skills, and courses. The user has requested information about: "{topic}"
This could be a request for:
1. A learning path on this topic or skill
2. Specific research or information on this subject
3. How to accomplish a task or learn a technique
4. Explanations or definitions of concepts
5. Course recommendations for a particular field
6. Career advice related to skills or technologies
7. Comparisons between different technologies, methods, or approaches
YOUR GOAL: Provide the most helpful, accurate, and comprehensive information possible about ANY educational topic the user asks about. You should be able to address questions about programming languages, data science, machine learning, web development, mobile development, cloud computing, cybersecurity, design, business, humanities, sciences, mathematics, or any other educational subject.
If it seems like they want a learning path:
- Provide a step-by-step progression from basics to advanced
- Include estimated time commitments for each stage
- Recommend specific resources (books, courses, tutorials) for each step
If it seems like they want specific information:
- Provide detailed, technically accurate information
- Include practical applications and examples
- Balance theoretical knowledge with practical insights
Your response should be thorough, accurate, helpful for any level of expertise, and include both theoretical understanding and practical application.
Additional context:
{context_str}
Provide your findings in this JSON format:
{{
"summary": "A clear 2-3 paragraph summary answering the query directly, with specific details and actionable insights",
"key_concepts": ["List of 4-6 key concepts relevant to the query, with brief explanations"],
"learning_path": ["Detailed steps for learning this topic in a logical order, from beginner to advanced"],
"resources": ["Specific recommended resources including books, courses, tutorials, documentation, and communities"],
"code_examples": ["Relevant code examples or practical exercises that demonstrate key concepts"],
"advanced_topics": ["More advanced topics to explore after mastering basics, with brief explanations of why they matter"],
"career_applications": ["How these skills apply to real-world jobs and career paths"],
"curiosity_trails": ["A list of 3-5 intriguing follow-up questions or related sub-topics to explore further, designed to spark curiosity and deeper learning."]
}}
For the "curiosity_trails", think about what someone who has just learned the main topic might wonder next, or what fascinating related areas they could branch into.
Be extremely thorough, accurate, and helpful. Don't just provide general advice - give specific, actionable information that would genuinely help someone learn this topic or skill.
"""
# Generate research findings with error handling
findings_json = self.model_orchestrator.generate_structured_response(
prompt=prompt,
output_schema="""
{
"summary": "string",
"key_concepts": ["string"],
"learning_path": ["string"],
"resources": ["string"],
"code_examples": ["string"],
"advanced_topics": ["string"],
"career_applications": ["string"],
"curiosity_trails": ["string"]
}
"""
)
if not findings_json:
return {
"success": False,
"message": "AI provider did not return a valid response. Please try again later."
}
try:
findings = json.loads(findings_json)
except Exception as e:
return {
"success": False,
"message": f"Failed to parse AI response: {str(e)}",
"raw_response": findings_json
}
# Store findings
self.add_to_memory(f"Research findings on {topic}: {json.dumps(findings)}")
self.research_history.append({
"topic": topic,
"timestamp": datetime.now().isoformat(),
"depth": depth,
"findings": findings
})
return {
"success": True,
"findings": findings,
"message": f"Successfully completed research on {topic}"
}
def update_knowledge(self, task: Dict[str, Any]) -> Dict[str, Any]:
"""
Update knowledge based on new information
Args:
task: Update task parameters
Returns:
Update results
"""
new_info = task.get("new_information")
related_topics = task.get("related_topics", [])
if not new_info:
return {
"success": False,
"message": "New information is required for knowledge update"
}
# Analyze new information
prompt = f"""
Analyze this new information and update existing knowledge:
{new_info}
"""
# Include related topics
related_topics = self._find_related_topics(new_info)
if related_topics:
related_topics_str = '\n'.join(related_topics)
prompt += f"\n\nRelated topics to consider:\n{related_topics_str}"
prompt += f"""
Identify:
1. What new knowledge should be added
2. What existing knowledge should be updated
3. What knowledge should be deprecated
"""
analysis = json.loads(self.model_orchestrator.generate_structured_response(
prompt=prompt,
output_schema="""
{
"new_knowledge": ["string"],
"updated_knowledge": ["string"],
"deprecated_knowledge": ["string"]
}
"""
))
# Update knowledge base
self.add_to_memory(f"Knowledge update: {json.dumps(analysis)}")
return {
"success": True,
"analysis": analysis,
"message": "Knowledge base updated successfully"
}
def analyze_trends(self, task: Dict[str, Any]) -> Dict[str, Any]:
"""
Analyze trends in a specific area
Args:
task: Trend analysis parameters
Returns:
Trend analysis results
"""
area = task.get("area")
timeframe = task.get("timeframe", "recent")
context = task.get("context", [])
if not area:
return {
"success": False,
"message": "Area is required for trend analysis"
}
# Create analysis prompt
prompt = f"""
Analyze current trends in: {area}
Timeframe: {timeframe}
"""
# Add context if available
if context:
context_str = '\n'.join(context)
prompt += f"\n\nContext:\n{context_str}"
prompt += f"""
Provide analysis in JSON format with:
- Current trends
- Emerging patterns
- Predicted developments
- Impact assessment
"""
# Generate trend analysis
analysis = json.loads(self.model_orchestrator.generate_structured_response(
prompt=prompt,
output_schema="""
{
"current_trends": ["string"],
"emerging_patterns": ["string"],
"predicted_developments": ["string"],
"impact": ["string"]
}
"""
))
# Store analysis
self.add_to_memory(f"Trend analysis for {area}: {json.dumps(analysis)}")
return {
"success": True,
"analysis": analysis,
"message": f"Successfully analyzed trends in {area}"
}
def plan_next_research(self) -> Dict[str, Any]:
"""
Plan next research task based on current knowledge
Returns:
Next research plan
"""
# Get current knowledge gaps
relevant_memories = self.get_relevant_memory("knowledge gaps")
# Create planning prompt
memory_summary = "\n".join(item["content"] for item in relevant_memories)
prompt = f"""
Based on current knowledge:
{memory_summary}
Identify:
1. Most important knowledge gaps
2. Areas requiring deeper research
3. Emerging topics to explore
Propose next research task with:
- Topic
- Research depth
- Related topics
"""
# Generate research plan
plan = json.loads(self.model_orchestrator.generate_structured_response(
prompt=prompt,
output_schema="""
{
"topic": "string",
"depth": "string",
"related_topics": ["string"],
"reason": "string"
}
"""
))
# Store plan
self.add_to_memory(f"Next research plan: {json.dumps(plan)}")
return plan
|