Créer un agent de programmation GPT-6 Astra capable d'auto-apprentissage, doté d'une mémoire de code

L'un des aspects frustrants de la programmation d'agents réside dans la facilité avec laquelle ils perdent des informations contextuelles utiles d'une session à l'autre. Astra peut passer une session entière à analyser un bug et aboutir à une solution testée. Lorsqu'un bug similaire apparaît lors d'une session ultérieure, elle peut être amenée à refaire en grande partie la même analyse, car la solution trouvée précédemment reste liée à l'ancienne session.

Nous avons vérifié si la mémoire externe pouvait conserver une correction validée lors de sessions ultérieures, à l’instar du test que nous avions effectué dans notre version du plugin de mémoire OpenClaw, où les corrections apportées par un assistant de codage devaient être conservées de la même manière. Lors du premier essai, Astra a résolu un bug de configuration d’origine et a stocké la solution dans la base de données Actian VectorAI. Un deuxième essai, portant sur un projet différent et utilisant une nouvelle chaîne de réponses, a néanmoins permis de récupérer cette enregistrement avant même de modifier le code. Astra a ensuite corrigé le bug de normalisation associé, et les quatre tests d’acceptation d’origine ont tous été réussis.

Ce tutoriel vous explique comment créer cette couche de mémoire à l'aide d'embeddings locaux, de VectorAI DB et de deux fonctions mises à disposition d'Astra via l'API Responses : l'une permet de rechercher des correctifs validés, et l'autre de les stocker.

Pourquoi Astra a besoin d'une couche mémoire distincte

GPT-6 Astra est capable de conserver le contexte d'un appel d'API à l'autre, mais l'application doit lui indiquer quelle interaction précédente correspond à l'interaction en cours. Dans l'API Responses, previous_response_id relie un réponse à la précédente, ce qui permet à Astra de poursuivre le même fil de discussion. Pour les conversations qui doivent être conservées d'une session à l'autre, d'un appareil à l'autre ou d'une tâche à l'autre, cette même API prend également en charge les objets « Conversation » persistants.

Ces deux options permettent de conserver une conversation que l'application sait déjà qu'elle souhaite poursuivre. Elles empêchent également Astra de lancer des chaînes de réponses distinctes chaque fois qu'elle rencontre un bug connu.

L'historique s'avère particulièrement utile lorsqu'un agent a besoin d'une information qui n'a pas été reprise dans le résumé de travail. BlackwellBoy donne un exemple concret:

«Si un échec de test inhabituel s'est produit il y a trois heures et n'a pas été consigné dans le résumé, Astra peut remonter dans les journaux pour le retrouver. »

Cet exemple consiste à retrouver un événement antérieur dans les journaux d'une instance de longue durée de tâche. Nous souhaitions également que les sessions ultérieures puissent trouver les corrections pertinentes issues d'autres travaux ; nous avons donc enregistré chaque correction validée dans la base de données VectorAI afin qu'une nouvelle chaîne de réponses puisse la rechercher.

Fonctionnement de l'architecture mémoire

Lorsqu'un test échoue, Astra envoie les détails de l'erreur à search_codebase_memory. L'outil génère une représentation locale et recherche dans la base de données VectorAI des solutions validées similaires. Chaque résultat indique le type d'erreur, les fichiers concernés, la solution et le résultat du test.

La boucle de l'agent renvoie le résultat de la recherche à Astra sous la forme d'un function_call_output lié à la demande initiale par le biais de son call_id. Astra compare ensuite tout correctif trouvé avec le code actuel avant d'apporter des modifications. Si la recherche ne donne aucun résultat pertinent, elle poursuit son analyse à l'aide des fichiers du projet et des résultats des tests.

Une fois la correction mise en œuvre, Astra exécute les tests d'acceptation. Si les tests sont réussis, cela lui permet d'appeler store_fix_memory, qui enregistre l'échec, la solution, les fichiers concernés et le résultat vérifié. Le gestionnaire rejette les enregistrements non confirmés, ce qui permet d'exclure les tentatives infructueuses des recherches ultérieures.

