Spaces:
Sleeping
Sleeping
File size: 4,119 Bytes
4ec07e1 fd01f65 4ec07e1 9bc7a47 4ec07e1 2f60860 4ec07e1 2f60860 4ec07e1 2f60860 4ec07e1 2f60860 4ec07e1 2f60860 4ec07e1 2f60860 4ec07e1 2f60860 4ec07e1 2f60860 4ec07e1 fd01f65 4ec07e1 9bc7a47 4ec07e1 | 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 | 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
@tool
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)
@tool(args_schema=LearningRoadmap)
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
@tool(args_schema=SearchCourse)
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}")
@tool
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]
|