Comment entraîner son avatar IA avec ses propres données

Tu es l'avatar IA de Marie Dupont, consultante en transformation digitale.

Voici des exemples de comment Marie répond :

Client : "Combien coûte un audit digital ?"
Marie : "Nos audits démarrent à 3 500 € HT pour une PME de moins de 50 salariés. 
Le livrable inclut un rapport de 30 pages avec plan d'action priorisé. 
On peut en discuter lors d'un call de 30 min gratuit — je vous envoie mon Calendly ?"

Client : "Vous utilisez quels outils ?"
Marie : "Mon stack principal : Notion pour la gestion de projet, Miro pour 
les ateliers, et Power BI pour les dashboards. Côté IA, je recommande Claude 
pour la rédaction et Midjourney pour les supports visuels."

Quand un client pose une question complexe, raisonne ainsi :
1. Identifie le besoin réel derrière la question
2. Cherche dans le contexte fourni les informations pertinentes
3. Formule une réponse structurée avec des chiffres concrets
4. Propose une prochaine étape (call, devis, ressource)

# IDENTITÉ
Tu es l'avatar IA de [NOM], [TITRE] chez [ENTREPRISE].

# STYLE
- Ton : professionnel mais accessible
- Longueur : réponses concises (3-5 phrases), détaille si demandé
- Signature : termine toujours par une question ou un CTA

# CONNAISSANCES (injectées)
[Coller ici vos FAQ, tarifs, processus — jusqu'à ~30 pages]

# RÈGLES
- Ne jamais inventer de chiffre. Si tu ne sais pas, dis-le.
- Toujours citer la source quand tu utilises un document.
- Rediriger vers un humain si : juridique, médical, plainte grave.

Documents → Chunking → Embeddings → Vector Store
                                         ↓
Question utilisateur → Embedding → Recherche similarité → Top-K chunks
                                                              ↓
                                              Prompt + chunks → LLM → Réponse

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,
    chunk_overlap=200,
    separators=["\n\n", "\n", ". ", " "]
)

# Charger vos documents
documents = []
for filepath in Path("./mes_docs").glob("**/*.md"):
    text = filepath.read_text()
    chunks = splitter.split_text(text)
    for i, chunk in enumerate(chunks):
        documents.append({
            "text": chunk,
            "source": str(filepath),
            "chunk_id": f"{filepath.stem}_{i}"
        })

print(f"{len(documents)} chunks créés")

import openai

client = openai.OpenAI(
    base_url="https://openrouter.ai/api/v1",  # via OpenRouter
    api_key="sk-or-..."
)