Cette séquence permet de faire précéder la récupération des données de l'édition et de faire suivre le stockage des données par la réussite des tests. Étant donné que VectorAI DB stocke les enregistrements en dehors de la chaîne de réponse, les sessions ultérieures peuvent récupérer les corrections enregistrées lors de travaux antérieurs.

Le projet nécessite un accès à Astra, une instance locale de la base de données VectorAI et un modèle d'embedding local.

Configuration de la pile

GPT-6 Astra n'est pas disponible dans le cadre du niveau d'API gratuit, et la facturation de l'API est distincte de celle de l'abonnement à ChatGPT. Le traitement standard coûte actuellement 10 $ par million de tokens d'entrée et 50 $ par million de tokens de sortie. Dans le cadre de notre expérience, la vérification du transport et les deux sessions en direct ont coûté environ 0,15 $. Une nouvelle exécution peut coûter plus ou moins cher en fonction du nombre de tokens utilisés et du nombre de requêtes « Responses API » dont l'agent a besoin.

Présentation du projet

Le projet sépare la couche mémoire, la boucle de l'agent Astra, les outils de codage locaux et l'workflow de démonstration en fichiers distincts :

astra-codebase-memory/
├── docker-compose.yml
├── requirements.txt
├── settings.py
├── vectoraidb_memory_tools.py
├── astra_agent.py
├── coding_tools.py
├── cost_guard.py
├──main.py
├── smoke_test.py
├── session_demo.py
├── demo_projects/
└── tests/

vectoraidb_memory_tools.py contient le backend de la base de données VectorAI, l'embedder local et les gestionnaires de mémoire. La boucle de l'API Responses se trouve dans astra_agent.py. Le main.py Le point d'entrée relie ces composants pour une seule session de codage, tandis que session_demo.py permet d'exécuter l'expérience contrôlée inter-sessions. L'implémentation complète, les tests et les projets de démonstration sont disponibles dans le GitHub dépôt.

Lancer VectorAI DB

Récupérer l'image actuelle de la base de données VectorAI :

docker pull actian/vectorai:latest

Le projet lance la configuration testée via docker-compose.yml:

docker compose -p astra-codebase-memory up -d
docker compose -p astra-codebase-memory ps

Le fichier Compose mappe le point de terminaison REST vers http://localhost:16573, qui est l'adresse utilisée par le client Python .

Installez les dépendances d'Python

Les versions des paquets épinglées sont stockées dans requirements.txt. À partir du répertoire racine « dépôt » dans WSL2, créez l'environnement et installez-les :

python3.12 -m venv .venv
.venv/bin/python -m pip install -r requirements.txt
.venv/bin/python -m pip install --no-deps --no-build-isolation -e .

Copiez le fichier d'environnement d'exemple :

cp .env.example .env

Ouvrir .env et ajoutez votre clé API OpenAI :

OPENAI_API_KEY=your_api_key

Les valeurs restantes dans .env.example configurer la connexion à la base de données VectorAI, le traitement via l'API standard et la limite de dépenses utilisée par le projet.

Vérifier la collection et les intégrations locales

vectoraidb_memory_tools.py crée le astra_codebase_memory ensemble de données avec des vecteurs à 384 dimensions et une distance cosinus :

created = await self._call(
    "collections_create",
    self.collection_name,
    {"vectors": {"size": EMBEDDING_DIMENSION, "distance": "Cosine"}},
    timeout=30.0,
)

Chaque vecteur stocké contient des données utiles comprenant la description de la correction, le type d'erreur, le chemin d'accès au fichier, le résultat confirmé, l'horodatage et l'identifiant de session.

Le même fichier s'ouvre sentence-transformers/all-MiniLM-L6-v2 sur l'processeur , puis normalise les vecteurs générés :

self._model = factory(self.model_name, device="cpu")

encoded = self._load_model().encode(
    clean_text,
    normalize_embeddings=True,
    show_progress_bar=False,
    convert_to_numpy=True,
)

Le modèle se télécharge lors de son premier chargement. Comme il génère ces représentations localement, il n'utilise pas de crédits d'embedding OpenAI.

La vérification finale s'effectue smoke_test.py:

