Entwickeln Sie einen sich selbst verbessernden GPT-6-Astra-Coding-Agenten mit Codebase-Memory

Ein frustrierender Aspekt beim Programmieren von Agenten ist, wie leicht ihnen zwischen den Sitzungen nützliche Zusammenhänge verloren gehen. Astra kann eine ganze Sitzung damit verbringen, einen Fehler zu beheben, und schließlich eine getestete Lösung finden. Wenn in einer späteren Sitzung ein ähnlicher Fehler auftritt, muss sie einen Großteil der Untersuchung wiederholen, da die frühere Lösung noch an die alte Sitzung gebunden ist.

Wir haben getestet, ob der externe Speicher eine bestätigte Korrektur in spätere Sitzungen übertragen kann – denselben Test, den wir bereits in unserer OpenClaw-Speicher-Plugin-Version durchgeführt hatten, bei der die Korrekturen eines Programmierassistenten auf dieselbe Weise beibehalten werden mussten. Im ersten Durchlauf behob Astra einen Fehler in der Ursprungskonfiguration und speicherte die Lösung in der Actian VectorAI-Datenbank. Bei einem zweiten Durchlauf wurden ein anderes Projekt und eine neue Antwortkette verwendet, dennoch wurde diese „ Aufzeichnung “ abgerufen, bevor überhaupt Code geändert wurde. Anschließend behob Astra den damit verbundenen Normalisierungsfehler, und alle vier ursprünglichen Abnahmetests wurden bestanden.

In diesem Tutorial erfahren Sie, wie Sie diese Speicherschicht mithilfe lokaler Einbettungen, VectorAI DB und zweier Funktionen erstellen, die Astra über die Responses-API zur Verfügung stehen – eine zum Suchen bestätigter Korrekturen und eine zum Speichern dieser Korrekturen.

Warum Astra eine separate Speicherschicht benötigt

GPT-6 Astra kann den Kontext über API-Aufrufe hinweg übertragen, allerdings muss die Anwendung dem System mitteilen, welche frühere Interaktion zur aktuellen gehört. In der Responses-API, previous_response_id verbindet ein Antwort an den vorherigen, sodass Astra denselben Thread fortsetzen kann. Für Konversationen, die über Sitzungen, Geräte oder Aufträge hinweg bestehen bleiben müssen, unterstützt dieselbe API auch dauerhafte „Conversation“-Objekte.

Beide Optionen sorgen dafür, dass eine Konversation fortgesetzt wird, von der die Anwendung bereits weiß, dass sie fortgesetzt werden soll. Außerdem verhindern sie, dass Astra jedes Mal, wenn ein bekannter Fehler auftritt, separate Antwortketten startet.

Der Verlauf erweist sich besonders dann als nützlich, wenn ein Mitarbeiter eine Information benötigt, die es nicht in die Arbeitszusammenfassung geschafft hat. BlackwellBoy nennt ein praktisches Beispiel:

„Wenn vor drei Stunden ein seltsamer Testfehler aufgetreten ist, der nicht in den Zusammenfassungsnotizen vermerkt wurde, kann Astra die Protokolle zurückverfolgen, um ihn zu finden.“

In diesem Beispiel geht es darum, ein früheres Ereignis in den Protokollen eines seit Langem laufenden Aufgabe zu finden. Außerdem wollten wir, dass spätere Sitzungen relevante Korrekturen aus anderen Arbeitsbereichen finden können. Daher haben wir jede bestätigte Korrektur in der VectorAI-Datenbank gespeichert, damit eine neue Antwortkette danach suchen kann.

So funktioniert die Speicherarchitektur

Wenn ein Test fehlschlägt, sendet Astra die Fehlerdetails an search_codebase_memory. Das Tool erstellt eine lokale Einbettung und durchsucht die VectorAI-Datenbank nach ähnlichen, bestätigten Lösungen. Jedes Ergebnis enthält die Fehlerart, die betroffenen Dateien, die Lösung und das Testergebnis.

