Spaces:
Sleeping
Sleeping
File size: 2,984 Bytes
4ec07e1 602f88e 4ec07e1 9bc7a47 602f88e 4ec07e1 602f88e 2f60860 602f88e 2f60860 4ec07e1 2f60860 4ec07e1 2f60860 4ec07e1 2f60860 4ec07e1 2f60860 4ec07e1 2f60860 4ec07e1 602f88e 4ec07e1 602f88e 4ec07e1 602f88e 4ec07e1 602f88e 4ec07e1 9bc7a47 4ec07e1 602f88e 4ec07e1 602f88e 4ec07e1 602f88e 4ec07e1 602f88e 4ec07e1 602f88e | 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 | from langchain_core.tools import tool
from typing import Optional
from app.vectordatabase.pinecone import retriever
from app.schemas.roadmap_agent_tools_argschema import LearningRoadmap,SearchCourse
import json
from typing import Dict, List,Any
from pathlib import Path
from typing import Annotated
from langchain_core.messages import ToolMessage
from langgraph.types import Command
from langchain_core.tools import InjectedToolCallId
BASE_DIR = Path(__file__).resolve().parent
@tool(args_schema=SearchCourse)
def search_courses(query: str):
"""
Search the course catalog for relevant modules based on a skill query
"""
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')
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,
tool_call_id: Annotated[str, InjectedToolCallId] # Injected automatically
):
"""
STRICTLY call this tool to submit the final structured learning roadmap.
"""
# 1. Construct the structured JSON
result_data = {
"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
]
}
# 2. Return Command to update "final_roadmap" in state
return Command(
update={
"final_roadmap": result_data, # This updates your state key
"messages": [
ToolMessage(
content="SUCCESS: Final roadmap has been submitted and saved to state.",
tool_call_id=tool_call_id
)
]
}
)
@tool
def submit_mermaid_visualization(
mermaid_code: str,
tool_call_id: Annotated[str, InjectedToolCallId] # Automatically injected by ToolNode
):
"""
STRICTLY call this tool to save the Mermaid.js visualization of the roadmap.
"""
if not mermaid_code:
return "No Mermaid visualization found."
# Return Command to update the state key 'mermaid_code'
return Command(
update={
"mermaid_code": mermaid_code, # Updates your graph state
"messages": [
ToolMessage(
content="SUCCESS: Mermaid visualization saved successfully.",
tool_call_id=tool_call_id
)
]
}
)
roadmap_planner_agent_tools=[search_courses,submit_final_roadmap,submit_mermaid_visualization]
|