.venv/bin/python -u smoke_test.py store
.venv/bin/python -u smoke_test.py verify --memory-id "PASTE_MEMORY_ID_FROM_STORE"

La première commande crée ou valide la collection, enregistre une solution temporaire et la recherche. Vous pouvez copier le résultat renvoyé memory_id dans la deuxième commande, qui vérifie que l'enregistrement a bien été conservée, puis la supprime.

À ce stade, VectorAI DB est capable de stocker et de récupérer des enregistrements de corrections « Embarqué ». Astra a encore besoin d'un moyen contrôlé d'utiliser cette base de données, et c'est là qu'interviennent les deux outils de mémoire.

Créer les outils de mémorisation

Ces deux outils de gestion de la mémoire se trouvent dans vectoraidb_memory_tools.py et utiliser le même MemoryService. Le service reçoit le backend VectorAI DB, l'embedder local et l'identifiant de la session en cours au démarrage de l'agent.

Lorsqu'Astra rencontre un problème, elle appelle search_codebase_memory avant de modifier le code. Cette méthode enregistre localement la description de l'erreur et recherche dans la base de données VectorAI les solutions correspondantes :

async def search_codebase_memory(
    self,
    query: str,
    *,
    error_type: str | None = None,
    top_k: int = 3,
) -> str:
    clean_query = _required_text(query, "query")
    _validate_search_arguments(top_k, error_type)

    embedding = await self.embedder.embed(clean_query)
    results = await self.backend.search(
        embedding,
        top_k=top_k,
        error_type=error_type,
    )

    return format_search_results(results)

query contient le message d'erreur ou les détails du test ayant échoué qu'Astra souhaite rechercher. Le paramètre facultatif error_type permet de restreindre la recherche à un type particulier de défaillance, tandis que top_k définit le nombre maximal de résultats renvoyés. La méthode convertit l'requête e en un encodage local et attend que VectorAI DB ait terminé la recherche avant qu'Astra ne poursuive son exécution.

format_search_results convertit les enregistrements correspondants en texte que l'API « Responses » peut renvoyer sous la forme d'un function_call_output:

def format_search_results(
    results: Sequence[MemorySearchResult],
) -> str:
    """Return compact readable evidence for a function-call output."""

    if not results:
        return "No relevant confirmed fixes found."

    blocks = []

    for index, result in enumerate(results, start=1):
        record = result.record
        blocks.append(
            "\n".join(
                (
                    f"Match {index} (score={result.score:.4f})",
                    f"Fix: {record.fix_description}",
                    f"Error type: {record.error_type}",
                    f"Files: {', '.join(record.file_paths)}",
                    f"Outcome: {record.outcome}",
                    f"Confirmed: {record.timestamp}",
                    f"Session: {record.session_id}",
                )
            )
        )

    return "\n\n".join(blocks)

Chaque correspondance fournit à Astra la solution antérieure ainsi que les détails dont elle a besoin pour évaluer sa pertinence par rapport au problème actuel. Le score de similarité provient de la base de données VectorAI et est arrondi à quatre décimales dans le résultat formaté. Les chemins d'accès aux fichiers et le résultat du test fournissent davantage de contexte sur la solution enregistrée.

Une fois les réparations effectuées par Astra, la boucle de l'agent exécute les tests du projet. Un résultat positif devient le résultat confirmé et accepté par store_fix_memory:

def confirm_outcome(self, outcome: str) -> None:
    self._confirmed_outcomes.add(
        _required_text(outcome, "outcome")
    )