Die Agent-Schleife gibt das Suchergebnis als function_call_output die über ihren call_id. Astra vergleicht anschließend jede gefundene Lösung mit dem aktuellen Code, bevor Änderungen vorgenommen werden. Wenn die Suche keine relevanten Ergebnisse liefert, wird die Untersuchung anhand der Projektdateien und der Testausgaben fortgesetzt.

Nach der Implementierung einer Korrektur führt Astra die Abnahmetests durch. Bei einem erfolgreichen Ergebnis kann es store_fix_memory, der den Fehler, die Lösung, die betroffenen Dateien und das überprüfte Ergebnis speichert. Der Handler lehnt unbestätigte Datensätze ab, sodass fehlgeschlagene Versuche bei späteren Suchvorgängen nicht berücksichtigt werden.

Bei dieser Reihenfolge erfolgt der Abruf aus dem Speicher vor der Bearbeitung und die Speicherung im Speicher erst nach erfolgreichen Tests. Da VectorAI DB die Datensätze außerhalb der Antwortkette speichert, können spätere Sitzungen die während früherer Arbeiten gespeicherten Korrekturen abrufen.

Für das Projekt sind ein Astra-Zugang, eine lokale VectorAI-DB-Instanz und ein lokales Embedding-Modell erforderlich.

Einrichten des Stacks

GPT-6 Astra ist in der kostenlosen API-Stufe nicht verfügbar, und die Abrechnung der API erfolgt unabhängig vom ChatGPT-Abonnement. Die Standardverarbeitung kostet derzeit 10 US-Dollar pro Million Eingabetoken und 50 US-Dollar pro Million Ausgabetoken. In unserem Experiment kosteten die Transportprüfung und zwei Live-Sitzungen etwa 0,15 US-Dollar. Ein neuer Durchlauf kann je nach Token-Verbrauch und der Anzahl der vom Agenten benötigten „Responses API“-Anfragen mehr oder weniger kosten.

Projektübersicht

Das Projekt trennt die Speicherschicht, die Astra-Agent-Schleife, die lokalen Codierungswerkzeuge und den „ Demo “-Workflow in eigene Dateien auf:

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 enthält das VectorAI-DB-Backend, den lokalen Embedder und die Speicher-Handler. Die „Responses“-API-Schleife befindet sich in astra_agent.py. Die main.py Der Einstiegspunkt verbindet diese Komponenten für eine einzelne Programmierungssitzung, während session_demo.py führt das kontrollierte sitzungsübergreifende Experiment durch. Die vollständige Implementierung, die Tests und die Projekte „ Demo “ sind verfügbar unter GitHub Lager.

VectorAI-Datenbank starten

Lade das aktuelle VectorAI-DB-Bild ab:

docker pull actian/vectorai:latest

Das Projekt startet die getestete Konfiguration über docker-compose.yml:

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

Die Compose-Datei ordnet den REST-Endpunkt zu http://localhost:16573, das ist die Adresse, die der „ Python “-Client verwendet.

Installieren Sie die Abhängigkeiten von „ Python “

Die festgehaltenen Paketversionen werden gespeichert in requirements.txt. Erstellen Sie im Stammverzeichnis von „ Lager “ in WSL2 die Umgebung und installieren Sie die Programme:

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 .

Kopieren Sie die Beispiel-Umgebungsdatei:

cp .env.example .env

Öffnen .env und fügen Sie Ihren OpenAI-API-Schlüssel hinzu:

OPENAI_API_KEY=your_api_key

Die übrigen Werte in .env.example Konfigurieren Sie die VectorAI-Datenbankverbindung, die Standard-API-Verarbeitung und das vom Projekt verwendete Ausgabenlimit.

Überprüfen Sie die Sammlung und die lokalen Einbettungen

