Spaces:
Sleeping
Sleeping
| from langchain_core.tools import tool | |
| from typing import Optional | |
| from app.utils.vectordatabase import retriever | |
| from app.schemas.pydanticschema import LearningRoadmap,SearchCourse | |
| import json | |
| from typing import Dict, List,Any | |
| from pathlib import Path | |
| BASE_DIR = Path(__file__).resolve().parent | |
| def search_courses(query: str): | |
| """ | |
| Search the course catalog for relevant modules based on a skill query | |
| Args: | |
| query:the skill to find with semantic terms (e.g., 'FastAPI', 'PostgreSQL', 'Docker','Enterprise VMS Strategy','Utilization Management'). | |
| """ | |
| results = retriever.invoke( | |
| query | |
| ) | |
| if not results: | |
| return f"No courses found for '{query}'." | |
| formatted_output = [] | |
| for doc in results: | |
| course_id = doc.metadata.get('course_id', 'N/A') | |
| # We include the ID for roadmap generation, followed by the full context | |
| # created during the transformation stage (Title, Desc, Outcomes, Prereqs). | |
| course_block = ( | |
| f"ID: {course_id}\n" | |
| f"{doc.page_content}\n" | |
| "---" | |
| ) | |
| formatted_output.append(course_block) | |
| return "\n".join(formatted_output) | |
| def submit_final_roadmap(candidate_name, target_role, roadmap, onboarding_summary): | |
| """ | |
| STRICTLY call this tool to submit the final structured learning roadmap. | |
| This saves the data to the global system and the graph state. | |
| """ | |
| result = { | |
| "candidate_name": candidate_name, | |
| "target_role": target_role, | |
| "onboarding_summary": onboarding_summary, | |
| "roadmap": [ | |
| step.model_dump() if hasattr(step, "model_dump") else step | |
| for step in roadmap | |
| ] | |
| } | |
| # Return to LangGraph (will be stored in state via a post-processing node) | |
| return result | |
| def submit_mermaid_visualization(mermaid_code: str): | |
| """ | |
| STRICTLY call this tool to save the Mermaid.js visualization of the roadmap. | |
| """ | |
| # 1. Tell Python to use the variable from the outer scope | |
| # 2. Now this assignment updates the global variable | |
| mermaid_roadmap_code = mermaid_code | |
| return "Mermaid visualization stored successfully." | |
| class CourseLookup: | |
| def __init__(self, catalog_path: str = "course_catalog.json"): | |
| self.catalog_path = catalog_path | |
| self.courses_map = {} | |
| self._load_catalog() | |
| def _load_catalog(self): | |
| """Loads the catalog into a dictionary for O(1) lookup speed.""" | |
| try: | |
| with open(self.catalog_path, 'r') as f: | |
| catalog = json.load(f) | |
| # Key the dictionary by course_id for instant retrieval | |
| self.courses_map = {course['course_id']: course for course in catalog} | |
| except FileNotFoundError: | |
| print(f"Error: {self.catalog_path} not found.") | |
| except json.JSONDecodeError: | |
| print(f"Error: Failed to decode {self.catalog_path}.") | |
| def get_course_details(self, course_id: str) -> Optional[Dict[str, Any]]: | |
| """Retrieves full details of a course by its ID.""" | |
| return self.courses_map.get(course_id) | |
| DATA_PATH = BASE_DIR / "Catalog.json" | |
| if DATA_PATH.exists(): | |
| lookup_service = CourseLookup(DATA_PATH) | |
| else: | |
| raise FileNotFoundError(f"Catalog file not found: {DATA_PATH}") | |
| def get_course_by_id(course_id: str) -> str: | |
| """ | |
| Retrieves full details for a specific course using its unique course_id. | |
| Use this tool when you find a prerequisite ID in another course and | |
| need to fetch its title, description, and duration to add to the roadmap. | |
| """ | |
| details = lookup_service.get_course_details(course_id) | |
| if not details: | |
| return f"Error: Course with ID {course_id} not found in catalog." | |
| # Return a clean string for the agent to process | |
| return json.dumps(details, indent=2) | |
| roadmap_planner_agent_tools=[search_courses, get_course_by_id,submit_final_roadmap,submit_mermaid_visualization] | |