async def store_fix_memory(
    self,
    *,
    fix_description: str,
    error_type: str,
    file_paths: Sequence[str],
    outcome: str,
    timestamp: datetime | None = None,
) -> str:
    clean_fix = _required_text(
        fix_description,
        "fix_description",
    )
    clean_error = _required_text(error_type, "error_type")
    clean_outcome = _required_text(outcome, "outcome")

    if clean_outcome not in self._confirmed_outcomes:
        raise UnconfirmedOutcomeError(
            "outcome was not confirmed by the current run"
        )

    if isinstance(file_paths, (str, bytes)):
        raise MemoryValidationError(
            "file_paths must be a sequence of paths"
        )

    paths = tuple(file_paths)
    confirmed_at = timestamp or datetime.now(UTC)

    if (
        confirmed_at.tzinfo is None
        or confirmed_at.utcoffset() is None
    ):
        raise MemoryValidationError(
            "timestamp must include a timezone"
        )

    memory_id = str(
        uuid5(
            NAMESPACE_URL,
            "\n".join(
                (
                    self.session_id,
                    clean_fix,
                    clean_error,
                    clean_outcome,
                    *paths,
                )
            ),
        )
    )

    record = MemoryRecord(
        memory_id=memory_id,
        fix_description=clean_fix,
        error_type=clean_error,
        outcome=clean_outcome,
        file_paths=paths,
        timestamp=confirmed_at.isoformat(),
        session_id=self.session_id,
        embedding=await self.embedder.embed(clean_fix),
    )

    await self.backend.upsert(record)
    return memory_id

La boucle de l'agent appelle confirm_outcome en indiquant le dernier résultat de test réussi juste avant d'enregistrer la correction. store_fix_memory vérifie cette valeur, génère un identifiant à partir de la session et des détails de la réparation, intègre localement la description de la correction, puis enregistre l’ enregistrement t terminée dans la base de données VectorAI. L’UUID généré devient l’identifiant de l’ enregistrementdans la base de données VectorAI. Si la même session effectue une nouvelle tentative pour une requête de stockage identique, elle génère le même UUID ; elle met donc à jour l’ enregistrement existante au lieu de la dupliquer.

Astra peut appeler ces méthodes dès lors qu’elles ont été déclarées en tant qu’outils de fonction de l’API Responses. Les définitions figurant dans le même fichier décrivent chaque fonction et les arguments que le modèle doit fournir. La définition de la recherche utilise également l’option d’appel asynchrone des outils d’Astra, ce qui permet au modèle de poursuivre son travail de manière indépendante pendant que l’application exécute la recherche en mémoire :

MEMORY_TOOL_DEFINITIONS = (
    {
        "type": "function",
        "name": "search_codebase_memory",
        "description": (
            "Search confirmed past debugging results "
            "using observed failure symptoms."
        ),
        "async": True,
        "strict": True,
        "parameters": {
            "type": "object",
            "properties": {
                "query": {"type": "string"},
                "error_type": {
                    "type": ["string", "null"]
                },
                "top_k": {
                    "type": "integer",
                    "minimum": 1,
                    "maximum": 10,
                },
            },
            "required": [
                "query",
                "error_type",
                "top_k",
            ],
            "additionalProperties": False,
        },
    },
    {
        "type": "function",
        "name": "store_fix_memory",
        "description": (
            "Store a debugging result after its outcome "
            "is confirmed by the current run."
        ),
        "strict": True,
        "parameters": {
            "type": "object",
            "properties": {
                "fix_description": {
                    "type": "string"
                },
                "error_type": {"type": "string"},
                "file_paths": {
                    "type": "array",
                    "items": {"type": "string"},
                },
                "outcome": {"type": "string"},
            },
            "required": [
                "fix_description",
                "error_type",
                "file_paths",
                "outcome",
            ],
            "additionalProperties": False,
        },
    },
)

Cadre strict à True garantit que chaque appel de fonction reste dans le cadre de son schéma déclaré. Astra doit fournir tous les arguments requis, et additionalProperties: False rejette les champs ne correspondant pas à la définition. L'outil de recherche est qualifié d'asynchrone car il attend le résultat de l'intégration locale et de la recherche dans la base de données VectorAI.

Une fois les fonctions de mémoire enregistrées, la boucle de l'agent peut transmettre les appels d'Astra à MemoryService et renvoyer chaque résultat dans la réponse appropriée.

La boucle de l'agent

La boucle dans astra_agent.py détermine à quel moment Astra peut effectuer une recherche dans la collection de mémoire, inspecter le projet, modifier un fichier et enregistrer une correction validée. Les instructions du système définissent cet ordre au début de chaque session :