vectoraidb_memory_tools.py erstellt die astra_codebase_memory Datensatz mit 384-dimensionalen Vektoren und Kosinus-Abstand:

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

Jeder gespeicherte Vektor enthält eine Nutzlast mit der Beschreibung der Korrektur, der Fehlerart, dem Dateipfad, dem bestätigten Ergebnis, dem Zeitstempel und der Sitzungs-ID.

Die gleiche Datei wird geladen sentence-transformers/all-MiniLM-L6-v2 auf dem „ CPU “ und normalisiert die erzeugten Vektoren:

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,
)

Das Modell wird beim ersten Laden heruntergeladen. Da es diese Embeddings lokal generiert, werden keine OpenAI-Embedding-Credits verbraucht.

Die abschließende Überprüfung läuft durch 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"

Der erste Befehl erstellt oder überprüft die Sammlung, speichert eine vorübergehende Korrektur und sucht danach. Sie können das zurückgegebene Ergebnis kopieren memory_id in den zweiten Befehl, der bestätigt, dass die Datei „ Aufzeichnung “ weiterhin vorhanden ist, und diese anschließend löscht.

Derzeit kann VectorAI DB „ eingebettet “-Korrekturdatensätze speichern und abrufen. Astra benötigt jedoch noch eine kontrollierte Möglichkeit, diese Datenbank zu nutzen – und genau hier kommen die beiden Speicher-Tools ins Spiel.

Die Gedächtnis-Tools erstellen

Beide Speicher-Tools befinden sich in vectoraidb_memory_tools.py und verwende dasselbe MemoryService. Der Dienst erhält beim Start des Agenten das VectorAI-DB-Backend, den lokalen Embedder und die aktuelle Sitzungs-ID.

Wenn bei Astra ein Fehler auftritt, ruft es search_codebase_memory bevor Sie Änderungen am Code vornehmen. Die Methode speichert die Fehlerbeschreibung lokal ab und durchsucht die VectorAI-Datenbank nach entsprechenden Lösungen:

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 enthält die Fehlermeldung oder die Details zum fehlgeschlagenen Test, nach denen Astra suchen soll. Das optionale error_type kann die Suche auf eine bestimmte Art von Fehler eingrenzen, während top_k Legt die maximale Anzahl der zurückgegebenen Treffer fest. Die Methode wandelt die „ abfragen “ in eine lokale Einbettung um und wartet, bis VectorAI DB die Suche abgeschlossen hat, bevor Astra fortfährt.

format_search_results wandelt die übereinstimmenden Datensätze in Text um, den die Responses-API als 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)

Jeder Treffer liefert Astra die frühere Korrektur sowie die Details, die erforderlich sind, um die Relevanz für das aktuelle Problem zu beurteilen. Der Ähnlichkeitswert stammt aus der VectorAI-Datenbank und wird im formatierten Ergebnis auf vier Dezimalstellen gerundet. Die Dateipfade und das Testergebnis liefern weiteren Kontext zur gespeicherten Korrektur.

Nach der Astra-Reparatur führt die Agent-Schleife die Tests des Projekts aus. Ein erfolgreiches Ergebnis wird zum bestätigten Ergebnis, das von 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

Die Agent-Schleife ruft confirm_outcome mit dem letzten bestandenen Testergebnis unmittelbar vor dem Speichern der Korrektur. store_fix_memory überprüft diesen Wert, generiert anhand der Sitzung und der Reparaturdetails eine ID, bettet die Beschreibung der Korrektur lokal ein und schreibt den abgeschlossenen „ Aufzeichnung “ in die VectorAI-Datenbank. Die generierte UUID wird zur ID des „ Aufzeichnung“ in der VectorAI-Datenbank. Wenn dieselbe Sitzung eine identische Speicheranforderung erneut versucht, wird dieselbe UUID generiert, sodass der vorhandene „ Aufzeichnung “ aktualisiert wird, anstatt ihn zu duplizieren.

