Wiring an autogen rag agent group chat changes how your multi-agent system handles grounded responses. This guide shows you how to plug a retrieval-augmented agent into an existing AutoGen group chat without rewriting the orchestration layer. We assume you already run AutoGen 0.2.x and have a corpus worth querying.
Step 1: Install dependencies and build a local vector index
Start with a clean Python 3.10+ environment. You need AutoGen, a vector store, and an embedding model that runs locally to avoid per-call API cost during indexing.
pip install pyautogen faiss-cpu sentence-transformersA common mistake is embedding whole files. Split text into overlapping chunks of ~500 tokens so retrieval returns focused context. Below is a minimal chunker and indexer:
import os, glob, re
from sentence_transformers import SentenceTransformer
import faiss
import numpy as np
def chunk(text, size=400, overlap=50):
words = text.split()
for i in range(0, len(words), size - overlap):
yield " ".join(words[i:i+size])
docs = []
for path in glob.glob("./docs/*.txt"):
with open(path) as f:
for piece in chunk(f.read()):
docs.append(piece)
model = SentenceTransformer("all-MiniLM-L6-v2")
embeds = model.encode(docs, normalize_embeddings=True)
index = faiss.IndexFlatIP(embeds.shape[1])
index.add(np.array(embeds))
faiss.write_index(index, "docs.index")Persist docs alongside the index (e.g., pickle.dump(docs, open("docs.pkl","wb"))). For a working autogen rag agent group chat prototype this static index is enough; in production you would rebuild on a cron or use a managed store.
Step 2: Write a deterministic retrieval helper
The agent must not ask the LLM whether to retrieve—do that synchronously before the model sees the prompt. Wrap FAISS in a pure function that returns formatted context and a score so you can threshold low-confidence hits.
import faiss, numpy as np, pickle
from sentence_transformers import SentenceTransformer
model = SentenceTransformer("all-MiniLM-L6-v2")
index = faiss.read_index("docs.index")
docs = pickle.load(open("docs.pkl","rb"))
def retrieve(query: str, k: int = 3, min_score: float = 0.3) -> str:
q_emb = model.encode([query], normalize_embeddings=True)
scores, idx = index.search(np.array(q_emb), k)
out = []
for sc, i in zip(scores[0], idx[0]):
if i != -1 and sc >= min_score:
out.append(f"[doc {i} | score {sc:.2f}]\n{docs[i]}")
return "\n---\n".join(out) if out else "NO_RELEVANT_DOCS"Keep this function side-effect free. You can unit test it with a fixed query and assert that expected chunk IDs appear. That test will survive LLM provider changes.
Step 3: Create the RAG agent with context injection
AutoGen ships a RetrieveAssistantAgent in autogen.agentchat.contrib, but it is coupled to a RetrieveUserProxyAgent and assumes a two-agent retrieve-then-chat loop. In a group chat with multiple speakers, that coupling breaks speaker selection. The robust path is a custom AssistantAgent that registers a reply hook.
from autogen import AssistantAgent
class RAGAssistant(AssistantAgent):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.register_reply(
trigger=AssistantAgent,
reply_func=self._rag_reply,
position=1, # run before the default LLM reply
)
def _rag_reply(self, recipient, messages, sender, config):
# Use the most recent user-facing task, not intermediate agent chatter
task = messages[-1]["content"]
context = retrieve(task)
augmented = f"{self.system_message}\n\nRetrieved context:\n{context}"
original = self.system_message
self.system_message = augmented
reply = self.generate_reply(messages=messages, sender=sender)
self.system_message = original
return True, replyThis keeps the group chat manager agnostic. The RAG agent looks like any participant, but its replies are grounded. If retrieve returns NO_RELEVANT_DOCS, the system message explicitly says so, which prevents hallucinated citations.
Step 4: Define the rest of the group and the chat manager
A useful autogen rag agent group chat needs a task initiator, a domain expert (the RAG agent), and at least one reviewer. Below we add a coder and a critic, plus a silent user proxy to satisfy AutoGen’s requirement that a group chat be initiated by a proxy.
import os
from autogen import UserProxyAgent, GroupChat, GroupChatManager
llm_config = {
"config_list": [{
"model": "gpt-4o-mini",
"api_key": os.environ["OPENAI_API_KEY"],
}],
"cache_seed": 42,
}
rag_agent = RAGAssistant(
name="rag_agent",
system_message="You answer using provided docs. Cite the doc id.",
llm_config=llm_config,
)
coder = AssistantAgent(
name="coder",
system_message="You write Python to implement solutions.",
llm_config=llm_config,
)
critic = AssistantAgent(
name="critic",
system_message="You flag ungrounded claims and logical gaps.",
llm_config=llm_config,
)
user_proxy = UserProxyAgent(
name="admin",
human_input_mode="NEVER",
code_execution_config=False,
)
group = GroupChat(
agents=[user_proxy, rag_agent, coder, critic],
messages=[],
max_round=10,
speaker_selection_method="auto",
)
manager = GroupChatManager(group=group, llm_config=llm_config)If you route through an OpenAI-compatible gateway such as n4n.ai, you can swap the base_url in config_list and gain automatic fallback when a provider is rate-limited without changing agent code. The group topology stays identical.
Step 5: Kick off the chat and verify retrieval
Initiate from the user proxy with a task that requires the corpus:
user_proxy.initiate_chat(
manager,
message="Our API returns 429 on burst traffic. How should we back off? Use the docs.",
)Verification should be mechanical, not visual. Add a log line inside retrieve and assert the RAG agent’s first response contains a doc id:
import logging
logging.basicConfig(level=logging.INFO)
# inside retrieve(): logging.info("RAG query: %s", query[:60])
# After chat completes:
rag_msgs = [m for m in group.messages if m["name"] == "rag_agent"]
assert any("doc" in m["content"] for m in rag_msgs), "RAG agent did not cite docs"If the assertion fails, check register_reply position. A position greater than 1 may let the default LLM reply fire first, skipping your hook. Also confirm speaker_selection_method="auto" actually selects rag_agent early; you can force it by setting group.speaker_selection_method to a custom function that prioritizes RAG on the first round.
Step 6: Make retrieval robust in multi-turn dialogue
Group chats interleave messages. The naive _rag_reply above uses only the last message, which might be a critic’s nitpick. Scan backward for the most recent task owner:
def _latest_task(messages):
for m in reversed(messages):
if m.get("name") == "admin" or m["role"] == "user":
return m["content"]
return messages[-1]["content"]
# In _rag_reply replace:
task = _latest_task(messages)This stops the RAG agent from retrieving on its own intermediate thoughts or on the coder’s diff snippets. It also avoids feeding the embedding model irrelevant code blocks, which would dilute similarity search.
Step 7: Production hardening
- Cache embeddings. Recomputing on every turn wastes latency and money. Load the FAISS index once at process start.
- Set
nprobefor larger indexes.index.nprobe = 8trades recall for speed on big corpora. - Use stable cache keys. If your gateway honors provider cache-control hints, prefix the static doc context with a fixed string so repeated retrievals hit the provider’s prompt cache. Gateways like n4n.ai provide per-token usage metering, so you can attribute the RAG augmentation overhead to the rag_agent specifically and tune
kto control cost. - Timeout the retriever. Wrap
retrievein a 200ms deadline; on miss, returnNO_RELEVANT_DOCSrather than blocking the group chat. - Isolate the system message mutation. The temporary swap in
_rag_replyis not thread-safe. If you run concurrent groups, deep-copy the agent or use a per-call context var.
The autogen rag agent group chat is now grounded, observable, and resilient to provider hiccups. You can extend the retriever to hybrid BM25+vector search or add a reranker without touching the group topology—only the retrieve function changes.