import os
import json
import logging
import gradio as gr
from pydantic import BaseModel, Field
from openai import OpenAI
from langchain.vectorstores import FAISS
from langchain.embeddings import HuggingFaceEmbeddings
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from langchain_core.tools import Tool
from langgraph.graph import MessagesState, StateGraph, START
# ============================================================
# 1️⃣ LOGGING
# ============================================================
logging.basicConfig(level=logging.INFO, format='[%(asctime)s] [%(levelname)s] %(message)s')
log = logging.getLogger(__name__)
# ============================================================
# 2️⃣ CLIENT GPT-OSS
# ============================================================
client = OpenAI(
base_url="https://router.huggingface.co/v1",
api_key=os.getenv("HF_TOKEN")
)
# ============================================================
# 3️⃣ PROMPT SYSTÈME
# ============================================================
PROMPT_SYSTEM = """
Tu es un assistant expert en recertification d’applications.
Tu disposes des outils suivants :
- smart_get_info : Donne l'équipe responsable et la classification d'une application.
- smart_audit_access : Liste les derniers accès à une application.
- smart_recertification_status : Informe si une application doit être recertifiée.
- smart_rag_recertification_context : Recherche dans la base RAG les définitions, étapes et acteurs pour lancer la recertification d'une application.
Règles :
- Si la question est générale (ex. « c’est quoi une recertification ? »), réponds directement sans appeler d’outil.
- Si une information précise est demandée sur une application, appelle un outil si besoin.
- Ne répète jamais un outil si tu as déjà reçu l’Observation.
- Réponds toujours de façon claire, concise et utile.
"""
# ============================================================
# 4️⃣ TOOLS avec args_schema (BaseModel)
# ============================================================
# ---- 1. Tool smart_get_info ----
class SmartGetInfoArgs(BaseModel):
name: str = Field(..., description="Nom exact de l'application.")
description: str = Field("", description="Courte description de l'application.")
keywords: str = Field("", description="Mots-clés associés à l'application.")
def smart_get_info(name: str, description: str, keywords: str) -> str:
log.info(f"[TOOL] smart_get_info(name={name}, description={description}, keywords={keywords})")
return (
f"L'application '{name}' ({keywords}) est décrite comme '{description}'. "
"Elle est gérée par l'équipe Développement et classée 'Confidentielle'."
)
# ---- 2. Tool smart_audit_access ----
class SmartAuditAccessArgs(BaseModel):
app: str = Field(..., description="Nom de l'application pour laquelle afficher les derniers accès.")
def smart_audit_access(app: str) -> str:
log.info(f"[TOOL] smart_audit_access: {app}")
return f"Derniers accès à {app} : UserX, UserY."
# ---- 3. Tool smart_recertification_status ----
class SmartRecertificationStatusArgs(BaseModel):
app: str = Field(..., description="Nom de l'application à vérifier.")
def smart_recertification_status(app: str) -> str:
log.info(f"[TOOL] smart_recertification_status: {app}")
return f"{app} doit être recertifiée avant la fin du mois."
# ---- 4. Tool RAG ----
embedding_model = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2")
db = FAISS.load_local("recertification_index", embedding_model)
retriever = db.as_retriever(search_kwargs={"k": 3})
class SmartRagContextArgs(BaseModel):
query: str = Field(..., description="Texte décrivant ce que l’on cherche dans le contexte de recertification.")
def search_recertification_context(query: str) -> str:
docs = retriever.invoke(query)
return "\n\n".join([doc.page_content for doc in docs])
# ---- Enregistrement des tools ----
tools = [
Tool(name="smart_get_info", func=smart_get_info, args_schema=SmartGetInfoArgs,
description="Retourne les informations détaillées sur une application à partir de plusieurs critères."),
Tool(name="smart_audit_access", func=smart_audit_access, args_schema=SmartAuditAccessArgs,
description="Retourne les derniers accès à une application."),
Tool(name="smart_recertification_status", func=smart_recertification_status, args_schema=SmartRecertificationStatusArgs,
description="Indique si une application doit être recertifiée."),
Tool(name="smart_rag_recertification_context", func=search_recertification_context,
args_schema=SmartRagContextArgs,
description="Recherche dans la base RAG les définitions, étapes et acteurs liés à la recertification.")
]
# ============================================================
# 5️⃣ TOOL EXECUTOR LOCAL
# ============================================================
class ToolExecutor:
def __init__(self, tools):
self.tool_map = {t.name: t for t in tools}
def invoke(self, call):
name = call.get("name")
args = call.get("arguments", {})
if name not in self.tool_map:
raise ValueError(f"Outil non trouvé : {name}")
tool = self.tool_map[name]
if isinstance(args, str):
args = json.loads(args)
return tool.run(args)
tool_executor = ToolExecutor(tools)
# ============================================================
# 6️⃣ GPT-OSS WRAPPER
# ============================================================
class GPTOSSWrapper:
def __init__(self, model="openai/gpt-oss-120b"):
self.model = model
def invoke(self, messages):
# Convert LangGraph messages → base_parameters.input
base_input = []
for m in messages:
role = "user" if isinstance(m, HumanMessage) else \
"system" if isinstance(m, SystemMessage) else "assistant"
base_input.append({"role": role, "content": m.content})
# Extraire le JSON Schema depuis args_schema
tool_defs = []
for t in tools:
schema = t.args_schema.schema() if hasattr(t.args_schema, "schema") else {}
tool_defs.append({
"type": "function",
"name": t.name,
"description": t.description,
"parameters": schema
})
resp = client.responses.create(
model=self.model,
base_parameters={
"input": base_input,
"tools": tool_defs,
"tool_choice": "auto",
"temperature": 0.1
}
)
try:
data = json.loads(resp.output_text)
except Exception as e:
return AIMessage(content=f"[Erreur parsing modèle] {e}")
if data.get("analysis"):
return AIMessage(content=f"[ACTION] {data['analysis']['action']}",
additional_kwargs={"analysis": data["analysis"]})
if data.get("final"):
return AIMessage(content=data["final"]["answer"])
return AIMessage(content="Aucune réponse détectée.")
llm_oss = GPTOSSWrapper()
# ============================================================
# 7️⃣ NODES LANGGRAPH
# ============================================================
def assistant(state: MessagesState):
log.info("[GRAPH] Assistant node triggered.")
response = llm_oss.invoke(state["messages"])
return {"messages": [response]}
def tools_node(state: MessagesState):
last = state["messages"][-1]
analysis = last.additional_kwargs.get("analysis", {})
if not analysis:
return {"messages": [AIMessage(content="Aucun outil appelé.")]}
name = analysis["action"]
args = analysis.get("action_input", {})
log.info(f"[TOOL EXEC] {name}({args})")
try:
output = tool_executor.invoke({"name": name, "arguments": args})
msg = AIMessage(content=f"[Observation] {output}")
except Exception as e:
msg = AIMessage(content=f"[Erreur] {e}")
return {"messages": [msg]}
def stop_condition(state: MessagesState) -> str:
last = state["messages"][-1]
if isinstance(last, AIMessage) and "analysis" in last.additional_kwargs:
return "continue"
return "end"
# ============================================================
# 8️⃣ GRAPH
# ============================================================
builder = StateGraph(MessagesState)
builder.add_node("assistant", assistant)
builder.add_node("tools", tools_node)
builder.add_node("end", lambda s: s)
builder.add_edge(START, "assistant")
builder.add_conditional_edges("assistant", stop_condition, path_map={"continue": "tools", "end": "end"})
builder.add_conditional_edges("tools", lambda s: "assistant", path_map={"assistant": "assistant"})
builder.set_finish_point("end")
graph = builder.compile()
# ============================================================
# 9️⃣ GRADIO
# ============================================================
chat_history = []
def ask_agent(message: str) -> str:
global chat_history
if not chat_history:
chat_history.append(SystemMessage(content=PROMPT_SYSTEM))
chat_history.append(HumanMessage(content=message))
result = graph.invoke({"messages": chat_history})
msg = result["messages"][-1]
chat_history.append(msg)
if isinstance(msg, AIMessage):
return msg.content
return "[Erreur] Pas de réponse générée."
demo = gr.Interface(
fn=ask_agent,
inputs=gr.Textbox(label="Posez votre question"),
outputs=gr.Textbox(label="Réponse de l'agent"),
title="Assistant Recertification – GPT-OSS-120b (args_schema BaseModel)",
theme="default"
)
if __name__ == "__main__":
demo.launch(server_name="127.0.0.1", share=False)