SYSTEM_INSTRUCTIONS = """You are fixing a bug in a confined demonstration workspace.
Call search_codebase_memory using only observed failure symptoms before stating a diagnosis or
editing code. Wait for its function output. An empty result still completes the required search.
Use only the supplied coding tools. Store a fix only after run_tests returns exit code 0, and use
the exact confirmed_outcome string returned by that test call. Do not assume memory will help."""

L'application renforce ces instructions à l'aide de vérifications effectuées à l'intérieur de la boucle. Chaque appel à la fonction `run` commence par previous_response_id réglé sur None, de sorte que la première requête API déclenche une nouvelle chaîne de réponses. Une fois qu'Astra a répondu, la boucle conserve l'identifiant de réponse renvoyé et l'associe à la requête suivante au sein de la même session.

async def run(
    self,
    prompt: str,
    *,
    session_id: str | None = None,
) -> AgentRunResult:
    if not isinstance(prompt, str) or not prompt.strip():
        raise ValueError("prompt must be a non-empty string")

    run_session_id = session_id or str(uuid4())
    previous_response_id: str | None = None
    next_input: object = [
        {"role": "user", "content": prompt.strip()}
    ]
    memory_result_delivered = False
    pending_memory_delivery = False
    last_confirmed_outcome: str | None = None
    start_event_index = len(self.event_logger.events)
    run_model_requests = 0
    run_local_tools = 0

    while True:
        if run_model_requests >= self.limits.max_turns:
            raise AgentLimitError(
                "model turn limit reached"
            )

        payload: dict[str, Any] = {
            "model": "gpt-6-astra",
            "instructions": SYSTEM_INSTRUCTIONS,
            "input": next_input,
            "tools": self.tool_definitions,
            "reasoning": {"effort": "low"},
            "text": {"verbosity": "low"},
            "max_output_tokens": (
                self.limits.max_output_tokens
            ),
        }

        if previous_response_id is not None:
            payload["previous_response_id"] = (
                previous_response_id
            )

        if pending_memory_delivery:
            memory_result_delivered = True
            pending_memory_delivery = False

        response = await self._request(
            payload,
            run_session_id,
        )
        run_model_requests += 1

        response_id, output = self._validate_response(
            response
        )
        function_calls = [
            item
            for item in output
            if item.get("type") == "function_call"
        ]

        if not function_calls:
            if not memory_result_delivered:
                raise MemoryGateError(
                    "agent produced a diagnosis before "
                    "receiving memory output"
                )

            final_text = self._extract_text(output)

            return AgentRunResult(
                session_id=run_session_id,
                final_text=final_text,
                final_response_id=response_id,
                model_request_count=run_model_requests,
                local_tool_call_count=run_local_tools,
                events=tuple(
                    self.event_logger.events[
                        start_event_index:
                    ]
                ),
            )

        outputs: list[dict[str, str]] = []
        memory_ready_at_response_start = (
            memory_result_delivered
        )
        search_completed = False

        for item in function_calls:
            if (
                run_local_tools
                >= self.limits.max_tool_calls
            ):
                raise AgentLimitError(
                    "local tool-call limit reached"
                )

            (
                tool_output,
                confirmed_outcome,
                was_search,
            ) = await self._dispatch(
                item,
                session_id=run_session_id,
                response_id=response_id,
                memory_ready=(
                    memory_ready_at_response_start
                ),
                last_confirmed_outcome=(
                    last_confirmed_outcome
                ),
            )
            run_local_tools += 1

            if item.get("name") == "run_tests":
                last_confirmed_outcome = (
                    confirmed_outcome
                )

            if item.get("name") == "apply_edit":
                last_confirmed_outcome = None

            search_completed = (
                search_completed or was_search
            )

            outputs.append(
                {
                    "type": "function_call_output",
                    "call_id": str(item["call_id"]),
                    "output": tool_output,
                }
            )

        if search_completed:
            pending_memory_delivery = True

        previous_response_id = response_id
        next_input = outputs

