Prompt for root directory and list it
This commit is contained in:
+181
@@ -0,0 +1,181 @@
|
||||
# agent/agent.py
|
||||
from typing import Any, Dict, List
|
||||
import json
|
||||
|
||||
from .llm import DeepSeekClient
|
||||
from .memory import Memory
|
||||
from .tools import make_tools, Tool
|
||||
|
||||
class Agent:
|
||||
def __init__(self, llm: DeepSeekClient, memory: Memory):
|
||||
self.llm = llm
|
||||
self.memory = memory
|
||||
self.tools: Dict[str, Tool] = make_tools(memory)
|
||||
|
||||
def _build_system_prompt(self) -> str:
|
||||
ctx = {"project_root": self.memory.get_project_root()}
|
||||
tools_desc = "\n".join(
|
||||
f"- {t.name}: {t.description}" for t in self.tools.values()
|
||||
)
|
||||
return (
|
||||
"Tu es un agent IA qui aide un développeur senior à gérer son projet local.\n"
|
||||
"Tu peux demander des informations de base (comme le chemin du projet)\n"
|
||||
"et tu peux utiliser des outils pour interagir avec le système de fichiers.\n\n"
|
||||
"Contexte utilisateur (JSON):\n"
|
||||
f"{json.dumps(ctx, ensure_ascii=False)}\n\n"
|
||||
"Règles IMPORTANTES pour les outils:\n"
|
||||
"1. Si tu ne connais pas la valeur d'un argument (par exemple project_root), "
|
||||
"TU NE DOIS PAS mettre null ou une valeur inventée.\n"
|
||||
" À la place, tu dois poser une question à l'utilisateur pour obtenir l'information.\n"
|
||||
"2. Ne propose set_user_profile QUE lorsque l'utilisateur a donné un chemin de projet explicite.\n"
|
||||
"3. Quand tu veux utiliser un outil, réponds STRICTEMENT avec un JSON de la forme :\n"
|
||||
'{ "thought": "...", "action": { "name": "tool_name", "args": { ... } } }\n'
|
||||
" - Pas de texte avant/après.\n"
|
||||
" - Les args doivent être COMPLETS et NON nuls.\n"
|
||||
"4. Quand JE (le système) te fournis un object JSON contenant 'tool_result' et 'intent', "
|
||||
"tu dois ALORS répondre à l'utilisateur en TEXTE NATUREL, et NE PAS renvoyer de JSON d'action.\n\n"
|
||||
"Tools disponibles:\n"
|
||||
f"{tools_desc}\n"
|
||||
)
|
||||
|
||||
def _parse_intent(self, text: str) -> Dict[str, Any] | None:
|
||||
try:
|
||||
data = json.loads(text)
|
||||
except json.JSONDecodeError:
|
||||
return None
|
||||
|
||||
if not isinstance(data, dict):
|
||||
return None
|
||||
|
||||
action = data.get("action")
|
||||
if not isinstance(action, dict):
|
||||
return None
|
||||
|
||||
name = action.get("name")
|
||||
if not isinstance(name, str):
|
||||
return None
|
||||
|
||||
return data
|
||||
|
||||
def _execute_action(self, intent: Dict[str, Any]) -> Dict[str, Any]:
|
||||
action = intent["action"]
|
||||
name: str = action["name"]
|
||||
args: Dict[str, Any] = action.get("args", {}) or {}
|
||||
|
||||
if name == "set_user_profile":
|
||||
project_root = args.get("project_root")
|
||||
if not project_root:
|
||||
return {
|
||||
"error": "missing_project_root",
|
||||
"message": (
|
||||
"Le modèle a demandé set_user_profile sans project_root. "
|
||||
"Tu dois d'abord demander à l'utilisateur de fournir un chemin "
|
||||
"de projet valide (ex: /home/francois/mon_projet)."
|
||||
),
|
||||
}
|
||||
|
||||
tool = self.tools.get(name)
|
||||
if not tool:
|
||||
return {"error": "unknown_tool", "tool": name}
|
||||
|
||||
try:
|
||||
result = tool.func(**args)
|
||||
except TypeError as e:
|
||||
# Mauvais arguments
|
||||
return {"error": "bad_args", "message": str(e)}
|
||||
|
||||
return result
|
||||
|
||||
def step(self, user_input: str) -> str:
|
||||
print("Starting a new step...")
|
||||
"""
|
||||
Un 'tour' d'agent :
|
||||
- construit le prompt system
|
||||
- interroge DeepSeek
|
||||
- si JSON d'intent -> exécute tool, refait un appel, renvoie réponse finale
|
||||
- sinon -> renvoie texte brut
|
||||
"""
|
||||
print("User input:", user_input)
|
||||
root = self.memory.data.get("project_root")
|
||||
print("Current project_root in memory:", root)
|
||||
|
||||
# Unified system prompt that always allows tools
|
||||
tools_desc = "\n".join(
|
||||
f"- {t.name}: {t.description}\n Paramètres: {json.dumps(t.parameters, ensure_ascii=False)}"
|
||||
for t in self.tools.values()
|
||||
)
|
||||
|
||||
if root is None:
|
||||
print("No project_root set - asking user and allowing tool use")
|
||||
system_prompt = (
|
||||
"Tu es un agent IA qui aide un développeur à gérer son projet local.\n\n"
|
||||
"CONTEXTE ACTUEL:\n"
|
||||
f"- project_root: {root} (NON DÉFINI)\n\n"
|
||||
"RÈGLES IMPORTANTES:\n"
|
||||
"1. Le project_root n'est pas encore défini. Tu DOIS d'abord demander à l'utilisateur "
|
||||
"le chemin absolu de son projet (ex: /home/user/mon_projet).\n"
|
||||
"2. Quand l'utilisateur te donne un chemin, tu DOIS immédiatement utiliser l'outil "
|
||||
"'set_project_root' pour le sauvegarder.\n"
|
||||
"3. Pour utiliser un outil, réponds STRICTEMENT avec ce format JSON:\n"
|
||||
' { "thought": "explication", "action": { "name": "nom_outil", "args": { "arg": "valeur" } } }\n'
|
||||
"4. Si tu réponds en texte (pas d'outil), réponds normalement en français.\n"
|
||||
"5. Quand le système te renvoie un tool_result, réponds à l'utilisateur en TEXTE NATUREL.\n\n"
|
||||
"OUTILS DISPONIBLES:\n"
|
||||
f"{tools_desc}\n"
|
||||
)
|
||||
else:
|
||||
print("Project_root is set - normal operation mode")
|
||||
system_prompt = (
|
||||
"Tu es un agent IA qui aide un développeur à gérer son projet local.\n\n"
|
||||
"CONTEXTE ACTUEL:\n"
|
||||
f"- project_root: {root}\n\n"
|
||||
"RÈGLES IMPORTANTES:\n"
|
||||
"1. Le project_root est défini. Tu peux utiliser les outils disponibles.\n"
|
||||
"2. Pour utiliser un outil, réponds STRICTEMENT avec ce format JSON:\n"
|
||||
' { "thought": "explication", "action": { "name": "nom_outil", "args": { "param": "valeur" } } }\n'
|
||||
" EXEMPLE pour lister le dossier 'src':\n"
|
||||
' { "thought": "L\'utilisateur veut voir le contenu de src", "action": { "name": "list_directory", "args": { "path": "src" } } }\n'
|
||||
" EXEMPLE pour lister la racine du projet:\n"
|
||||
' { "thought": "L\'utilisateur veut voir la racine", "action": { "name": "list_directory", "args": { "path": "." } } }\n'
|
||||
"3. Si tu réponds en texte (pas d'outil), réponds normalement en français.\n"
|
||||
"4. Quand le système te renvoie un tool_result, réponds à l'utilisateur en TEXTE NATUREL.\n"
|
||||
"5. IMPORTANT: Extrais le chemin demandé par l'utilisateur et passe-le comme argument 'path'.\n\n"
|
||||
"OUTILS DISPONIBLES:\n"
|
||||
f"{tools_desc}\n"
|
||||
)
|
||||
|
||||
messages: List[Dict[str, Any]] = [
|
||||
{"role": "system", "content": system_prompt},
|
||||
{"role": "user", "content": user_input},
|
||||
]
|
||||
|
||||
raw_first = self.llm.complete(messages)
|
||||
intent = self._parse_intent(raw_first)
|
||||
print("raw_first:", raw_first)
|
||||
print("Intent:", intent)
|
||||
if not intent:
|
||||
# Réponse texte simple
|
||||
#self.memory.append_history("user", user_input)
|
||||
#self.memory.append_history("assistant", raw_first)
|
||||
return raw_first
|
||||
|
||||
# Exécuter l'action
|
||||
tool_result = self._execute_action(intent)
|
||||
|
||||
followup_messages: List[Dict[str, Any]] = [
|
||||
{"role": "system", "content": system_prompt},
|
||||
{"role": "user", "content": user_input},
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": json.dumps(
|
||||
{"tool_result": tool_result, "intent": intent},
|
||||
ensure_ascii=False,
|
||||
),
|
||||
},
|
||||
]
|
||||
|
||||
raw_second = self.llm.complete(followup_messages)
|
||||
|
||||
#self.memory.append_history("user", user_input)
|
||||
#self.memory.append_history("assistant", raw_second)
|
||||
return raw_second
|
||||
@@ -0,0 +1,109 @@
|
||||
# agent/commands.py
|
||||
from dataclasses import dataclass
|
||||
from typing import Callable, Dict, List
|
||||
import os
|
||||
|
||||
from .memory import Memory
|
||||
|
||||
@dataclass
|
||||
class Command:
|
||||
name: str
|
||||
description: str
|
||||
needs_project_root: bool
|
||||
handler: Callable[[List[str], Memory], str]
|
||||
|
||||
|
||||
class CommandRegistry:
|
||||
def __init__(self, memory: Memory):
|
||||
self.memory = memory
|
||||
self.commands: Dict[str, Command] = {}
|
||||
self._register_defaults()
|
||||
|
||||
def _register_defaults(self):
|
||||
# /setroot <path>
|
||||
def cmd_setroot(args: List[str], mem: Memory) -> str:
|
||||
if not args:
|
||||
return "Usage: `/setroot <chemin_absolu_vers_projet>`"
|
||||
|
||||
path = args[0]
|
||||
if not os.path.isdir(path):
|
||||
return f"Le chemin `{path}` n'existe pas ou n'est pas un dossier."
|
||||
|
||||
mem.set_project_root(path)
|
||||
return f"✅ Chemin du projet défini sur `{path}`."
|
||||
|
||||
self.register(
|
||||
Command(
|
||||
name="setroot",
|
||||
description="Définit le chemin racine du projet.",
|
||||
needs_project_root=False,
|
||||
handler=cmd_setroot,
|
||||
)
|
||||
)
|
||||
|
||||
# /scan [rel_path]
|
||||
def cmd_scan(args: List[str], mem: Memory) -> str:
|
||||
root = mem.get_project_root()
|
||||
rel = args[0] if args else "."
|
||||
full = os.path.abspath(os.path.join(root, rel))
|
||||
|
||||
# sécurité basique
|
||||
if not full.startswith(os.path.abspath(root)):
|
||||
return "❌ Tu ne peux pas sortir du project_root."
|
||||
|
||||
if not os.path.isdir(full):
|
||||
return f"❌ `{rel}` n'est pas un dossier dans le projet."
|
||||
|
||||
entries = sorted(os.listdir(full))
|
||||
if not entries:
|
||||
return f"📁 `{rel}` est vide."
|
||||
|
||||
lines = [f"📁 Scan de `{rel}` (dans `{root}`):"]
|
||||
for e in entries:
|
||||
p = os.path.join(full, e)
|
||||
if os.path.isdir(p):
|
||||
lines.append(f" 📂 {e}/")
|
||||
else:
|
||||
lines.append(f" 📄 {e}")
|
||||
return "\n".join(lines)
|
||||
|
||||
self.register(
|
||||
Command(
|
||||
name="scan",
|
||||
description="Liste les fichiers/dossiers du projet.",
|
||||
needs_project_root=True,
|
||||
handler=cmd_scan,
|
||||
)
|
||||
)
|
||||
|
||||
def register(self, cmd: Command):
|
||||
self.commands[cmd.name] = cmd
|
||||
|
||||
def handle(self, raw_input: str) -> str:
|
||||
"""
|
||||
Parse et exécute une commande du type `/scan src` ou `/setroot /chemin`.
|
||||
"""
|
||||
text = raw_input.strip()
|
||||
if not text.startswith("/"):
|
||||
return "Internal error: not a command."
|
||||
|
||||
parts = text[1:].split()
|
||||
if not parts:
|
||||
return "Commande vide."
|
||||
|
||||
name = parts[0]
|
||||
args = parts[1:]
|
||||
|
||||
cmd = self.commands.get(name)
|
||||
if not cmd:
|
||||
return f"Commande inconnue: `/{name}`.\nCommandes disponibles: {', '.join('/'+n for n in self.commands.keys())}"
|
||||
|
||||
# dépendance project_root
|
||||
if cmd.needs_project_root and not self.memory.get_project_root():
|
||||
return (
|
||||
"❗ Aucun `project_root` défini pour l'instant.\n"
|
||||
"Commence par le définir avec:\n"
|
||||
"`/setroot /chemin/vers/ton/projet`"
|
||||
)
|
||||
|
||||
return cmd.handler(args, self.memory)
|
||||
@@ -0,0 +1,15 @@
|
||||
from dataclasses import dataclass
|
||||
import os
|
||||
from dotenv import load_dotenv
|
||||
|
||||
load_dotenv()
|
||||
|
||||
@dataclass
|
||||
class Settings:
|
||||
deepseek_api_key: str = os.getenv("DEEPSEEK_API_KEY", "")
|
||||
deepseek_base_url: str = os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com")
|
||||
model: str = os.getenv("DEEPSEEK_MODEL", "deepseek-chat")
|
||||
temperature: float = float(os.getenv("TEMPERATURE", "0.2"))
|
||||
memory_file: str = os.getenv("MEMORY_FILE", "memory.json")
|
||||
|
||||
settings = Settings()
|
||||
@@ -0,0 +1,41 @@
|
||||
# agent/llm.py
|
||||
from typing import List, Dict, Any
|
||||
import requests
|
||||
|
||||
from .config import settings
|
||||
|
||||
class DeepSeekClient:
|
||||
def __init__(
|
||||
self,
|
||||
api_key: str | None = None,
|
||||
base_url: str | None = None,
|
||||
model: str | None = None,
|
||||
):
|
||||
self.api_key = api_key or settings.deepseek_api_key
|
||||
self.base_url = base_url or settings.deepseek_base_url
|
||||
self.model = model or settings.model
|
||||
|
||||
def complete(self, messages: List[Dict[str, Any]]) -> str:
|
||||
"""
|
||||
messages: liste de dicts {role: 'system'|'user'|'assistant', content: str}
|
||||
Retourne content (str) du premier choix.
|
||||
"""
|
||||
if not self.api_key:
|
||||
return "Erreur côté agent : DEEPSEEK_API_KEY manquant dans l'environnement backend."
|
||||
|
||||
url = f"{self.base_url}/v1/chat/completions"
|
||||
headers = {
|
||||
"Authorization": f"Bearer {self.api_key}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
payload = {
|
||||
"model": self.model,
|
||||
"messages": messages,
|
||||
"temperature": settings.temperature,
|
||||
}
|
||||
|
||||
resp = requests.post(url, headers=headers, json=payload, timeout=60)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
print("DeepSeek response:", data)
|
||||
return data["choices"][0]["message"]["content"]
|
||||
@@ -0,0 +1,38 @@
|
||||
# agent/memory.py
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict
|
||||
import json
|
||||
|
||||
from .config import settings
|
||||
|
||||
|
||||
class Memory:
|
||||
def __init__(self, path: str = "memory.json"):
|
||||
print("init memory")
|
||||
self.file = Path(path)
|
||||
self.data: Dict[str, Any] = {}
|
||||
self.load()
|
||||
|
||||
def load(self) -> None:
|
||||
if self.file.exists():
|
||||
self.data = json.loads(self.file.read_text(encoding="utf-8"))
|
||||
else:
|
||||
self.data = {
|
||||
"project_root": None,
|
||||
"history": [],
|
||||
}
|
||||
|
||||
def save(self) -> None:
|
||||
self.file.write_text(
|
||||
json.dumps(self.data, indent=2, ensure_ascii=False),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
def get_project_root(self) -> str | None:
|
||||
"""Ce qu'on injecte dans le prompt pour le LLM."""
|
||||
return self.data.get("project_root")
|
||||
|
||||
def set_project_root(self, path: str) -> None:
|
||||
print('Setting project root in memory to:', path)
|
||||
self.data["project_root"] = path
|
||||
self.save()
|
||||
@@ -0,0 +1,76 @@
|
||||
# agent/tools.py
|
||||
from dataclasses import dataclass
|
||||
from typing import Callable, Any, Dict
|
||||
import os
|
||||
|
||||
from .memory import Memory
|
||||
|
||||
@dataclass
|
||||
class Tool:
|
||||
name: str
|
||||
description: str
|
||||
func: Callable[..., Dict[str, Any]]
|
||||
parameters: Dict[str, Any] # JSON Schema des paramètres
|
||||
|
||||
def make_tools(memory: Memory) -> dict[str, Tool]:
|
||||
def set_project_root(project_root: str) -> Dict[str, Any]:
|
||||
if not os.path.isdir(project_root):
|
||||
return {"error": "invalid_path", "message": f"Le chemin {project_root} n'est pas un dossier valide."}
|
||||
|
||||
print(f"Setting project root to: {project_root}")
|
||||
print("Memory before:", memory.data)
|
||||
memory.set_project_root(project_root)
|
||||
print("Memory after:", memory.data)
|
||||
return {"status": "ok", "project_root": project_root}
|
||||
|
||||
def list_directory(path: str) -> Dict[str, Any]:
|
||||
print("Proper tool used")
|
||||
if not memory.data.get("project_root"):
|
||||
return {"error": "no_project_root", "message": "Project root not set."}
|
||||
|
||||
root = memory.data.get("project_root")
|
||||
full_path = os.path.abspath(os.path.join(root, path))
|
||||
root_abs = os.path.abspath(root)
|
||||
if not full_path.startswith(root_abs):
|
||||
return {"error": "forbidden", "message": "Path outside project_root."}
|
||||
|
||||
try:
|
||||
entries = os.listdir(full_path)
|
||||
return {"path": path, "entries": entries}
|
||||
except Exception as e:
|
||||
return {"error": "os_error", "message": str(e)}
|
||||
|
||||
tools = [
|
||||
Tool(
|
||||
name="set_project_root",
|
||||
description="Enregistre le path du dossier racine du projet.",
|
||||
func=set_project_root,
|
||||
parameters={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"project_root": {
|
||||
"type": "string",
|
||||
"description": "Chemin absolu du dossier racine du projet (ex: /home/user/mon_projet)"
|
||||
}
|
||||
},
|
||||
"required": ["project_root"]
|
||||
}
|
||||
),
|
||||
Tool(
|
||||
name="list_directory",
|
||||
description="Liste le contenu d'un dossier relatif au project_root.",
|
||||
func=list_directory,
|
||||
parameters={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"path": {
|
||||
"type": "string",
|
||||
"description": "Chemin relatif du dossier à lister (ex: 'src' ou '.' pour la racine)"
|
||||
}
|
||||
},
|
||||
"required": ["path"]
|
||||
}
|
||||
),
|
||||
]
|
||||
|
||||
return {t.name: t for t in tools}
|
||||
Reference in New Issue
Block a user