# --- requirements_config table ---
# Supposons que tu as une table SQL ou un fichier YAML avec la structure suivante :
# Columns: tool_name | action_type | rag_index_name | rag_query_text | rag_subtool_key
# Exemple de contenu :
# compare_rgpd | rag_multi | rgpd_index | Quelle est la politique de conservation ? | conservation
# compare_rgpd | rag_multi | rgpd_index | Quelles données personnelles sont traitées ? | data_types
# --- get_subqueries (dans RequirementManager) ---
class RequirementManager:
def __init__(self, db_connection):
self.conn = db_connection
def get_subqueries(self, tool_name: str):
query = """
SELECT rag_index_name, rag_query_text, rag_subtool_key
FROM requirements_config
WHERE tool_name = ? AND action_type = 'rag_multi'
"""
df = pd.read_sql(query, self.conn, params=(tool_name,))
return df.to_dict(orient="records")
# --- search_index (faiss par index_name) ---
def search_index(index_name: str, query: str) -> str:
db = FAISS.load_local(index_name, HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2"))
retriever = db.as_retriever(search_kwargs={"k": 3})
docs = retriever.invoke(query)
return "\n\n".join([doc.page_content for doc in docs])
# --- ToolExecutor updated ---
class ToolExecutor:
def __init__(self, tools, requirement_manager):
self.tool_map = {tool.name: tool for tool in tools}
self.requirement_manager = requirement_manager
def invoke(self, tool_call):
tool_name = tool_call.get("name")
arguments = tool_call.get("arguments", {})
# Gestion dynamique rag_multi
if self.requirement_manager:
subqueries = self.requirement_manager.get_subqueries(tool_name)
if subqueries:
results = []
for subtool in subqueries:
query = subtool["rag_query_text"]
index = subtool["rag_index_name"]
key = subtool["rag_subtool_key"]
output = search_index(index, query)
results.append(f"### {key}\n{output.strip()}")
return "\n\n".join(results)
# Fallback: normal tool call
if tool_name not in self.tool_map:
raise ValueError(f"Outil non trouvé : {tool_name}")
tool = self.tool_map[tool_name]
if isinstance(arguments, str):
try:
arguments = json.loads(arguments)
except Exception as e:
raise ValueError(f"Arguments JSON invalides : {e}")
return tool.run(arguments)
_____
