666ghj/MiroFish · error · RuntimeError

{len(self._failed_batches)} Zep activity batch(es) failed; s

Error message

{len(self._failed_batches)} Zep activity batch(es) failed; simulation graph ingestion is incomplete

What it means

After the worker drains and buffers flush, stop() checks self._failed_batches; if any activity batch could not be ingested into the Zep graph it raises RuntimeError('N Zep activity batch(es) failed; simulation graph ingestion is incomplete'). This is a deliberate fail-loud: the simulation's graph in Zep is missing data, so downstream graph analysis would be silently wrong.

Source

Thrown at backend/app/services/zep_graph_memory_updater.py:336

        # could enqueue after both the worker and final flush had exited.
        with self._acceptance_lock:
            self._running = False

        if self._worker_thread and self._worker_thread.is_alive():
            join_timeout = max(0.0, deadline - time.time())
            self._worker_thread.join(timeout=join_timeout)
            if self._worker_thread.is_alive():
                raise TimeoutError(
                    f"Zep updater worker did not stop within {join_timeout:.0f}s"
                )

        # The worker has drained the queue. Only now is it safe to flush
        # buffers; doing this before join loses an item already dequeued by the
        # worker but not yet buffered.
        self._flush_remaining(deadline=deadline)

        if self._failed_batches:
            raise RuntimeError(
                f"{len(self._failed_batches)} Zep activity batch(es) failed; "
                "simulation graph ingestion is incomplete"
            )

        self._wait_for_pending_episodes(deadline=deadline)
        
        logger.info(f"ZepGraphMemoryUpdater 已停止: graph_id={self.graph_id}, "
                   f"total_activities={self._total_activities}, "
                   f"batches_sent={self._total_sent}, "
                   f"items_sent={self._total_items_sent}, "
                   f"failed={self._failed_count}, "
                   f"skipped={self._skipped_count}")
    
    def add_activity(self, activity: AgentActivity):
        """
        添加一个agent活动到队列
        
        所有有意义的行为都会被添加到队列，包括：

View on GitHub (pinned to b5b53acc57)

Solutions

1. Check the logs for the per-batch failure reasons logged when each batch was first recorded as failed — that names the root cause.
2. Re-ingest the failed activities: replay actions.jsonl (or the recorded batch payloads) into the graph once the API issue (key/rate limit/network) is fixed.
3. Reduce batch size / increase retry budget for long simulations to avoid exhausting retries during transient outages.

Example fix

# before
updater.stop()  # RuntimeError: 3 Zep activity batch(es) failed ...

# after
try:
    updater.stop()
except RuntimeError as e:
    if "batch(es) failed" not in str(e):
        raise
    failed = updater.export_failed_batches()  # persist payloads
    quarantine_for_reingest(sim_id, failed)

try:
    updater.stop()
except RuntimeError as e:
    if "batch(es) failed" not in str(e):
        raise
    persist_failed_batches(sim_id)      # fallback: keep payloads
    enqueue_reingestion(sim_id)          # replay into Zep later

When it happens

Trigger: Any batch send to the Zep API failing past its retries during the run or the final flush (HTTP errors, auth expiry, rate limiting, payload size); then stop() (or the stop_simulation path that calls it) raises this instead of returning cleanly.

Common situations: Zep API key revoked mid-run; 429 rate limiting on large simulations; transient network drops whose retries also failed; oversized batches exceeding API limits.

Related errors

• Zep图谱写入未完整完成: {error}
• ZEP_API_KEY未配置
• Zep updater worker did not stop within {join_timeout:.0f}s
• Zep graph updater is not running
• Zep updater drain deadline elapsed

AI-assisted analysis of 666ghj/MiroFish@b5b53acc57 (2026-08-14). Data as JSON: /api/errors/daaae8c9ce170761. Report an issue: GitHub.