Multi-agent systems: what we improved

The first post covered what CrewAI gives you and what you have to build yourself to make a multi-agent system production-worthy. This one is about what we improved after it was live: four things that weren’t in the original design but turned out to be necessary once real users started using it.

Managing asyncio boundaries: CrewAI Flows

The crew execution pipeline created a subtle event-loop conflict. crew.kickoff() manages its own loop internally and is fine. But run_build() calls agent.execute_task() directly and execute_task() raises if it detects a running event loop, even one it didn’t create. Running inside asyncio.run() at the top of a Celery task and then calling execute_task() deep in the call stack causes it to fire.

The fix is CrewAI’s Flow abstraction. A Flow wraps the crew execution pipeline in an explicit step graph: @start for context loading, @listen for the crew run, @listen for finalization. The steps are async def and run on the caller’s event loop via flow.akickoff(), avoiding the “Future attached to a different loop” error that occurred when nested asyncio.run() calls created competing loops.

class PlanningFlow(Flow):
    @start()
    async def load_context(self) -> dict[str, Any]:
        self.context = await _load_crew_context(self.task_payload)
        return self.context

    @listen(load_context)
    async def run_crew(self, context: dict[str, Any]) -> dict[str, Any]:
        # _run_crew must run in a thread executor — agent.execute_task()
        # raises if called from within a running event loop.
        loop = asyncio.get_event_loop()
        self.crew_output = await loop.run_in_executor(
            None, _run_crew, context, self.task_payload, None
        )
        return self.crew_output

    @listen(run_crew)
    async def finalize(self, crew_output: dict[str, Any]) -> dict[str, Any]:
        self.crew_output = await _finalize_crew_output(...)
        return self.crew_output

The crew run goes into run_in_executor because agent.execute_task() has a stricter event-loop check than crew.kickoff(). Moving it to a thread means there’s no running loop in that thread, so the check passes.

The Flow also gave us a natural place to add a progress heartbeat thread that publishes thinking events to Redis while the crew is running. Without it, the SSE stream goes silent for up to 20 seconds and the frontend has no signal the backend is still alive.

heartbeat_thread = threading.Thread(
    target=_publish_progress_heartbeat,
    args=(self.task_payload.task_id, heartbeat_stop),
    daemon=True,
)
heartbeat_thread.start()
try:
    self.crew_output = await loop.run_in_executor(None, _run_crew, ...)
finally:
    heartbeat_stop.set()
    heartbeat_thread.join(timeout=0.2)

The heartbeat is a daemon thread so it doesn’t block shutdown. heartbeat_stop is a threading.Event the finally block sets regardless of success or failure.

Grounding the orchestrator with destination knowledge

The first production failure that surprised us wasn’t a hallucination in the obvious sense. The orchestrator would propose three cities for a two-night trip, or suggest visiting Rajasthan in June without mentioning that it’s peak heat season. The model had general knowledge of these facts, but it wasn’t reliably applying them when constructing trip proposals.

The fix is a knowledge base, not a better prompt.

CrewAI’s Knowledge system, backed by ChromaDB, gave us a way to give the orchestrator grounded, queryable access to trip planning rules and destination-specific guidance: how many cities are feasible for a given duration, which months are peak vs. low season, which trip types each destination suits.

The embedder is fastembed with BAAI/bge-small-en-v1.5, running locally inside the Celery worker — no external API call, no latency spike, no per-query cost:

class _FastembedEmbedder(CustomEmbeddingFunction):
    def __call__(self, input: Documents) -> Embeddings:
        model = _get_fastembed_model()
        return [v.tolist() for v in model.embed(list(input))]

The knowledge is scoped per destination. When the orchestrator is handling a Goa trip, it queries a ChromaDB collection that combines the global trip-planning rules with a Goa-specific file from knowledge/destinations/goa.md. When the destination is Kerala, it gets the Kerala file instead. Separate collections per region_slug prevent queries from cross-contaminating between destinations:

@functools.lru_cache(maxsize=32)
def build_destination_scoped_knowledge(region_slug: str) -> Knowledge:
    collection_name = f"veho_{region_slug}"
    # global rules + destination-specific file
    sources = [global_rules, global_dest_patterns, dest_file]
    storage = KnowledgeStorage(
        collection_name=collection_name,
        embedder=CustomProvider(embedding_callable=_FastembedEmbedder),
    )
    return Knowledge(collection_name=collection_name, sources=sources, storage=storage)

The per-slug cache means repeated crew instantiations for the same destination don’t re-embed. The model is loaded once per worker process via _get_fastembed_model().

The practical result: the orchestrator stopped proposing infeasible itineraries. It knew, from the knowledge base, that a 2-night trip can have at most 2 cities, that Rajasthan in June needs a specific disclaimer, that a solo trip to a beach destination has different city-count constraints than a family trip to the hills. This is knowledge the model technically has but wasn’t consistently applying from a prompt alone.