Chaque résultat d'outil comporte le call_id à partir de l'appel de fonction d'origine d'Astra. Cela permet à la réponse suivante d'associer la valeur renvoyée à la requête correspondante. La barrière mémoire enregistre le moment où le résultat de la recherche a atteint Astra, tandis que last_confirmed_outcome conserve le dernier résultat de test réussi disponible pour store_fix_memory.

Les _dispatch Cette méthode effectue les vérifications finales avant l'enregistrement. Elle compare le résultat fourni par Astra au dernier résultat de test réussi avant de valider et d'enregistrer l'enregistrement:

elif name == "store_fix_memory":
    self._require_keys(
        arguments,
        {
            "fix_description",
            "error_type",
            "file_paths",
            "outcome",
        },
    )

    if not memory_ready:
        raise MemoryGateError(
            "memory storage attempted before "
            "memory result delivery"
        )

    if (
        last_confirmed_outcome is None
        or arguments["outcome"]
        != last_confirmed_outcome
    ):
        raise ConfirmedFixGateError(
            "store_fix_memory requires the latest "
            "passing-test confirmed_outcome"
        )

    self.memory_service.confirm_outcome(
        last_confirmed_outcome
    )
    result = {
        "memory_id": (
            await self.memory_service.store_fix_memory(
                **arguments
            )
        )
    }
    was_search = False
    confirmed_outcome = last_confirmed_outcome

Le projet envoie chaque charge utile à l'API « Responses » avec requests.post. L'appel HTTP se trouve à l'intérieur de RealHTTPResponsesTransport, qui vérifie également que le mode « live », la clé API et le budget du projet ont bien été configurés avant l'envoi d'une requête :

response = await asyncio.to_thread(
    requests.post,
    RESPONSES_URL,
    headers={
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    },
    json=payload,
    timeout=self.timeout_seconds,
)
response.raise_for_status()
result = response.json()

Organiser une session de programmation

main.py constitue le point d'entrée d'une session de codage. Après avoir chargé les paramètres et vérifié la limite de dépenses, il se connecte au backend de la base de données VectorAI, au service de mémoire, aux outils de codage et au transport de l'API Responses :

backend = ActianVectorAIBackend(
    settings.vectorai_url,
    collection_name=settings.vectorai_collection,
    grpc_url=settings.vectorai_grpc_url,
)
await backend.ensure_collection()

memory_service = MemoryService(
    backend,
    SentenceTransformerEmbedder(
        settings.embedding_model
    ),
    session_id=run_id,
)

transport = RecordedLiveTransport(
    RealHTTPResponsesTransport(
        guard,
        enabled=True,
        api_key=settings.api_key,
        processing_tier=settings.processing_tier,
    ),
    ledger=ledger,
    guard=guard,
    session_id=run_id,
    transcript_path=(
        evidence_dir
        / "private"
        / f"{run_id}-transcript.jsonl"
    ),
)

agent = AstraAgent(
    transport,
    memory_service,
    SafeCodingTools(workspace),
    limits=AgentLimits(
        max_turns=LIVE_MAX_TURNS,
        max_output_tokens=LIVE_MAX_OUTPUT_TOKENS,
    ),
)

result = await agent.run(
    prompt,
    session_id=run_id,
)

L'intégralité de main.py Le fichier situé dans le répertoire « dépôt » vérifie également l'espace de travail, la clé API, le niveau de traitement, l'ID d'exécution et l'autorisation relative aux coûts en temps réel avant de démarrer la session.

Le scénario Alpha contient le bug intentionnel sur lequel l'agent va enquêter. Copiez-le dans un dossier de travail afin qu'Astra puisse modifier les fichiers sans altérer la version originale :

mkdir -p runs/tutorial
cp -R demo_projects/templates/scenario_alpha runs/tutorial/scenario_alpha

Exécutez l'agent sur le projet copié :

.venv/bin/python main.py \
  --workspace runs/tutorial/scenario_alpha \
  --run-id tutorial-alpha \
  --approve-live-cost

Les --approve-live-cost L'indicateur confirme que la session est autorisée à envoyer des requêtes API « Responses » payantes. À la fin de l'exécution, main.py affiche la réponse d'Astra, le nombre de requêtes API et d'appels d'outils, ainsi que le coût de la session.

