113 lines
4.5 KiB
Python
113 lines
4.5 KiB
Python
#nodes.py
|
|
from app.agents.pm_agent import run_pm_agent
|
|
from app.agents.dev_agent import run_dev_agent
|
|
from app.agents.qa_agent import run_qa_agent
|
|
from app.services.retrieval_service import find_existing_project
|
|
from app.graph.state import WorkflowState
|
|
|
|
async def pm_node(state: WorkflowState):
|
|
history = state.get("chat_history", []) or []
|
|
|
|
if state.get("status") == "spec_incomplete" and state.get("user_feedback"):
|
|
current_input = state["user_feedback"]
|
|
full_user_input = f"{state['user_input']}\n{current_input}"
|
|
else:
|
|
current_input = state["user_input"]
|
|
full_user_input = current_input
|
|
|
|
spec = await run_pm_agent(user_input=current_input, history=history)
|
|
|
|
updated_history = list(history)
|
|
updated_history.append({"role": "user", "content": current_input})
|
|
|
|
if not spec.is_complete and spec.clarifying_question:
|
|
updated_history.append({"role": "assistant", "content": spec.clarifying_question})
|
|
|
|
return {
|
|
"spec": spec.model_dump(),
|
|
"status": "spec_ready" if spec.is_complete else "spec_incomplete",
|
|
"chat_history": updated_history,
|
|
"user_input": full_user_input,
|
|
"user_feedback": None,
|
|
"loop_count": 0,
|
|
}
|
|
|
|
async def retrieval_node(state: WorkflowState):
|
|
existing_project = await find_existing_project(state["user_input"])
|
|
return {
|
|
"existing_project": existing_project,
|
|
"status": "existing_found" if existing_project else "no_existing_project",
|
|
}
|
|
|
|
async def dev_node(state: WorkflowState):
|
|
qa_logs = state.get("qa_result", {}).get("logs", "") if state.get("qa_result") else None
|
|
|
|
generated_code = await run_dev_agent(state["spec"], qa_feedback=qa_logs)
|
|
return {
|
|
"generated_code": generated_code,
|
|
"status": "code_generated",
|
|
}
|
|
|
|
async def qa_node(state: WorkflowState):
|
|
qa_result = await run_qa_agent(state["generated_code"])
|
|
current_loops = state.get("loop_count", 0)
|
|
|
|
is_success = True
|
|
|
|
clean_qa_result = {"success": is_success, "raw": qa_result}
|
|
|
|
return {
|
|
"qa_result": clean_qa_result,
|
|
"loop_count": current_loops if is_success else current_loops + 1,
|
|
"status": "qa_done",
|
|
}
|
|
|
|
async def human_review_node(state: WorkflowState):
|
|
print("[Human Review] Passage en mode automatique (Mock)...")
|
|
|
|
return {
|
|
"existing_project_approved": True,
|
|
"is_completed": True,
|
|
"status": "approved_by_human"
|
|
}
|
|
|
|
# def inspect_specifications_gaps(spec_dict: dict) -> list[str]:
|
|
# missing_fields = []
|
|
|
|
# # 1. Vérifications globales (clés alignées sur ProjectSpec)
|
|
# if not spec_dict.get("title"): missing_fields.append("title")
|
|
# if not spec_dict.get("description"): missing_fields.append("description")
|
|
# if not spec_dict.get("requirements"): missing_fields.append("requirements")
|
|
|
|
# # 2. Entrées/Sorties
|
|
# io = spec_dict.get("io_config", {})
|
|
# if io.get("has_inputs") is True and not io.get("input_type"):
|
|
# missing_fields.append("io_config.input_type")
|
|
# if io.get("has_outputs") is True and not io.get("output_type"):
|
|
# missing_fields.append("io_config.output_type")
|
|
|
|
# # 3. Authentification
|
|
# auth = spec_dict.get("auth_config", {})
|
|
# if auth.get("requires_auth") is True and not auth.get("auth_method"):
|
|
# missing_fields.append("auth_config.auth_method")
|
|
|
|
# return missing_fields
|
|
|
|
# def generate_clarifying_questions_prompt(missing_fields: list[str]) -> str:
|
|
# # Le mapping utilise désormais les EXACTES mêmes clés
|
|
# mapping_instructions = {
|
|
# "title": "- Préciser un titre pour le projet.",
|
|
# "description": "- Expliquer le but global du script.",
|
|
# "requirements": "- Lister les fonctionnalités attendues étape par étape.",
|
|
# "io_config.input_type": "- Préciser le type et le format des fichiers/données d'entrée (CSV, dossier, etc.).",
|
|
# "io_config.output_type": "- Préciser le type et le format attendus en sortie (Excel, PDF, log, etc.).",
|
|
# "auth_config.auth_method": "- Clarifier la méthode d'accès ou d'authentification exigée pour l'outil tiers."
|
|
# }
|
|
|
|
# bullet_points = "\n".join([mapping_instructions[field] for field in missing_fields if field in mapping_instructions])
|
|
|
|
# return f"""
|
|
# ATTENTION : Les données suivantes sont obligatoires mais absentes.
|
|
# Vous devez formuler une question naturelle et polie pour demander à l'utilisateur de préciser :
|
|
# {bullet_points}
|
|
# """ |