Astra kann diese Methoden aufrufen, sobald sie als „Responses API“-Funktionswerkzeuge deklariert wurden. Die Definitionen in derselben Datei beschreiben die einzelnen Funktionen und die Argumente, die das Modell bereitstellen muss. Die Suchdefinition nutzt zudem die Option von Astra zum asynchronen Aufruf von Werkzeugen, wodurch das Modell seine Arbeit unabhängig fortsetzen kann, während die Anwendung die Speichersuche durchführt:

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,
        },
    },
)

Einstellung strict zu True sorgt dafür, dass jeder Funktionsaufruf innerhalb seines deklarierten Schemas bleibt. Astra muss jedes erforderliche Argument bereitstellen, und additionalProperties: False lehnt Felder ab, die nicht in der Definition enthalten sind. Das Suchwerkzeug wird als asynchron gekennzeichnet, da es auf die lokale Einbettung und die Abfrage in der VectorAI-Datenbank wartet.

Nachdem die Speicherfunktionen registriert wurden, kann die Agentenschleife die Aufrufe von Astra an MemoryService und jedes Ergebnis in die richtige Antwort umwandeln.

Die Agent-Schleife

Die Schleife in astra_agent.py legt fest, wann Astra die Speichersammlung durchsuchen, das Projekt überprüfen, eine Datei bearbeiten und eine bestätigte Korrektur speichern kann. Die Systemanweisungen legen diese Reihenfolge zu Beginn jeder Sitzung fest:

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."""

Die Anwendung ergänzt diese Anweisungen durch Überprüfungen innerhalb der Schleife. Jeder Aufruf von „run“ beginnt mit previous_response_id auf … einstellen None, sodass die erste API-Anfrage eine neue Antwortkette auslöst. Sobald Astra antwortet, speichert die Schleife die zurückgegebene Antwort-ID und fügt sie der nächsten Anfrage in derselben Sitzung hinzu.

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

Jedes Tool-Ergebnis enthält die call_id aus dem ursprünglichen Funktionsaufruf von Astra. Dadurch kann die nächste Antwort den Rückgabewert der richtigen Anfrage zuordnen. Das Speichergatter zeichnet auf, wann die Suchergebnisse Astra erreicht haben, während last_confirmed_outcome stellt das letzte bestandene Testergebnis zur Verfügung für store_fix_memory.

Das _dispatch Die Methode führt abschließende Speicherprüfungen durch. Sie vergleicht das von Astra gelieferte Ergebnis mit dem letzten erfolgreichen Testergebnis, bevor sie die Datei „ Aufzeichnung “ bestätigt und speichert:

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

Das Projekt sendet jede Nutzlast an die Responses-API mit requests.post. Der HTTP-Aufruf befindet sich innerhalb von RealHTTPResponsesTransport, das zudem überprüft, ob der Live-Modus, der API-Schlüssel und das Projektbudget konfiguriert wurden, bevor eine Anfrage gesendet wird:

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()

Führe eine Programmier-Sitzung durch

main.py bildet den Einstieg in eine Programmierungssitzung. Nach dem Laden der Einstellungen und der Überprüfung des Ausgabenlimits stellt es eine Verbindung zum VectorAI-DB-Backend, zum Speicherdienst, zu den Programmierwerkzeugen und zum „Responses API“-Transport her:

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,
)

Die vollständige main.py Die Datei im Verzeichnis „ Lager “ überprüft vor dem Start der Sitzung außerdem den Arbeitsbereich, den API-Schlüssel, die Verarbeitungsstufe, die Lauf-ID und die Genehmigung der Live-Kosten.

Das Szenario „Alpha“ enthält den absichtlich eingebauten Fehler, den der Agent untersuchen soll. Kopieren Sie es in einen Arbeitsordner, damit Astra die Dateien bearbeiten kann, ohne die Originalversion zu verändern:

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

Führen Sie den Agenten für das kopierte Projekt aus:

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

Das --approve-live-cost Das Flag bestätigt, dass die Sitzung bezahlte API-Anfragen an die Responses-API senden darf. Wenn der Lauf endet, main.py Gibt die Antwort von Astra, die Anzahl der API-Anfragen und Tool-Aufrufe sowie die Sitzungskosten aus.

Um das Gedächtnis über verschiedene Reaktionsketten hinweg zu testen, session_demo.py führt zwei Projekte mit derselben VectorAI-DB-Sammlung aus. Dabei werden für jedes Projekt ein neuer Speicherdienst und ein neuer Agent erstellt, während die Vollfunktion zudem die Validierung und Bereinigung übernimmt:

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
)

In der ersten Sitzung standen keine gespeicherten Korrekturen zur Verfügung. Astra führte die fehlgeschlagene Ursprungsprüfung auf Leerzeichen um die durch Kommas getrennten Werte zurück und aktualisierte app/service.py, und speicherte das Ergebnis, nachdem alle vier Tests erfolgreich abgeschlossen waren:

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.

In der zweiten Sitzung wechselten wir zu einem anderen Projekt und starteten eine neue Antwortkette. Bei der Speichersuche wurde die Datei „ Aufzeichnung “ aus dem ersten Durchlauf gefunden:

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

Dem Protokoll zufolge erfolgte der Abruf vor jeglichem relevanten Dateizugriff, jeglicher Diagnose oder jeglicher Codeänderung. Astra behob später einen separaten Normalisierungsfehler in app/service.py, wobei Werte wie Audit-Log zu audit_log. Da bei der Live-Ausführung auch eine Assertion in tests/test_service.py, haben wir die Korrektur der Anwendung noch einmal in einem neuen Arbeitsbereich überprüft, der die ursprünglichen, unveränderten Tests enthielt:

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

Das Bestehen aller vier ursprünglichen Abnahmetests im neuen Arbeitsbereich bestätigt die Behebung des Problems in der Anwendung aus Sitzung 2. Aus dem Protokoll geht zudem hervor, dass eine bestätigte Fehlerbehebung aus Sitzung 1 in einer separaten Antwortkette bei Astra eingegangen war, bevor die Diagnose und Bearbeitung begannen.

Sie können denselben Workflow mit zwei Sitzungen mit der VectorAI DB Community Edition ausführen; den vollständigen Code finden Sie im begleitenden Artikel „ Lager “.

Außerdem haben wir jeden Tool-Aufruf protokolliert, um festzustellen, ob sich der Speicherverbrauch auf den Arbeitsaufwand in späteren Sitzungen ausgewirkt hat.

Worin der Makler besser wird

Wir haben denselben Szenario-Beta- Aufgabe us in fünf Live-Astra-Sitzungen durchgeführt. Bei jeder Sitzung wurden ein neuer Arbeitsbereich und eine separate Antwortkette verwendet, sodass die erste Anfrage keine previous_response_id.

Sitzung 1 begann mit einer leeren VectorAI-DB-Sammlung und speicherte deren bestätigte Korrektur, nachdem die Tests erfolgreich abgeschlossen waren. Die Sitzungen 2 bis 5 starteten mit dieser „ Aufzeichnung “ als einzigem verfügbaren Speicher. Die von diesen späteren Sitzungen erstellten Datensätze wurden vor dem nächsten Durchlauf entfernt.

In jeder Sitzung wurden sieben Responses-API-Anfragen und neun lokale Tool-Aufrufe durchgeführt. Astra hat dies bearbeitet app/service.py und tests/test_service.py, daher haben wir jede Anwendungskorrektur in einem neuen Arbeitsbereich mit den ursprünglichen Tests überprüft. Alle vier Tests wurden jedes Mal erfolgreich bestanden.

In den Sitzungen 2 bis 5 wurde die in Sitzung 1 getestete Lösung wiederverwendet, obwohl neue Antwortketten gestartet wurden. Dies ist besonders nützlich, wenn ein Fehler einem ähnelt, den der Agent bereits zuvor gelöst hat. Neue Fehler und weiterreichende Entwurfsentscheidungen erfordern jedoch weiterhin eine Untersuchung des aktuellen Projekts.

Zum Abschluss

Unsere Testreihe mit fünf Durchläufen hat gezeigt, dass eine bestätigte Korrektur über die VectorAI-Datenbank von einer Astra-Reaktionskette in eine andere übertragen werden kann. In den Durchläufen 2 bis 5 wurden die in Durchlauf 1 gespeicherten Datensätze abgerufen, und jeder Durchlauf bestand die vier ursprünglichen Abnahmetests.

Mit der VectorAI DB Community Edition steht Ihnen eine lokale Datenbank zur Verfügung, mit der Sie denselben Ansatz bei Ihren eigenen Programmieraufgaben ausprobieren können. Laden Sie das Bild mit folgendem Befehl herunter:

docker pull actian/vectorai:latest

Von dort aus können Sie die Such- und Speichertools aus diesem Tutorial mit Astra verbinden, damit erfolgreiche Korrekturen gespeichert und in späteren Sitzungen wiedergefunden werden können.

Häufig gestellte Fragen

Erinnert sich GPT-6 Astra an frühere Sitzungen?

Nicht automatisch. previous_response_id Oder ein „Conversation“-Objekt kann den früheren Kontext beibehalten, doch eine unabhängige Antwortkette benötigt ein externes Speicherwerkzeug, um Korrekturen aus anderen Sitzungen abzurufen.

Wie füge ich einem GPT-6-Astra-Agenten persistenten Speicher hinzu?

Speichern Sie bestätigte Korrekturen außerhalb der Reaktionskette und stellen Sie Tools zum Suchen und Hinzufügen von Datensätzen bereit. In diesem Tutorial führt Astra vor der Bearbeitung eine Suche durch und speichert eine Korrektur erst, nachdem die Tests erfolgreich abgeschlossen wurden.

Kann ich mit GPT-6 Astra eine externe Vektordatenbank verwenden?

Ja. Definieren Sie „Responses API“-Funktionswerkzeuge, die die Datenbank durchsuchen und aktualisieren und anschließend die Ergebnisse als Werkzeugausgabe an Astra zurückgeben.

Was ist der Unterschied zwischen OpenAI’s file_search ein Tool und eine externe Vektordatenbank?

OpenAI’s file_search ist ein gehostetes Tool zur Suche in hochgeladenen Dateien. Eine externe Vektordatenbank ermöglicht Ihrer Anwendung die direkte Steuerung von Speicherung, Einbettungen, Metadaten, Filterung, Aktualisierungen und Löschungen.

Über den Autor

Über Alex Novotny

Alex Novotny ist Leiter des Bereichs „Developer Growth“ bei Actian, wo er die Strategie zur Einbindung von Entwicklern und zur Beschleunigung der Einführung der Daten- und KI-Technologien von Actian, darunter VectorAI DB, leitet. Mit über zehn Jahren Erfahrung in der Technologiebranche und mehr als fünf Jahren im Bereich Developer Relations verbindet Alex technische Inhalte, Entwickler-Advocacy, Community-Arbeit und Wachstum, um komplexe Technologien leichter zugänglich, verständlicher und einsetzbar zu machen. Zuvor war er als „Senior Developer Advocate“ bei Box tätig und entwickelte dort in Zusammenarbeit mit führenden Technologieplattformen KI- und RAG-Anwendungen, MCP-Integrationen, Entwicklertools sowie technische Inhalte. Seine Leidenschaft gilt der Gestaltung von Entwicklererfahrungen, die Neugier in die tatsächliche Nutzung und Prototypen in die Produktion verwandeln.