Pour évaluer la mémoire à travers différentes chaînes de réponse, session_demo.py exécute deux projets sur la même collection de bases de données VectorAI. Il crée un nouveau service de mémoire et un nouvel agent pour chaque projet, tandis que la fonction complète se charge également de la validation et du nettoyage :

for scenario, suffix in (
    ("scenario_alpha", "alpha"),
    ("scenario_beta", "beta"),
):
    session_id = f"{run_id}-session-{suffix}"
    workspace = reset_workspace(run_id, scenario)
    coding_tools = SafeCodingTools(workspace)

    transport = RecordedLiveTransport(
        transport_factory(),
        ledger=ledger,
        guard=guard,
        session_id=session_id,
        transcript_path=(
            evidence_dir
            / "private"
            / f"{run_id}-{suffix}-transcript.jsonl"
        ),
    )

    service = TrackedMemoryService(
        backend,
        embedder,
        session_id=session_id,
        tracker=tracker,
    )

    agent = AstraAgent(
        transport,
        service,
        coding_tools,
        limits=AgentLimits(
            max_turns=LIVE_MAX_TURNS,
            max_output_tokens=LIVE_MAX_OUTPUT_TOKENS,
        ),
        event_logger=EventLogger(
            evidence_dir
            / "private"
            / f"{run_id}-{suffix}-events.jsonl"
        ),
    )

    result = await agent.run(
        DEMO_PROMPT,
        session_id=session_id,
    )

new_chains = all(
    bool(transport.requests)
    and "previous_response_id"
    not in transport.requests[0]
    for transport in transports
)

Lors de la première session, il n'y avait aucun correctif enregistré sur lequel s'appuyer. Astra a identifié la cause de l'échec de la vérification de l'origine : des espaces autour des valeurs séparées par des virgules, puis a mis à jour app/service.py, puis j'ai enregistré le résultat une fois les quatre tests réussis :

Fixed `app/service.py` to trim whitespace around comma-separated allowed origins. The leading space caused the second origin to be rejected.

All 4 tests pass. Stored the confirmed fix result.

Pour la deuxième session, nous sommes passés à un autre projet et avons lancé une nouvelle chaîne de réponses. La recherche en mémoire a permis de retrouver l'enregistrement , issue de la première exécution :

Match 1 (score=0.1650)
Fix: Trim surrounding whitespace from each comma-separated APP_ALLOWED_ORIGINS value in configured_values so the second configured origin matches exactly and receives Access-Control-Allow-Origin.
Error type: AssertionError
Files: app/service.py
Outcome: pytest passed with exit code 0
Confirmed: 2026-09-15T12:11:51.934136+00:00
Session: live-session1-20260915-1207-session-alpha

D'après la transcription, cette récupération a eu lieu avant toute lecture de fichier pertinente, tout diagnostic ou toute modification du code. Astra a par la suite corrigé un autre bug de normalisation dans app/service.py, en convertissant des valeurs telles que Audit-Log à audit_log. En effet, l'exécution en production a également modifié une assertion dans tests/test_service.py, nous avons vérifié une nouvelle fois le correctif de l'application dans un nouvel espace de travail contenant les tests d'origine, qui n'avaient pas été modifiés :

....                                                                  [100%]
4 passed in 0.17s

La réussite des quatre tests d'acceptation initiaux dans l'espace de travail vierge confirme la correction apportée à l'application lors de la session 2. Le compte-rendu atteste également qu'une correction validée lors de la session 1 a été transmise à Astra dans une chaîne de réponses distincte avant le début du diagnostic et des modifications.

Vous pouvez suivre cette même « workflow » en deux sessions avec VectorAI DB Community Edition et le code complet disponible sur le site associé dépôt.

Nous avons également enregistré chaque appel de fonction pour déterminer si la mémoire avait une incidence sur la quantité de travail effectuée lors des sessions suivantes.

Ce que l'agent apprend à mieux faire

Nous avons exécuté le même scénario bêta tâche au cours de cinq sessions Astra en production. Chaque session utilisait un espace de travail nouveau et une chaîne de réponse distincte ; ainsi, sa première requête ne comprenait pas previous_response_id.

