Spaces:
Running
Running
File size: 4,008 Bytes
478dec6 | 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 | from pydantic import BaseModel
from typing import Optional, List, Annotated, TypedDict
from langchain_core.messages import BaseMessage, HumanMessage
from langgraph.graph.message import add_messages
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import ToolNode
# tool_node = ToolNode(tools)
from services.llms.LLM import model_4o_2
from services.agents.LLMAgent import LLMAgent
PROMPT_PATH = 'src/prompts/LearningRoadmapPrompt.md'
with open(PROMPT_PATH, encoding='utf-8') as f:
prompt_template = f.read()
class Competency(BaseModel):
skill: str
description: str
resources: List[str]
class Roadmap(BaseModel):
roadmap: List[Competency]
insight: str
class Roadmaps(BaseModel):
roadmaps: List[Roadmap]
class AgentState(TypedDict):
"""State for our agent"""
messages: Annotated[list[BaseMessage], add_messages]
roadmaps: Roadmaps
def gap_analysis(state: AgentState) -> AgentState:
"""Gatekeeper: answer directly"""
roadmap_llm = model_4o_2.with_structured_output(Roadmaps)
gap_analysis_prompt = """\
{user_profile_and_target}
Task: Create 3 options of learning roadmap for the user to achieve their target.
Each roadmap has 2 keys, 'roadmap' and 'insight'. In the 'roadmap' key provide list of 'skill', 'description', and 'resources' for each competency.
The 'insight' key should provide a brief summary of the roadmap and how it relates to the user's profile and target.
"""
user_profile_and_target = state['messages'][-1].content
if not user_profile_and_target:
return {"error": "user_profile_and_target must be provided."}
input_gap_analysis_prompt = gap_analysis_prompt.format(
user_profile_and_target = user_profile_and_target
)
result = roadmap_llm.invoke(input_gap_analysis_prompt)
recommendations = {}
for i, a_roadmap in enumerate(result.roadmaps):
recommendations[f'roadmap_{i+1}'] = {}
recommendations[f'roadmap_{i+1}']['competencies'] = []
recommendations[f'roadmap_{i+1}']['insight'] = a_roadmap.insight
for a_competency in a_roadmap.roadmap:
recommendations[f'roadmap_{i+1}']['competencies'].append({"skill":a_competency.skill, "description": a_competency.description, "resources": a_competency.resources})
return {**state, "roadmaps": recommendations}
class LearningRoadmapAgent(LLMAgent):
def __init__(self):
super().__init__()
self.user_profile: Optional[str] = None
self.user_target: Optional[str] = None
self.thread_id = None
self.session_id = None
self.roadmap_agent_graph = StateGraph(AgentState)
self.roadmap_agent_graph.add_node("gap_analysis", gap_analysis)
self.roadmap_agent_graph.add_edge(START, "gap_analysis")
self.roadmap_agent_graph.add_edge("gap_analysis", END)
self.roadmap_agent = self.roadmap_agent_graph.compile(checkpointer=MemorySaver())
def generate_roadmap(self, user_profile: str = None, user_target: str = None) -> list:
"""
Generate a learning roadmap based on user_profile and user_target.
"""
if user_profile is None or user_target is None:
return [{"error": "user_profile and user_target must be provided."}]
merged_user_profile_target = """
Please create roadmap from following input:
User Profile: {user_profile}
User Target: {user_target}
""".format(user_profile=user_profile, user_target=user_target).replace('{','{{').replace('}','}}')
config = {"configurable": {"thread_id": self.thread_id, "session_id": self.session_id}}
result = self.roadmap_agent.invoke({"messages": [HumanMessage(content=merged_user_profile_target)]}, config)
if result and 'roadmaps' in result:
return result['roadmaps']
else:
return [{"error": "Failed to generate roadmap."}] |