The lesson: when an agent keeps making the same category of error despite prompt instructions, the problem usually isn’t the prompt — it’s that the model needs a reliable retrieval path to structured constraints, not just a longer backstory.

Cutting data collection from four turns to one

The original picker system injected one UI component per turn: ask trip type → user answers → ask pax count → user answers → ask dates → user answers. For a new user with no existing trip context, that’s four separate round trips before the conversation can move forward.

We added multi_picker_group: when two or more independent fields are missing simultaneously, inject them together as a tabbed component instead of sequentially.

The logic in inject_pickers checks how many independent fields (trip_type, pax, dates, duration) are unknown before deciding what to show:

_INDEPENDENT_PICKERS = ("trip_type_picker", "pax_picker", "date_picker", "duration_picker")

missing: list[dict] = []
for ptype in _INDEPENDENT_PICKERS:
    spec = _build_picker_spec(ptype, trip_context, region_info_raw, user_message)
    if spec:
        missing.append(spec)

if len(missing) >= 2:
    crew_output["ui_components"].append({
        "type": "multi_picker_group",
        "label": "Tell us about your trip",
        "pickers": missing,
    })
    return crew_output

This only fires for genuinely independent fields. Dependent fields child_ages (only relevant once we know there are children), vibe (only relevant once we know the trip type), city routing combos (only relevant after destination is picked) still inject one at a time in priority order, because showing them prematurely doesn’t make sense.

The question scrubbing logic runs before the multi-picker check. When a picker is injected for a field, any question component the LLM produced asking about that same field gets stripped. Without this, the user sees both the LLM asking “How many adults will be travelling?” and a pax picker redundant and confusing.

First-turn cold start went from four exchanges to one.

Bridging CrewAI’s memory protocol to pgvector

CrewAI’s unified Memory system defines a StorageBackend protocol: a synchronous interface for save, search, delete, update, and related operations. The problem: we already had user preference data stored in a pgvector memory_embeddings table, accumulated over months. We didn’t want two memory stores or a migration.

The solution is a PostgresVectorStorageBackend that implements the protocol against the existing table.

The protocol is sync. CrewAI’s UnifiedMemory calls save() and search() synchronously, on a thread-pool thread for saves and on the caller thread for reads. The underlying database operations are async (SQLAlchemy asyncpg). The bridge is asyncio.run():

class PostgresVectorStorageBackend(StorageBackend):
    def save(self, records: list[MemoryRecord]) -> None:
        _run_async(self._async_save(records))

    def search(self, query_embedding, ...) -> list[tuple[MemoryRecord, float]]:
        return _run_async(self._async_search(query_embedding, ...))

asyncio.run() works here because the saves happen on thread-pool threads that have no event loop bound — asyncio.run() creates a fresh loop, runs the coroutine, closes the loop. This costs a few hundred microseconds per call, which is acceptable for a memory store that batches writes through CrewAI’s encoding pipeline.

The field mapping between MemoryRecord and the existing MemoryEmbedding model:

MemoryRecord field	MemoryEmbedding column
id	id (UUID)
content	content
scope	derived: /user/{user_id}
categories	[memory_type.value]
importance	weight (0.0–1.0)
embedding	embedding (384-dim pgvector)

The scope path /user/{uuid} follows CrewAI’s convention for user-scoped memories. The backend is single-user — instantiated once per Celery task with the session’s user_id — so it only reads and writes records belonging to that user.

The _async_search implementation does a full table scan filtered to the user, then computes cosine similarity in Python:

async def _async_search(self, query_embedding, ...) -> list[...]:
    async with self._session_factory() as db:
        stmt = select(MemoryEmbedding).where(
            MemoryEmbedding.user_id == uuid.UUID(self._user_id)
        )
        rows = (await db.execute(stmt)).scalars().all()

    results = []
    for row in rows:
        score = _cosine_similarity(query_embedding, list(row.embedding))
        if score >= min_score:
            results.append((self._to_record(row), score))
    results.sort(key=lambda r: r[1], reverse=True)
    return results[:limit]

This is intentionally per-user record counts are small enough that a full table scan with Python-side similarity is faster than the round-trip cost of a pgvector ANN index query. If that changes, the index query is a one-line swap in _async_search.

The outcome: CrewAI’s memory system writes to the same table the existing memory extract Celery task writes to. No migration, no dual stores, no sync between them — they share the same rows.

---

Each of these addressed a specific failure mode or UX problem that only became visible once real users were using the system. The knowledge base stopped itinerary proposals that were technically valid JSON but operationally wrong. The multi-picker group eliminated unnecessary round trips. The memory backend preserved the existing preference data while adopting CrewAI’s new protocol.

The pattern is the same as in the first post: the framework handles agent execution. Everything that makes the system actually useful for users is built on top of it.