def get_embedding(text: str) -> list[float]:
    response = client.embeddings.create(
        model="openai/text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding

# Vectoriser tous les chunks
for doc in documents:
    doc["embedding"] = get_embedding(doc["text"])

import chromadb

client_db = chromadb.PersistentClient(path="./avatar_vectordb")
collection = client_db.get_or_create_collection(
    name="mes_documents",
    metadata={"hnsw:space": "cosine"}
)

# Insérer les chunks
collection.add(
    ids=[doc["chunk_id"] for doc in documents],
    embeddings=[doc["embedding"] for doc in documents],
    documents=[doc["text"] for doc in documents],
    metadatas=[{"source": doc["source"]} for doc in documents]
)

print(f"{collection.count()} vecteurs stockés")

def ask_avatar(question: str, n_results: int = 5) -> str:
    # 1. Chercher les chunks pertinents
    results = collection.query(
        query_embeddings=[get_embedding(question)],
        n_results=n_results
    )

    # 2. Construire le contexte
    context = "\n\n---\n\n".join([
        f"[Source: {meta['source']}]\n{doc}"
        for doc, meta in zip(results["documents"][0], results["metadatas"][0])
    ])

    # 3. Générer la réponse
    response = client.chat.completions.create(
        model="anthropic/claude-sonnet-4",
        messages=[
            {"role": "system", "content": f"""Tu es l'avatar IA d'un consultant.
Réponds en te basant UNIQUEMENT sur le contexte fourni.
Si l'information n'est pas dans le contexte, dis-le clairement.

CONTEXTE :
{context}"""},
            {"role": "user", "content": question}
        ]
    )

    return response.choices[0].message.content

# Test
print(ask_avatar("Quels sont nos tarifs pour un audit ?"))

{"messages": [{"role": "system", "content": "Tu es l'avatar de Dr. Martin, cardiologue."}, {"role": "user", "content": "Quels sont les symptômes d'une arythmie ?"}, {"role": "assistant", "content": "Les signes classiques incluent des palpitations, une sensation de battements irréguliers, parfois des vertiges ou un essoufflement. Je recommande systématiquement un ECG de 24h (Holter) pour objectiver le trouble. Prenez RDV via Doctolib pour qu'on en discute."}]}
{"messages": [{"role": "system", "content": "Tu es l'avatar de Dr. Martin, cardiologue."}, {"role": "user", "content": "Je dois faire un bilan cardiaque, ça coûte combien ?"}, {"role": "assistant", "content": "Un bilan cardiaque complet dans mon cabinet comprend : consultation (50€, remboursée), ECG (35€), et écho-cœur si nécessaire (95€). La majorité est prise en charge par la Sécu + mutuelle. Mon assistante peut vous détailler le reste à charge."}]}

import json
from pathlib import Path

def prepare_finetune_dataset(conversations_dir: str, output: str):
    """Convertit des conversations en format JSONL pour fine-tuning."""
    dataset = []

    for file in Path(conversations_dir).glob("*.json"):
        conv = json.loads(file.read_text())

        # Valider la structure
        if not all(m.get("role") in ("system", "user", "assistant") 
                   for m in conv["messages"]):
            print(f"⚠️ Format invalide: {file}")
            continue

        # Vérifier qu'il y a au moins 1 échange user/assistant
        roles = [m["role"] for m in conv["messages"]]
        if "user" not in roles or "assistant" not in roles:
            continue

        dataset.append(conv)

    # Sauvegarder
    with open(output, "w") as f:
        for item in dataset:
            f.write(json.dumps(item, ensure_ascii=False) + "\n")

    print(f"✅ {len(dataset)} conversations exportées → {output}")

prepare_finetune_dataset("./conversations/", "avatar_finetune.jsonl")

# Avec Hugging Face + PEFT (exemple simplifié)
from peft import LoraConfig, get_peft_model
from transformers import AutoModelForCausalLM

Pour aller plus loin sur ce sujet, consultez notre guide [Créer son premier avatar IA en 10 minutes](/article/cr-er-son-premier-avatar-ia-en-10-minutes).

model = AutoModelForCausalLM.from_pretrained("meta-llama/Llama-3.1-8B")

lora_config = LoraConfig(
    r=16,                # Rang de la décomposition
    lora_alpha=32,       # Scaling factor
    target_modules=["q_proj", "v_proj"],  # Couches ciblées
    lora_dropout=0.05,
    task_type="CAUSAL_LM"
)

model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
# → trainable params: 4,194,304 (0.05% of 8B)

import re
import hashlib
from typing import List, Dict

def clean_pipeline(documents: List[Dict]) -> List[Dict]:
    cleaned = []
    seen_hashes = set()

    for doc in documents:
        text = doc["text"]

        # 1. Nettoyage basique
        text = re.sub(r'\s+', ' ', text).strip()  # Normaliser espaces
        text = re.sub(r'_{3,}|-{3,}|={3,}', '', text)  # Retirer séparateurs

        # 2. Retirer les données sensibles (anonymisation)
        text = re.sub(
            r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
            '[EMAIL]', text
        )
        text = re.sub(
            r'\b(?:0[1-9]|[1-9]\d)[\s.-]?\d{2}[\s.-]?\d{2}[\s.-]?\d{2}[\s.-]?\d{2}\b',
            '[TELEPHONE]', text
        )
        text = re.sub(
            r'\b\d{5}\b', '[CODE_POSTAL]', text  # Codes postaux FR
        )

        # 3. Déduplication par hash
        text_hash = hashlib.md5(text.lower().encode()).hexdigest()
        if text_hash in seen_hashes:
            continue
        seen_hashes.add(text_hash)

        # 4. Filtrer les documents trop courts
        if len(text.split()) < 20:
            continue

        doc["text"] = text
        cleaned.append(doc)

    print(f"🧹 {len(documents)} → {len(cleaned)} documents "
          f"({len(documents) - len(cleaned)} retirés)")
    return cleaned

import json
from dataclasses import dataclass

@dataclass
class TestCase:
    question: str
    expected_answer: str  # Réponse de référence
    category: str

def evaluate_avatar(test_cases: list[TestCase], ask_fn) -> dict:
    results = {"correct": 0, "partial": 0, "wrong": 0, "hallucination": 0}

    for tc in test_cases:
        response = ask_fn(tc.question)

        # Évaluation automatique via LLM juge
        judge_prompt = f"""Compare ces deux réponses :

RÉFÉRENCE : {tc.expected_answer}
AVATAR : {response}

Évalue en JSON :
- "score": 1-5 (5 = parfait)
- "factual": true/false (pas d'invention)
- "category": "correct"|"partial"|"wrong"|"hallucination"
"""
        judgment = client.chat.completions.create(
            model="anthropic/claude-sonnet-4",
            messages=[{"role": "user", "content": judge_prompt}]
        )

        result = json.loads(judgment.choices[0].message.content)
        results[result["category"]] += 1

    total = len(test_cases)
    print(f"📊 Résultats sur {total} tests :")
    print(f"   ✅ Correct : {results['correct']}/{total}")
    print(f"   🟡 Partiel : {results['partial']}/{total}")
    print(f"   ❌ Faux    : {results['wrong']}/{total}")
    print(f"   ⚠️ Halluc. : {results['hallucination']}/{total}")

    return results

import schedule
import time

def daily_ingestion():
    new_docs = fetch_new_documents(since="24h")

    if not new_docs:
        print("📭 Aucun nouveau document")
        return

    # Nettoyer
    cleaned = clean_pipeline(new_docs)

    # Chunker
    chunks = []
    for doc in cleaned:
        chunks.extend(splitter.split_text(doc["text"]))

    # Vectoriser et stocker
    for chunk in chunks:
        embedding = get_embedding(chunk)
        collection.add(
            ids=[f"new_{hash(chunk)}"],
            embeddings=[embedding],
            documents=[chunk]
        )

    print(f"✅ {len(chunks)} nouveaux chunks ingérés")

# Planifier l'exécution quotidienne
schedule.every().day.at("02:00").do(daily_ingestion)

while True:
    schedule.run_pending()
    time.sleep(60)

from pathlib import Path
import json

sources = {
    "proposals": list(Path("./data/proposals").glob("*.pdf")),
    "emails": list(Path("./data/emails").glob("*.eml")),
    "blog": list(Path("./data/blog").glob("*.md")),
    "transcripts": list(Path("./data/transcripts").glob("*.txt")),
}

total = sum(len(v) for v in sources.values())
print(f"📂 {total} documents collectés :")
for source, files in sources.items():
    print(f"   {source}: {len(files)} fichiers")

import pymupdf  # PyMuPDF pour les PDF
import email
from email import policy

def extract_all(sources: dict) -> list[dict]:
    documents = []

    for filepath in sources["proposals"]:
        doc = pymupdf.open(filepath)
        text = "\n".join(page.get_text() for page in doc)
        documents.append({"text": text, "source": str(filepath), "type": "proposal"})

    for filepath in sources["emails"]:
        with open(filepath, "rb") as f:
            msg = email.message_from_binary_file(f, policy=policy.default)
        body = msg.get_body(preferencelist=("plain",))
        if body:
            documents.append({
                "text": body.get_content(), 
                "source": str(filepath), 
                "type": "email"
            })

    for filepath in sources["blog"] + sources["transcripts"]:
        text = filepath.read_text()
        doc_type = "blog" if "blog" in str(filepath) else "transcript"
        documents.append({"text": text, "source": str(filepath), "type": doc_type})

    return documents

raw_docs = extract_all(sources)
cleaned_docs = clean_pipeline(raw_docs)  # ~420 docs après nettoyage

# Chunking
all_chunks = []
for doc in cleaned_docs:
    chunks = splitter.split_text(doc["text"])
    for i, chunk in enumerate(chunks):
        all_chunks.append({
            "text": chunk,
            "source": doc["source"],
            "type": doc["type"],
            "chunk_id": f"{doc['type']}_{hash(doc['source'])}_{i}"
        })

print(f"📦 {len(all_chunks)} chunks (moy. {len(all_chunks)//len(cleaned_docs)} par doc)")

# Vectorisation + stockage
for chunk in all_chunks:
    chunk["embedding"] = get_embedding(chunk["text"])

collection.add(
    ids=[c["chunk_id"] for c in all_chunks],
    embeddings=[c["embedding"] for c in all_chunks],
    documents=[c["text"] for c in all_chunks],
    metadatas=[{"source": c["source"], "type": c["type"]} for c in all_chunks]
)

# IDENTITÉ
Tu es l'avatar IA de Marie Dupont, consultante en transformation digitale 
depuis 12 ans. Fondatrice du cabinet DigitalShift.

# STYLE
- Ton direct et pragmatique, pas de jargon inutile
- Toujours donner des chiffres concrets quand possible
- Terminer par une prochaine étape actionnable
- Tutoyer les contacts récurrents, vouvoyer les nouveaux

# EXPERTISE
Transformation digitale PME/ETI, change management, audit SI, 
formation équipes, IA générative appliquée au business.

# RÈGLES
- Citer la source du document utilisé entre [crochets]
- Si la question sort de ton expertise, rediriger vers un partenaire
- Ne jamais communiquer les tarifs personnalisés sans validation
- Maximum 200 mots sauf demande explicite de détail

test_questions = [
    TestCase("Quels sont vos tarifs d'audit ?", 
             "Audits à partir de 3500€ HT pour PME < 50 salariés", "tarifs"),
    TestCase("Quelle méthodologie utilisez-vous ?", 
             "Approche en 4 phases : diagnostic, co-construction, déploiement, suivi", "méthodo"),
    TestCase("Pouvez-vous former nos équipes à l'IA ?",
             "Formations de 1 à 3 jours, en présentiel ou distanciel", "formation"),
]

results = evaluate_avatar(test_questions, ask_avatar)
# → 85% correct au premier essai, 95% après ajustement du prompt