diff --git a/.env.example b/.env.example index cebe38c..4293731 100644 --- a/.env.example +++ b/.env.example @@ -29,6 +29,9 @@ REDIS_URL=redis://localhost:6379/0 OPENAI_API_KEY= OPENAI_MODEL=gpt-4o-mini OPENAI_EMBEDDING_MODEL=text-embedding-3-small +GROQ_API_KEY= +HF_API_KEY= +HF_MODEL=Qwen/Qwen2.5-7B-Instruct GOOGLE_API_KEY= GEMINI_MODEL=gemini-3.6-flash diff --git a/src/backend/app/agents/nodes/recruiter_decision.py b/src/backend/app/agents/nodes/recruiter_decision.py index b0cd49b..4205aa4 100644 --- a/src/backend/app/agents/nodes/recruiter_decision.py +++ b/src/backend/app/agents/nodes/recruiter_decision.py @@ -1,10 +1,94 @@ import asyncio from src.backend.app.agents.nodes.base import BaseNode -from src.backend.app.agents.state import ATSState, RecruiterDecisionOutput, RecruiterDecisionInput +from src.backend.app.agents.state import ( + ATSState, + CandidateContext, + CandidateRecommendation, + RecruiterDecisionInput, + RecruiterDecisionOutput, +) from src.backend.app.services.llm_provider import LLMProvider + class RecruiterDecisionNode(BaseNode): + batch_size = 2 + history_limit = 3 + + @staticmethod + def _candidate_code(candidate_id: str) -> str: + return candidate_id[-4:].upper() + + @classmethod + def _normalize_recommendation( + cls, + recommendation: CandidateRecommendation, + candidates_by_id: dict[str, CandidateContext], + ) -> CandidateRecommendation: + candidate = candidates_by_id.get(recommendation.candidate_id) + full_name = (candidate.full_name if candidate else None) or "Ứng viên" + code = cls._candidate_code(recommendation.candidate_id) + return recommendation.model_copy( + update={ + "candidate_code": code, + "full_name": full_name, + "display_name": f"{full_name} (#{code})", + } + ) + + @staticmethod + def _compact_candidate(candidate: CandidateContext) -> CandidateContext: + return candidate.model_copy( + update={ + "summary": candidate.summary[:400], + "skills": candidate.skills[:8], + "strengths": candidate.strengths[:8], + "weaknesses": candidate.weaknesses[:6], + "experiences": [ + experience.model_copy( + update={ + "company": experience.company[:100], + "position": experience.position[:100], + "duration": experience.duration[:50], + "highlights": [highlight[:180] for highlight in experience.highlights[:2]], + } + ) + for experience in candidate.experiences[:3] + ], + "github_summary": ( + candidate.github_summary[:400] + if candidate.github_summary + else None + ), + "linkedin_summary": ( + candidate.linkedin_summary[:400] + if candidate.linkedin_summary + else None + ), + } + ) + + @staticmethod + def _compact_mission(mission): + return mission.model_copy( + update={ + "objective": mission.objective[:1000], + "plan": [step[:160] for step in mission.plan[:5]], + } + ) + + @staticmethod + def _compact_history(history): + return [ + action.model_copy( + update={ + "action": action.action[:160], + "decision": action.decision[:300] if action.decision else None, + } + ) + for action in history + ] + def __init__(self, llm_provider: LLMProvider): super().__init__() self.llm_provider = llm_provider @@ -14,8 +98,8 @@ async def execute(self, state: ATSState) -> ATSState: if not state.candidate_search.candidates: state.candidate_search.final_decision = RecruiterDecisionOutput( - recommendations=[], - final_summary="No candidates were returned by the retrieval pipeline.", + candidates=[], + summary="No candidates were returned by the retrieval pipeline.", ) self.record_action( state=state, @@ -24,19 +108,48 @@ async def execute(self, state: ATSState) -> ATSState: ) return state - # 1. Read state (Fix các biến bị trỏ sai) - decision_input = RecruiterDecisionInput( - mission=state.candidate_search.mission, # Đã fix - candidates=state.candidate_search.candidates, - history=state.candidate_search.action_history # Đã fix naming + # Evaluate small batches so one large candidate set cannot exceed the + # provider's per-request or tokens-per-minute budget. + candidates = [ + self._compact_candidate(candidate) + for candidate in state.candidate_search.candidates + ] + mission = self._compact_mission(state.candidate_search.mission) + history = self._compact_history( + state.candidate_search.action_history[-self.history_limit :] ) - - # 2. Call LLM - decision_output = await asyncio.to_thread( - self.llm_provider.invoke, - recruiter_prompt, - decision_input, - RecruiterDecisionOutput, + recommendations: list[CandidateRecommendation] = [] + + for start in range(0, len(candidates), self.batch_size): + decision_input = RecruiterDecisionInput( + mission=mission, + candidates=candidates[start : start + self.batch_size], + history=history, + ) + batch_output = await asyncio.to_thread( + self.llm_provider.invoke, + recruiter_prompt, + decision_input, + RecruiterDecisionOutput, + ) + recommendations.extend( + self._normalize_recommendation(item, {candidate.candidate_id: candidate for candidate in candidates}) + for item in batch_output.candidates + ) + + recommendations.sort(key=lambda item: item.confidence, reverse=True) + top_recommendations = recommendations[:5] + recommendation_summary = "; ".join( + f"{item.display_name}: {item.recommendation} ({item.confidence:.0%})" + for item in top_recommendations + ) + decision_output = RecruiterDecisionOutput( + candidates=recommendations, + summary=( + f"Evaluated {len(candidates)} candidates in " + f"{(len(candidates) + self.batch_size - 1) // self.batch_size} batches. " + f"Top matches: {recommendation_summary or 'No recommendations returned.'}" + ), ) # 3. Update state diff --git a/src/backend/app/agents/nodes/retrieval.py b/src/backend/app/agents/nodes/retrieval.py index 19833cb..4440c45 100644 --- a/src/backend/app/agents/nodes/retrieval.py +++ b/src/backend/app/agents/nodes/retrieval.py @@ -57,6 +57,7 @@ def get_val(key: str, default: str | list | None = None) -> str | list | None: mapped_candidates.append( CandidateContext( candidate_id=str(get_val("candidate_id", "")), + full_name=get_val("full_name"), semantic_score=float(raw_score), summary=str(get_val("summary", "")), skills=list(get_val("skills", []) or []), diff --git a/src/backend/app/agents/prompts/planner_prompt.md b/src/backend/app/agents/prompts/planner_prompt.md index 77bfe6b..2152f76 100644 --- a/src/backend/app/agents/prompts/planner_prompt.md +++ b/src/backend/app/agents/prompts/planner_prompt.md @@ -12,7 +12,7 @@ You will receive a structured input containing: # Instructions 1. **Analyze Query Clarity**: - Determine if the request has enough detail to construct a meaningful search query. - - If information is missing or ambiguous, set `query_assessment.clarification_detail.status` to `"needed"` or `"not_enough"` and provide ONE concise question in `suggestion`. + - If information is missing or ambiguous, set `query_assessment.clarification.status` to `"not_enough"` and provide ONE concise question in `query_assessment.clarification.question`. 2. **Formulate Search Requirements**: - If the query is clear OR if you are revising search terms based on `reflection` feedback, generate a structured `search_requirement`. @@ -25,4 +25,6 @@ You will receive a structured input containing: # Constraints - NEVER search candidates yourself. - NEVER invent information not present in user query or context. -- Output MUST be valid JSON conforming strictly to the `PlannerOutput` schema. \ No newline at end of file +- Output MUST be valid JSON conforming strictly to the `PlannerOutput` schema. +- Use `query_assessment.clarification.status` with exactly `"enough"` or `"not_enough"`. +- Do not use fields named `clarification_detail`, `needed`, or `suggestion` inside `query_assessment`. \ No newline at end of file diff --git a/src/backend/app/agents/prompts/recruiter_prompt.md b/src/backend/app/agents/prompts/recruiter_prompt.md index e7e1b13..727f706 100644 --- a/src/backend/app/agents/prompts/recruiter_prompt.md +++ b/src/backend/app/agents/prompts/recruiter_prompt.md @@ -5,7 +5,7 @@ Your task is to analyze the retrieved top candidates against the hiring requirem # Context Provided You will receive: - `mission`: The hiring requirement and target profile. -- `candidates`: List of retrieved candidate profiles (including skills, experience, resume summary, GitHub/LinkedIn insights, and match scores). +- `candidates`: A small batch of retrieved candidate profiles (including skills, experience, resume summary, GitHub/LinkedIn insights, and match scores). - `history`: Actions taken to arrive at these candidates. # Instructions @@ -14,8 +14,9 @@ You will receive: - Focus on overall **hiring quality**, practical skill alignment, and experience depth—NOT just the raw semantic matching score. 2. **Generate Final Recommendations**: - - Select and rank the best candidates. + - Evaluate every candidate in this batch and rank them relative to the other candidates in this batch. - For each recommended candidate, provide: + - The candidate's full name when available, plus a candidate code made from the last four characters of the candidate ID. - Clear rationale on WHY they fit. - Any missing requirements or potential risks/gaps (e.g., lower experience in one specific skill, job-hopping history). @@ -25,4 +26,8 @@ You will receive: # Constraints - NEVER invent or assume candidate qualifications not provided in the data. - NEVER compare candidates using unprovided external knowledge. +- Return recommendations only for candidates present in this batch. +- Return `candidate_id` exactly as provided. Do not include CV URLs or UUIDs in `reasoning`. +- Return `key_strengths`, `missing_requirements`, and `risks` as short list items. +- The output fields are `summary` and `candidates`; each candidate includes `candidate_id`, `candidate_code`, `display_name`, `full_name`, `recommendation`, `confidence`, `reasoning`, `key_strengths`, `missing_requirements`, and `risks`. - Output MUST be valid JSON conforming strictly to the `RecruiterDecisionOutput` schema. \ No newline at end of file diff --git a/src/backend/app/agents/prompts/reflection_prompt.md b/src/backend/app/agents/prompts/reflection_prompt.md index edc2629..3df73a9 100644 --- a/src/backend/app/agents/prompts/reflection_prompt.md +++ b/src/backend/app/agents/prompts/reflection_prompt.md @@ -18,9 +18,18 @@ You will receive: - Set `retry` to `true` if search results are unsatisfactory and another search attempt is needed. - Set `retry` to `false` if results are good enough to proceed to final candidate evaluation. - Provide a clear, analytical `reason` explaining your decision. - - If `retry` is `true`, provide concrete `suggested_modifications` (e.g., "Drop location hard filter", "Remove rare framework from mandatory skills"). + - If `retry` is `true`, put concrete search modifications in `suggestion` (e.g., "Drop location hard filter", "Remove rare framework from mandatory skills"). + - If `retry` is `false`, set `suggestion` to `null` or a short next-step suggestion. # Constraints - Do NOT request or evaluate individual candidate resumes or full candidate profiles. Rely ONLY on `observation` metrics. - Do NOT perform retrieval or ranking. -- Output MUST be valid JSON conforming strictly to the `ReflectionOutput` schema. \ No newline at end of file +- Output MUST be one non-empty JSON object and nothing else. Use exactly this shape: + { + "reflection": { + "retry": false, + "reason": "The search result is sufficient for final evaluation.", + "suggestion": null + } + } +- Do not add fields such as `suggested_modifications`. \ No newline at end of file diff --git a/src/backend/app/agents/router.py b/src/backend/app/agents/router.py index 2482ae2..846958c 100644 --- a/src/backend/app/agents/router.py +++ b/src/backend/app/agents/router.py @@ -1,6 +1,189 @@ from __future__ import annotations -from src.backend.app.agents.state import ATSState +import asyncio +import json +import logging +from collections.abc import AsyncIterator +from typing import Annotated +from uuid import UUID + +from fastapi import APIRouter, Depends, HTTPException, status +from fastapi.responses import StreamingResponse +from groq import ( + APIConnectionError, + APIError, + APITimeoutError, + AuthenticationError, + BadRequestError, + GroqError, + RateLimitError, +) +from modules.auth.domain.models import AuthUser +from modules.search.application.search_service import get_embedding_service +from modules.search.infra.legacy_bridge import ( + CandidateSearchRepository, + CandidateSearchService, + EnrichmentRepository, + RankingService, +) +from modules.shared.infrastructure.auth_dependencies import require_operational_roles +from modules.shared.infrastructure.config import Settings, get_settings +from modules.shared.infrastructure.rate_limit import agent_rate_limit +from modules.shared.infrastructure.supabase_client import get_supabase_admin_client +from pydantic import BaseModel, Field + +from src.backend.app.agents.nodes.interaction import CLIInteractionGateway +from src.backend.app.agents.state import ( + ATSState, + CandidateSearchState, + RecruiterDecisionOutput, + Mission, + MissionStatus, +) +from src.backend.app.services.llm_provider import ( + FallbackLLMProvider, + GroqProvider, + HFProvider, +) + +router = APIRouter(prefix="/agents", tags=["agents"]) +logger = logging.getLogger(__name__) + + +class AgentContext(BaseModel): + current_page: str = Field(default="candidates_dashboard", max_length=100) + user_id: str = Field(min_length=1, max_length=200) + + +class AgentChatRequest(BaseModel): + message: str = Field(min_length=1, max_length=12000) + conversation_id: UUID + context: AgentContext + + +def _agent_graph(settings: Settings): + # Import lazily because graph.py imports the route decision functions from + # this module while the application is importing the router. + from src.backend.app.agents.graph import build_graph + + client = get_supabase_admin_client(settings) + if client is None: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="Agent search is unavailable: the database is not configured.", + ) + + search_service = CandidateSearchService( + search_repository=CandidateSearchRepository(client), + enrichment_repository=EnrichmentRepository(client), + embedding_service=get_embedding_service(), + ranking_service=RankingService(), + ) + return build_graph( + llm_provider=FallbackLLMProvider( + primary=GroqProvider(), + fallback=HFProvider(), + ), + search_service=search_service, + interaction_gateway=CLIInteractionGateway(), + ) + + +def _sse(event: str, payload: dict) -> str: + return f"event: {event}\ndata: {json.dumps(payload, ensure_ascii=False)}\n\n" + + +def _agent_error_message(error: Exception) -> str: + if isinstance(error, HTTPException): + return str(error.detail) + if isinstance(error, RateLimitError): + return "The AI service is rate-limited. Please try again in a moment." + if isinstance(error, (AuthenticationError, BadRequestError)): + return "The AI service is not configured correctly. Please contact an administrator." + if isinstance(error, (APITimeoutError, APIConnectionError, TimeoutError, asyncio.TimeoutError)): + return "The AI service did not respond in time. Please try again." + if isinstance(error, APIError): + return "The AI service is temporarily unavailable. Please try again." + if isinstance(error, GroqError): + return "The AI service is not configured correctly. Please contact an administrator." + return "The agent could not complete this request." + + +def _extract_final_decision(value: object): + if isinstance(value, dict): + if "final_decision" in value: + return value["final_decision"] + for nested in value.values(): + decision = _extract_final_decision(nested) + if decision is not None: + return decision + if isinstance(value, (list, tuple)): + for nested in value: + decision = _extract_final_decision(nested) + if decision is not None: + return decision + decision = getattr(value, "final_decision", None) + if decision is not None: + return decision + candidate_search = getattr(value, "candidate_search", None) + if candidate_search is not None: + return _extract_final_decision(candidate_search) + return None + + +async def _stream_agent(request: AgentChatRequest, settings: Settings) -> AsyncIterator[str]: + try: + graph = _agent_graph(settings) + initial_state = ATSState( + messages=[request.message], + candidate_search=CandidateSearchState( + mission=Mission( + objective=request.message, + current_step="Planner Assessment", + status=MissionStatus.PENDING, + ) + ), + ) + + yield _sse("status", {"message": "Agent is thinking..."}) + final_state = None + async for update in graph.astream(initial_state, stream_mode="updates"): + final_state = update + for node_name in update: + yield _sse("status", {"message": f"Completed {node_name}"}) + + if not final_state: + raise RuntimeError("Agent completed without a result") + + decision = _extract_final_decision(final_state) + if decision is None: + result = {"summary": "The agent completed without a recommendation.", "candidates": []} + elif isinstance(decision, dict): + result = RecruiterDecisionOutput.model_validate(decision).model_dump(mode="json") + else: + result = decision.model_dump(mode="json") + + yield _sse( + "done", + {"conversation_id": str(request.conversation_id), "result": result}, + ) + except Exception as exc: + logger.exception("Agent stream failed", extra={"error_type": type(exc).__name__}) + yield _sse("error", {"message": _agent_error_message(exc)}) + + +@router.post("", response_class=StreamingResponse) +async def chat_with_agent( + request: AgentChatRequest, + _user: Annotated[AuthUser, Depends(require_operational_roles())], + settings: Annotated[Settings, Depends(get_settings)], + _limit: Annotated[None, Depends(agent_rate_limit)] = None, +) -> StreamingResponse: + return StreamingResponse( + _stream_agent(request, settings), + media_type="text/event-stream", + headers={"Cache-Control": "no-cache", "Connection": "keep-alive"}, + ) def route_after_planner(state: ATSState) -> str: diff --git a/src/backend/app/agents/state.py b/src/backend/app/agents/state.py index 20054d2..fed0459 100644 --- a/src/backend/app/agents/state.py +++ b/src/backend/app/agents/state.py @@ -5,7 +5,7 @@ from typing import Any from typing_extensions import Literal -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, model_validator """ ========================================================== @@ -311,6 +311,8 @@ class CandidateContext(BaseModel): candidate_id: str + full_name: str | None = None + semantic_score: float summary: str @@ -346,6 +348,12 @@ class CandidateRecommendation(BaseModel): candidate_id: str + candidate_code: str = "" + + display_name: str = "" + + full_name: str = "" + recommendation: Literal[ "Strong Hire", "Hire", @@ -353,7 +361,7 @@ class CandidateRecommendation(BaseModel): "Reject", ] - confidence: float + confidence: float = Field(ge=0.0, le=1.0) reasoning: str @@ -363,12 +371,40 @@ class CandidateRecommendation(BaseModel): risks: list[str] + @model_validator(mode="after") + def normalize_identity(self) -> "CandidateRecommendation": + code = self.candidate_id[-4:].upper() + self.candidate_code = code + if not self.full_name: + self.full_name = "Ứng viên" + self.display_name = f"{self.full_name} (#{code})" + return self + class RecruiterDecisionOutput(BaseModel): - recommendations: list[CandidateRecommendation] + summary: str = "" + + candidates: list[CandidateRecommendation] = Field(default_factory=list) + + @model_validator(mode="before") + @classmethod + def accept_legacy_field_names(cls, values: Any) -> Any: + if isinstance(values, dict): + values = dict(values) + if "candidates" not in values and "recommendations" in values: + values["candidates"] = values.pop("recommendations") + if "summary" not in values and "final_summary" in values: + values["summary"] = values.pop("final_summary") + return values + + @property + def recommendations(self) -> list[CandidateRecommendation]: + return self.candidates - final_summary: str + @property + def final_summary(self) -> str: + return self.summary # ========================================================== diff --git a/src/backend/app/services/llm_provider.py b/src/backend/app/services/llm_provider.py index a4c9d9b..1962f1a 100644 --- a/src/backend/app/services/llm_provider.py +++ b/src/backend/app/services/llm_provider.py @@ -1,16 +1,19 @@ from __future__ import annotations -from abc import ABC, abstractmethod -from typing import Any - import json +import logging import os +from abc import ABC, abstractmethod +from typing import Any from dotenv import load_dotenv from groq import Groq +from groq import RateLimitError +from openai import OpenAI from pydantic import BaseModel load_dotenv() +logger = logging.getLogger(__name__) class LLMProvider(ABC): @@ -34,12 +37,27 @@ def invoke( ) -> Any: pass + def generate_text( + self, + prompt: str, + temperature: float = 0.1, + ) -> str: + """Compatibility helper for services that only need plain text.""" + result = self.invoke( + system_prompt="You are a helpful assistant.", + user_input=prompt, + temperature=temperature, + ) + if not isinstance(result, str): + raise TypeError("LLM provider returned a non-text response") + return result + class GroqProvider(LLMProvider): def __init__( self, - model: str = "llama-3.1-8b-instant", + model: str = "openai/gpt-oss-20b", ): self.client = Groq( api_key=os.getenv("GROQ_API_KEY") @@ -89,11 +107,8 @@ def invoke( # ---------------------------- if response_model is not None: - schema = json.dumps( - response_model.model_json_schema(), - ensure_ascii=False, - indent=2, - ) + schema = response_model.model_json_schema() + schema_text = json.dumps(schema, ensure_ascii=False, indent=2) system_prompt = ( f"{system_prompt}\n\n" @@ -101,7 +116,7 @@ def invoke( "Do NOT explain.\n" "Do NOT wrap in markdown.\n" "The JSON MUST follow this schema:\n" - f"{schema}" + f"{schema_text}" ) # ---------------------------- @@ -111,7 +126,11 @@ def invoke( if response_model is not None: request_kwargs["response_format"] = { - "type": "json_object" + "type": "json_schema", + "json_schema": { + "name": response_model.__name__, + "schema": schema, + }, } response = self.client.chat.completions.create( @@ -142,3 +161,114 @@ def invoke( # Structured Output # ---------------------------- return response_model.model_validate_json(content) + + +class HFProvider(LLMProvider): + """Hugging Face Router provider using its OpenAI-compatible endpoint.""" + + def __init__( + self, + model: str | None = None, + api_key: str | None = None, + ): + self._api_key = api_key or os.getenv("HF_API_KEY") + self._client: OpenAI | None = None + self.model = model or os.getenv( + "HF_MODEL", "Qwen/Qwen2.5-7B-Instruct" + ) + + @property + def client(self) -> OpenAI: + if self._client is None: + self._client = OpenAI( + api_key=self._api_key, + base_url="https://router.huggingface.co/v1", + ) + return self._client + + def invoke( + self, + system_prompt: str, + user_input: Any, + response_model: type[BaseModel] | None = None, + temperature: float = 0.1, + ) -> Any: + if isinstance(user_input, BaseModel): + user_content = user_input.model_dump_json(indent=2) + else: + user_content = str(user_input) + + request_prompt = system_prompt + request_kwargs: dict[str, Any] = {} + if response_model is not None: + schema = response_model.model_json_schema() + request_prompt = ( + f"{system_prompt}\n\n" + "Return ONLY a valid JSON object. Do NOT use markdown. " + "The JSON MUST follow this schema:\n" + f"{json.dumps(schema, ensure_ascii=False, indent=2)}" + ) + request_kwargs["response_format"] = {"type": "json_object"} + + response = self.client.chat.completions.create( + model=self.model, + messages=[ + {"role": "system", "content": request_prompt}, + {"role": "user", "content": user_content}, + ], + temperature=temperature, + **request_kwargs, + ) + content = response.choices[0].message.content or "" + if response_model is None: + return content + return response_model.model_validate_json(content) + + +HF_Provider = HFProvider + + +def _is_groq_rate_limit(error: Exception) -> bool: + if isinstance(error, RateLimitError): + return True + if getattr(error, "status_code", None) == 429: + return True + if getattr(error, "code", None) == "rate_limit_exceeded": + return True + response = getattr(error, "response", None) + return getattr(response, "status_code", None) == 429 + + +class FallbackLLMProvider(LLMProvider): + """Use the fallback provider only for an upstream Groq rate limit.""" + + def __init__(self, primary: LLMProvider, fallback: LLMProvider): + self.primary = primary + self.fallback = fallback + + def invoke( + self, + system_prompt: str, + user_input: Any, + response_model: type[BaseModel] | None = None, + temperature: float = 0.1, + ) -> Any: + try: + return self.primary.invoke( + system_prompt=system_prompt, + user_input=user_input, + response_model=response_model, + temperature=temperature, + ) + except Exception as error: + if not _is_groq_rate_limit(error): + raise + logger.warning( + "Groq Rate Limit exceeded. Switching to Hugging Face Provider (Qwen 7B)..." + ) + return self.fallback.invoke( + system_prompt=system_prompt, + user_input=user_input, + response_model=response_model, + temperature=temperature, + ) diff --git a/src/backend/apps/main.py b/src/backend/apps/main.py index 55f2580..f472f52 100644 --- a/src/backend/apps/main.py +++ b/src/backend/apps/main.py @@ -17,6 +17,7 @@ from modules.review.adapters.routes import router as review_router from modules.search.adapters.routes import router as search_router from modules.scoring.adapters.routes import router as scoring_router +from src.backend.app.agents.router import router as agents_router from modules.shared.infrastructure.config import get_settings structlog.configure( @@ -60,6 +61,7 @@ app.include_router(review_router) app.include_router(search_router) app.include_router(scoring_router) +app.include_router(agents_router) @app.get("/health") diff --git a/src/backend/modules/shared/infrastructure/rate_limit.py b/src/backend/modules/shared/infrastructure/rate_limit.py index b9e7098..ccfdf63 100644 --- a/src/backend/modules/shared/infrastructure/rate_limit.py +++ b/src/backend/modules/shared/infrastructure/rate_limit.py @@ -105,3 +105,5 @@ def reset(self) -> None: #: Nộp CV công khai — mỗi lượt là một file tới 10MB đi vào Azure và một lượt #: gọi LLM. Không giới hạn thì một script biến nó thành hoá đơn. ingest_rate_limit = RateLimit("ingestion.ingest", limit=20, seconds=3600) + +agent_rate_limit = RateLimit("agents.chat", limit=20, seconds=600) diff --git a/src/frontend/app/providers.tsx b/src/frontend/app/providers.tsx index e5000c0..7baea6d 100644 --- a/src/frontend/app/providers.tsx +++ b/src/frontend/app/providers.tsx @@ -5,6 +5,8 @@ import { AuthProvider } from "@/contexts/AuthContext"; import { AuthGuard } from "@/components/AuthGuard"; import { WorkspaceProvider } from "@/contexts/WorkspaceContext"; import { RequireCalendarModal } from "@/components/RequireCalendarModal"; +import { AgentChatProvider } from "@/hooks/useAgentChat"; +import { AgentChatDrawer } from "@/components/AgentChatDrawer"; export function Providers({ children }: { children: React.ReactNode }) { const clientId = process.env.NEXT_PUBLIC_GOOGLE_CLIENT_ID; @@ -14,8 +16,11 @@ export function Providers({ children }: { children: React.ReactNode }) { - {children} - + + {children} + + + {/* */} diff --git a/src/frontend/components/AgentChatDrawer.tsx b/src/frontend/components/AgentChatDrawer.tsx new file mode 100644 index 0000000..ca62ad4 --- /dev/null +++ b/src/frontend/components/AgentChatDrawer.tsx @@ -0,0 +1,131 @@ +"use client"; + +import { FormEvent, useEffect, useRef, useState } from "react"; +import { Bot, ChevronDown, ExternalLink, Loader2, Send, X } from "lucide-react"; +import { useRouter } from "next/navigation"; +import { useAgentChat } from "../hooks/useAgentChat"; +import { useAuth } from "../contexts/AuthContext"; +import { D } from "../lib/shared"; + +function MarkdownMessage({ content }: { content: string }) { + const renderInline = (text: string) => text.split(/(\*\*[^*]+\*\*|`[^`]+`)/g).map((part, index) => { + if (part.startsWith("**") && part.endsWith("**")) return {part.slice(2, -2)}; + if (part.startsWith("`") && part.endsWith("`")) return {part.slice(1, -1)}; + return {part}; + }); + + const renderText = (text: string) => { + const blocks: React.ReactNode[] = []; + let listItems: string[] = []; + const flushList = () => { + if (!listItems.length) return; + blocks.push(); + listItems = []; + }; + text.split("\n").forEach((line, index) => { + if (/^\s*[-*]\s+/.test(line)) { + listItems.push(line.replace(/^\s*[-*]\s+/, "")); + return; + } + flushList(); + if (!line.trim()) return; + if (line.startsWith("#")) { + blocks.push({renderInline(line.replace(/^#+\s*/, ""))}); + return; + } + blocks.push({renderInline(line)}); + }); + flushList(); + return blocks; + }; + + return ( +
+ {content.split(/(```[\s\S]*?```)/g).map((part, index) => part.startsWith("```") ? ( + 
{part.replace(/^```\w*\n?/, "").replace(/```$/, "")}
+ ) : {renderText(part)})} +
+ ); +} + +type CandidateResult = { + candidate_id: string; + candidate_code: string; + display_name: string; + recommendation: "Strong Hire" | "Hire" | "Consider" | "Reject"; + confidence: number; + reasoning: string; + key_strengths: string[]; + missing_requirements: string[]; + risks: string[]; +}; + +type AgentResult = { summary: string; candidates: CandidateResult[] }; + +function parseAgentResult(content: string): AgentResult | null { + try { + const result = JSON.parse(content) as AgentResult; + return typeof result.summary === "string" && Array.isArray(result.candidates) ? result : null; + } catch { + return null; + } +} + +function CandidateCards({ result }: { result: AgentResult }) { + const router = useRouter(); + return
+
{result.summary}
+ {result.candidates.map((candidate) => { + const tone = candidate.recommendation === "Reject" ? D.red : candidate.recommendation === "Consider" ? D.amber : D.mint; + const concerns = [...candidate.missing_requirements, ...candidate.risks]; + return
+
{candidate.display_name || `Ứng viên (#${candidate.candidate_code})`}{candidate.recommendation}
+
Confidence: {(candidate.confidence * 100).toFixed(0)}%
+
{candidate.reasoning}
+ {candidate.key_strengths.length > 0 &&
{candidate.key_strengths.map((item) => {item})}
} + {concerns.length > 0 && } + +
; + })} +
; +} + +function MessageContent({ message }: { message: { role: "assistant" | "user" | "error"; content: string } }) { + const result = message.role === "assistant" ? parseAgentResult(message.content) : null; + return result ? : ; +} + +export function AgentChatDrawer() { + const { user } = useAuth(); + const { messages, isLoading, isOpen, setIsOpen, sendMessage, retry } = useAgentChat(); + const [input, setInput] = useState(""); + const bottomRef = useRef(null); + useEffect(() => { + bottomRef.current?.scrollIntoView({ behavior: "smooth" }); + }, [messages]); + if (!user) return null; + + async function submit(event: FormEvent) { + event.preventDefault(); + const message = input.trim(); + if (!message || isLoading) return; + setInput(""); + await sendMessage(message); + } + + return ( + <> + {!isOpen && } + {isOpen && } + + ); +} \ No newline at end of file diff --git a/src/frontend/hooks/useAgentChat.tsx b/src/frontend/hooks/useAgentChat.tsx new file mode 100644 index 0000000..f6a8101 --- /dev/null +++ b/src/frontend/hooks/useAgentChat.tsx @@ -0,0 +1,114 @@ +"use client"; + +import { createContext, useContext, useEffect, useMemo, useState } from "react"; +import { streamClient } from "../services/httpClient"; +import { useAuth } from "../contexts/AuthContext"; + +export type AgentMessage = { + id: string; + role: "user" | "assistant" | "error"; + content: string; + retryMessage?: string; +}; + +type AgentChatContextValue = { + messages: AgentMessage[]; + isLoading: boolean; + isOpen: boolean; + conversationId: string; + sendMessage: (message: string) => Promise; + retry: (message: string) => Promise; + setIsOpen: (open: boolean) => void; +}; + +const STORAGE_KEY = "smartats_agent_chat"; +const AgentChatContext = createContext(null); + +function makeId(): string { + return typeof crypto !== "undefined" && "randomUUID" in crypto + ? crypto.randomUUID() + : `${Date.now()}-${Math.random()}`; +} + +function readStoredState(): Pick { + if (typeof window === "undefined") return { messages: [], conversationId: makeId(), isOpen: false }; + try { + const stored = JSON.parse(localStorage.getItem(STORAGE_KEY) ?? "null"); + if (stored?.conversationId && Array.isArray(stored.messages)) return stored; + } catch { + // Corrupt local history should not prevent the application from opening. + } + return { messages: [], conversationId: makeId(), isOpen: false }; +} + +export function AgentChatProvider({ children }: { children: React.ReactNode }) { + const { user } = useAuth(); + const initial = useMemo(readStoredState, []); + const [messages, setMessages] = useState(initial.messages); + const [conversationId] = useState(initial.conversationId); + const [isOpen, setIsOpen] = useState(initial.isOpen); + const [isLoading, setIsLoading] = useState(false); + + useEffect(() => { + localStorage.setItem(STORAGE_KEY, JSON.stringify({ messages, conversationId, isOpen })); + }, [messages, conversationId, isOpen]); + + async function sendMessage(message: string) { + const trimmed = message.trim(); + if (!trimmed || isLoading) return; + const assistantId = makeId(); + setMessages((current) => [ + ...current, + { id: makeId(), role: "user", content: trimmed }, + { id: assistantId, role: "assistant", content: "" }, + ]); + setIsLoading(true); + try { + const response = await streamClient("/agents", { + method: "POST", + body: JSON.stringify({ + message: trimmed, + conversation_id: conversationId, + context: { current_page: window.location.pathname, user_id: user?.id ?? "unknown" }, + }), + }); + const reader = response.body!.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + while (true) { + const { value, done } = await reader.read(); + if (done) break; + buffer += decoder.decode(value, { stream: true }); + const events = buffer.split("\n\n"); + buffer = events.pop() ?? ""; + for (const event of events) { + const dataLine = event.split("\n").find((line) => line.startsWith("data: ")); + if (!dataLine) continue; + const data = JSON.parse(dataLine.slice(6)) as { text?: string; message?: string; result?: unknown }; + if (event.includes("event: delta") && data.text) { + setMessages((current) => current.map((item) => item.id === assistantId ? { ...item, content: item.content + data.text } : item)); + } + if (event.includes("event: error")) throw new Error(data.message ?? "Agent request failed"); + if (event.includes("event: done") && data.result) { + setMessages((current) => current.map((item) => item.id === assistantId + ? { ...item, content: JSON.stringify(data.result) } + : item)); + } + } + } + } catch (error) { + setMessages((current) => current.map((item) => item.id === assistantId ? { id: item.id, role: "error", content: error instanceof Error ? error.message : "Agent request failed", retryMessage: trimmed } : item)); + } finally { + setIsLoading(false); + } + } + + const value = useMemo(() => ({ messages, isLoading, isOpen, conversationId, sendMessage, retry: sendMessage, setIsOpen }), [messages, isLoading, isOpen, conversationId]); + return {children}; +} + +export function useAgentChat(): AgentChatContextValue { + const value = useContext(AgentChatContext); + if (!value) throw new Error("useAgentChat must be used inside "); + return value; +} \ No newline at end of file diff --git a/src/frontend/services/httpClient.ts b/src/frontend/services/httpClient.ts index fdc2beb..bb68c9d 100644 --- a/src/frontend/services/httpClient.ts +++ b/src/frontend/services/httpClient.ts @@ -160,6 +160,38 @@ async function getValidAccessToken(): Promise { return null; } +export async function streamClient( + path: string, + options: RequestInit = {}, +): Promise { + const accessToken = await getValidAccessToken(); + const headers = new Headers(options.headers); + headers.set("Content-Type", "application/json"); + if (accessToken) headers.set("Authorization", `Bearer ${accessToken}`); + + let response = await fetch(`${API_BASE_URL}${path}`, { ...options, headers }); + if (response.status === 401) { + const newToken = await refreshAccessTokenOnce(); + if (newToken) { + headers.set("Authorization", `Bearer ${newToken}`); + response = await fetch(`${API_BASE_URL}${path}`, { ...options, headers }); + } + if (response.status === 401) notifySessionExpired(); + } + if (!response.ok) { + let message = `Request failed (${response.status})`; + try { + const body = (await response.json()) as { detail?: string; message?: string }; + message = body.detail ?? body.message ?? message; + } catch { + /* Keep the HTTP status when a stream endpoint has no JSON error body. */ + } + throw new ApiError(message, response.status); + } + if (!response.body) throw new Error("Agent response did not include a stream"); + return response; +} + function buildHeaders( initHeaders: HeadersInit | undefined, accessToken: string | null, diff --git a/tests/manual/run_e2e_agent.py b/tests/manual/run_e2e_agent.py index 9747daa..843ca2d 100644 --- a/tests/manual/run_e2e_agent.py +++ b/tests/manual/run_e2e_agent.py @@ -6,7 +6,7 @@ from collections.abc import Mapping from pathlib import Path -ROOT_DIR = Path(__file__).resolve().parents[1] +ROOT_DIR = Path(__file__).resolve().parents[2] SRC_DIR = ROOT_DIR / "src" for path in (SRC_DIR, ROOT_DIR): @@ -156,7 +156,7 @@ async def run_e2e_test() -> None: min_similarity=0.5, ) - llm_provider = GroqProvider(model="llama-3.1-8b-instant") + llm_provider = GroqProvider() interaction_gateway = CLIInteractionGateway() app = build_graph(