Créer un avatar IA expert dans votre métier

import os
import json
from pathlib import Path

def collect_documents(source_dir: str, extensions: list[str]) -> list[dict]:
    documents = []

    for ext in extensions:
        for filepath in Path(source_dir).rglob(f"*{ext}"):
            try:
                content = filepath.read_text(encoding="utf-8")
                documents.append({
                    "source": str(filepath),
                    "content": content,
                    "type": ext,
                    "size": len(content)
                })
            except (UnicodeDecodeError, PermissionError):
                print(f"Unable to read: {filepath}")

    return documents

# Collecting professional documents
docs = collect_documents(
    source_dir="./my_expertise",
    extensions=[".md", ".txt", ".pdf", ".docx"]
)

print(f"{len(docs)} documents collected")

# Saving for later processing
with open("raw_corpus.json", "w") as f:
    json.dump(docs, f, ensure_ascii=False, indent=2)

import re
from typing import Optional

def clean_document(text: str) -> str:
    # Remove repetitive headers/footers
    text = re.sub(r"Page \d+ sur \d+", "", text)
    text = re.sub(r"Confidential - Do not distribute", "", text)

    # Normalize spaces and line breaks
    text = re.sub(r"\n{3,}", "\n\n", text)
    text = re.sub(r"[ \t]+", " ", text)

    return text.strip()


def chunk_document(
    text: str,
    chunk_size: int = 1000,
    overlap: int = 200,
    metadata: Optional[dict] = None
) -> list[dict]:
    chunks = []
    sentences = re.split(r'(?<=[.!?])\s+', text)

    current_chunk = ""
    for sentence in sentences:
        if len(current_chunk) + len(sentence) > chunk_size and current_chunk:
            chunks.append({
                "text": current_chunk.strip(),
                "metadata": metadata or {},
                "char_count": len(current_chunk.strip())
            })
            # Keep overlap
            words = current_chunk.split()
            overlap_text = " ".join(words[-overlap // 5:])
            current_chunk = overlap_text + " " + sentence
        else:
            current_chunk += " " + sentence

    if current_chunk.strip():
        chunks.append({
            "text": current_chunk.strip(),
            "metadata": metadata or {},
            "char_count": len(current_chunk.strip())
        })

    return chunks

metadata = {
    "source": "technical_guide_v3.md",
    "category": "procedure",        # procedure, advice, analysis, client_case
    "domain": "real_estate_law",    # your professional domain
    "confidence": "high",            # high, medium, low
    "date": "2025-06-15",
    "author": "main_expert"
}

User question
        ↓
   [Embedding]  →  Vector search in your corpus
        ↓
   Top-K relevant documents
        ↓
   [LLM + Expert System Prompt]  →  Contextualized response
        ↓
   Response with source citations

import chromadb
from chromadb.utils import embedding_functions
import requests

# Initialize the vector database
client = chromadb.PersistentClient(path="./avatar_db")

# Use a performant embedding model
embedding_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
    model_name="intfloat/multilingual-e5-large"
)

collection = client.get_or_create_collection(
    name="professional_expertise",
    embedding_function=embedding_fn,
    metadata={"hnsw:space": "cosine"}
)

# Index the corpus
def index_corpus(chunks: list[dict]):
    collection.add(
        documents=[c["text"] for c in chunks],
        metadatas=[c["metadata"] for c in chunks],
        ids=[f"chunk_{i}" for i in range(len(chunks))]
    )
    print(f"{len(chunks)} chunks indexed")

# Search + generation
def ask_avatar(question: str, n_results: int = 5) -> str:
    # Search for relevant passages
    results = collection.query(
        query_texts=[question],
        n_results=n_results
    )

    # Build context
    context = "\n\n---\n\n".join(results["documents"][0])
    sources = [m["source"] for m in results["metadatas"][0]]

    # Call LLM via OpenRouter
    response = requests.post(
        "https://openrouter.ai/api/v1/chat/completions",
        headers={
            "Authorization": "Bearer YOUR_API_KEY",
            "Content-Type": "application/json"
        },
        json={
            "model": "anthropic/claude-sonnet-4-20250514",
            "messages": [
                {"role": "system", "content": SYSTEM_PROMPT},
                {
                    "role": "user",
                    "content": f"PROFESSIONAL CONTEXT:\n{context}\n\n"
                               f"QUESTION:\n{question}"
                }
            ],
            "temperature": 0.3
        }
    )

    answer = response.json()["choices"][0]["message"]["content"]
    return f"{answer}\n\nSources: {', '.join(set(sources))}"

SYSTEM_PROMPT = """You are the expert AI avatar of [Your Name], [Your Professional Title] 
with [X] years of experience in [domain].

## Your Role
- Answer professional questions with precision and pragmatism
- Cite your sources when relying on the provided context
- Admit when a question is beyond your area of expertise

## Your Communication Style
- Tone: professional but accessible, like a senior consultant in a meeting
- Structure: organized responses with numbered key points
- Examples: systematically illustrate with concrete cases
- Jargon: use professional vocabulary but explain complex terms

## Strict Rules
- NEVER invent legal/technical/financial information
- If the provided context doesn't contain the answer, say so clearly
- Always remind that your responses don't replace personalized advice
- Don't answer questions outside your area of expertise
"""

# Example question to the dev avatar
question = "How do we handle pagination on our REST API?"

# The avatar retrieves from the corpus:
# - The API v2 architecture document
# - The Jira ticket about pagination refactoring
# - The Slack message where you explained the cursor-based choice

# Avatar response:
# "On our API, we use cursor-based pagination since v2. The choice was made for endpoints with high volume