Spaces:
Runtime error
Runtime error
File size: 5,258 Bytes
08ad87f |
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 |
# utils/crew_course_agents.py
from crewai import Agent, Task, Crew, Process
from langchain.llms import Anthropic
from langchain.tools import DuckDuckGoSearchRun
from pathlib import Path
import json
import streamlit as st
class CourseCrewOrchestrator:
def __init__(self, anthropic_api_key):
self.llm = Anthropic(anthropic_api_key=anthropic_api_key)
self.search_tool = DuckDuckGoSearchRun()
self.data_dir = Path("data/courses")
self.data_dir.mkdir(parents=True, exist_ok=True)
def create_agents(self):
"""Create specialized agents for course creation"""
curriculum_designer = Agent(
role='Curriculum Designer',
goal='Design comprehensive and structured trading course outlines',
backstory='Expert in educational design with years of experience in financial education',
tools=[self.search_tool],
llm=self.llm,
verbose=True
)
content_creator = Agent(
role='Content Creator',
goal='Create engaging and informative trading content',
backstory='Experienced financial writer and educator with expertise in explaining complex concepts',
tools=[self.search_tool],
llm=self.llm,
verbose=True
)
technical_expert = Agent(
role='Trading Technical Expert',
goal='Provide deep technical insights and practical trading knowledge',
backstory='Professional trader with 15+ years of market experience',
tools=[self.search_tool],
llm=self.llm,
verbose=True
)
exercise_creator = Agent(
role='Exercise Designer',
goal='Create practical exercises and assessments',
backstory='Expert in creating hands-on learning experiences in trading',
tools=[self.search_tool],
llm=self.llm,
verbose=True
)
return {
'curriculum': curriculum_designer,
'content': content_creator,
'technical': technical_expert,
'exercise': exercise_creator
}
def create_course_tasks(self, agents, topic):
"""Create sequential tasks for course creation"""
tasks = [
Task(
description=f"Design a comprehensive curriculum outline for {topic}. Include learning objectives, prerequisites, and module structure.",
agent=agents['curriculum']
),
Task(
description=f"Create detailed content for each module in the {topic} curriculum. Focus on clear explanations and examples.",
agent=agents['content']
),
Task(
description=f"Add technical insights and practical trading strategies for {topic}. Include real market examples.",
agent=agents['technical']
),
Task(
description=f"Design exercises, quizzes, and practical assignments for {topic}. Include answer keys and explanations.",
agent=agents['exercise']
)
]
return tasks
async def generate_course(self, course_id, topic):
"""Generate course content using CrewAI"""
try:
# Create status file
self._update_status(course_id, 'starting', 0)
# Initialize agents and tasks
agents = self.create_agents()
tasks = self.create_course_tasks(agents, topic)
# Create and run the crew
course_crew = Crew(
agents=list(agents.values()),
tasks=tasks,
process=Process.sequential
)
# Execute tasks and save results
result = await course_crew.run()
course_content = self._parse_crew_result(result)
# Save course content
self._save_course_content(course_id, course_content)
self._update_status(course_id, 'complete', 100)
return course_content
except Exception as e:
self._update_status(course_id, 'error', 0, str(e))
raise e
def _update_status(self, course_id, status, progress, error=None):
"""Update course generation status"""
status_data = {
'status': status,
'progress': progress,
'error': error
}
status_file = self.data_dir / f"{course_id}_status.json"
with open(status_file, 'w') as f:
json.dump(status_data, f)
def _save_course_content(self, course_id, content):
"""Save generated course content"""
content_file = self.data_dir / f"{course_id}_content.json"
with open(content_file, 'w') as f:
json.dump(content, f)
def _parse_crew_result(self, result):
"""Parse and structure the crew's output"""
# Implement parsing logic based on the crew's output format
return {
'curriculum': result.get('curriculum', {}),
'content': result.get('content', {}),
'technical': result.get('technical', {}),
'exercises': result.get('exercises', {})
} |