La session 1 a débuté avec une collection VectorAI DB vide et a enregistré la correction validée une fois les tests réussis. Les sessions 2 à 5 ont démarré avec cette « enregistrement » comme seule mémoire disponible. Les enregistrements créés par ces sessions ultérieures ont été supprimés avant l'exécution suivante.

Chaque session a généré sept requêtes API « Responses » et neuf appels à des outils locaux. Astra a modifié app/service.py et tests/test_service.py, nous avons donc vérifié chaque correction apportée à l'application dans un nouvel espace de travail contenant les tests d'origine. Les quatre tests ont été réussis à chaque fois.

Les sessions 2 à 5 ont réutilisé la solution testée lors de la session 1, bien qu'elles aient donné lieu à de nouvelles chaînes de réponses. Cela s'avère particulièrement utile lorsqu'une défaillance ressemble à une autre que l'agent a déjà résolue. Les nouvelles erreurs et les décisions de conception plus générales nécessitent toutefois encore une analyse approfondie du projet en cours.

Pour conclure

Nos cinq sessions ont montré qu'une correction validée pouvait passer d'une chaîne de réponse Astra à une autre via la base de données VectorAI. Les sessions 2 à 5 ont récupéré les enregistrements sauvegardés lors de la session 1, et chaque session a réussi les quatre tests d'acceptation initiaux.

VectorAI DB Community Edition vous offre une base de données locale pour tester cette approche dans le cadre de vos propres projets de programmation. Récupérez l'image à l'aide de la commande suivante :

docker pull actian/vectorai:latest

À partir de là, vous pouvez connecter les outils de recherche et de stockage présentés dans ce tutoriel à Astra afin que celui-ci puisse enregistrer les corrections réussies et les retrouver lors de sessions ultérieures.

Questions fréquemment posées

GPT-6 Astra se souvient-elle des sessions précédentes ?

Pas automatiquement. previous_response_id ou bien un objet « Conversation » peut conserver le contexte antérieur, mais une chaîne de réponses indépendante nécessite un outil de mémoire externe pour récupérer les corrections issues d'autres sessions.

Comment ajouter de la mémoire persistante à un agent Astra GPT-6 ?

Enregistrez les corrections validées en dehors de la chaîne de réponse et mettez à disposition des outils permettant de rechercher et d'ajouter des enregistrements. Dans ce tutoriel, Astra effectue une recherche avant toute modification et n'enregistre une correction qu'une fois les tests réussis.

Puis-je utiliser une base de données vectorielle externe avec GPT-6 Astra ?

Oui. Définissez des outils de fonction de l'API « Responses » qui effectuent des recherches et mettent à jour la base de données, puis renvoient les résultats à Astra en tant que sortie de l'outil.

Quelle est la différence entre le modèle d'OpenAI et file_search un outil et une base de données vectorielle externe ?

OpenAI’s file_search Il s'agit d'un outil hébergé permettant d'effectuer des recherches dans des fichiers téléchargés. Une base de données vectorielle externe permet à votre application de contrôler directement le stockage, les représentations vectorielles, l'métadonnées, le filtrage, les mises à jour et la suppression.

À propos de l'auteur

À propos de Alex Novotny

Alex Novotny est responsable de la croissance des développeurs chez Actian, où il dirige la stratégie visant à impliquer les développeurs et à accélérer l’adoption des technologies de données et d’IA d’Actian, notamment VectorAI DB. Fort de plus de dix ans d’expérience dans le secteur technologique et de plus de cinq ans dans les relations avec les développeurs, Alex allie contenu technique, promotion auprès des développeurs, communauté et croissance afin de rendre les technologies complexes plus faciles à découvrir, à comprendre et à mettre en œuvre. Ancien responsable senior de la promotion auprès des développeurs chez Box, il a développé des applications d’IA et de RAG, des intégrations MCP, des outils pour développeurs et du contenu technique en partenariat avec des plateformes s technologiques de premier plan. Il se passionne pour la création d’expériences développeurs qui transforment la curiosité en adoption et les prototypes en production.