Introduction: The "Hallucination Dilemma" of Multi-Agent Systems
Have you ever encountered a scenario where a multi-agent system was asked to tailor a resume for a target position based on a candidate's CV, but the generated content included company names, project details, or even fabricated employment dates that were never mentioned in the real resume?
This is not your fault—it is the "nature" of Large Language Models (LLMs). When LLMs are asked to "create" content, they tend to fill information gaps using patterns from their training data, even if this information completely contradicts facts.
Most multi-agent frameworks adopt a reactive approach: generate content first, then detect hallucinations, and finally attempt to fix them. But the problem is: the hallucination has already occurred in the output, and fixing it often amounts to just patching up the result.
llamaindex-pse takes a different approach: placing RAG retrieval before generation. It allows the Planner and Specialist to retrieve real documents from the knowledge base as context before producing any content. This way, agents are "grounded" from the source—they don't need to fabricate because real data is right at hand.
Section 1: Why Choose LlamaIndex Workflow Over StateGraph?
Intuitive Approach: StateGraph + Conditional Edges
If you are familiar with multi-agent orchestration frameworks, you might think of LangGraph's StateGraph. It controls process branching using explicit conditional edges:
START → node_a → conditional_edge ├─ yes → node_b → ... └─ no → node_c → ...
Every process branch requires manually written judgment logic—code is filled with conditionals determining which edge to take when the Evaluator passes or when there are issues.
Practical Approach: Event-Driven + @step Decorator
llamaindex-pse uses LlamaIndex's Workflow, defining nodes through typed events and the @step decorator:
class PlannerEvent(Event):
"""Triggers the Planner step."""
task_input: str = ""
class SpecialistEvent(Event):
"""Triggers the Specialist step."""
task_input: str = ""
plan: str = ""
class EvaluatorEvent(Event):
"""Triggers the Evaluator step."""
artifact: str = ""
attempts: int = 0
class FixEvent(Event):
"""Triggers the Fix step."""
artifact: str = ""
issues: list = field(default_factory=list)
Each method decorated with @step receives an input event and returns the next event. The flow follows the path of the event returned by the node.
Why Is This Better?
- Retry loops are naturally expressed: The Evaluator returns
FixEventorStopEvent, eliminating the need for manual conditional edges. - Centralized and type-safe state: All state is encapsulated using a dataclass.
@dataclass
class PSEState:
"""PSE Workflow state (dataclass, passed via Context)."""
task_input: str = ""
task_data: dict = field(default_factory=dict)
plan: str = ""
artifact: str = ""
attempts: int = 0
fictitious: list = field(default_factory=list)
verified: list = field(default_factory=list)
eval_issues: list = field(default_factory=list)
max_retries: int = 3
# RAG retrieval context (automatically populated by planner / specialist)
rag_context: str = ""
All nodes share the same state object, reading and writing via ctx.store, avoiding implicit global state or complex state-passing mechanisms.
- More declarative code: Control flow equals code structure, making it clear at a glance.
Section 2: Division of Labor and Collaboration Among the Three PSE Roles
Planner: Intelligence Collector
The Planner's responsibility is to understand the task and produce an execution plan. Its unique feature is optional RAG enhancement:
@step
async def planner(self, ctx: Context, ev: PlannerEvent) -> SpecialistEvent:
"""Planner: RAG retrieval of market/JD intelligence + read context → produce execution plan."""
# RAG Enhancement: Prioritize planner_retriever (market/JD intelligence), fallback to general retriever
rag_ctx = ""
pr = self._planner_retriever or self._retriever
if pr:
rag_ctx = await _retrieve_context(pr, ev.task_input, self._rag_top_k, reranker=self._reranker)
if rag_ctx:
label = "Market Intelligence" if self._planner_retriever else "Reference Documents"
rerank_tag = " (reranked)" if self._reranker else ""
print(f" 📚 RAG Retrieval {label}{rerank_tag} {len(rag_ctx)} chars")
full_input = ev.task_input
if rag_ctx:
full_input += f"\n\n## Retrieved Reference Documents\n{rag_ctx}"
messages = []
if self._planner_prompt:
messages.append({"role": "system", "content": self._planner_prompt})
messages.append({"role": "user", "content": full_input})
plan = self._llm.chat(messages, stage="planner")
print(f"✅ Planning completed ({len(plan)} chars)")
state: PSEState = await ctx.store.get("state")
state.plan = plan
state.rag_context = rag_ctx
await ctx.store.set("state", state)
return SpecialistEvent(task_input=ev.task_input, plan=plan)
Note the key design here: The Planner does not start planning directly but first retrieves relevant documents. For resume customization tasks, it retrieves market intelligence for the target position; for technical article generation, it retrieves metadata from project source code.
Specialist: Executor
The Specialist is responsible for unfolding the plan into the final artifact. It also enjoys RAG enhancement:
@step
async def specialist(self, ctx: Context, ev: SpecialistEvent) -> EvaluatorEvent:
"""Specialist: RAG retrieval of resume source data + unfold plan into final artifact."""
full = (ev.task_input + "\n\n## Execution Plan\n" + ev.plan) if ev.plan else ev.task_input
# RAG Enhancement: Specialist uses retriever (resume source data); if Planner already retrieved, reuse it
state: PSEState = await ctx.store.get("state")
rag_ctx = state.rag_context
if not rag_ctx and self._retriever:
rag_ctx = await _retrieve_context(self._retriever, ev.task_input, self._rag_top_k, reranker=self._reranker)
if rag_ctx:
rerank_tag = " (reranked)" if self._reranker else ""
print(f" 📚 RAG Retrieval of Resume Source Data{rerank_tag} {len(rag_ctx)} chars")
state.rag_context = rag_ctx
await ctx.store.set("state", state)
if rag_ctx:
full += f"\n\n## Retrieved Reference Documents (Artifact must be based on this, fabrication is prohibited)\n{rag_ctx}"
messages = []
if self._specialist_prompt:
messages.append({"role": "system", "content": self._specialist_prompt})
messages.append({"role": "user", "content": full})
artifact = self._llm.chat(messages, stage="specialist")
if not artifact:
raise RuntimeError("Specialist did not output any content")
state.artifact = artifact
await ctx.store.set("state", state)
return EvaluatorEvent(artifact=artifact, attempts=state.attempts)
There is an important design detail here: If the Planner has already retrieved context, the Specialist will reuse it. This avoids duplicate retrieval and ensures both nodes see consistent information.
Evaluator: Merge Gate
The Evaluator is the most innovative point in the PSE pattern—it is a merge gate combining two layers of checks:
@step
async def evaluator(self, ctx: Context, ev: EvaluatorEvent) -> FixEvent | StopEvent:
"""Evaluator (Merge Gate): LLM review (first round only) + programmatic verify_fn hard check (every round)."""
state: PSEState = await ctx.store.get("state")
# 1) LLM Review (First Round Only)
eval_issues: list = []
if ev.attempts == 0 and self._evaluator_prompt:
scan = state.task_data.get("scan_result", {})
scan_str = json.dumps(scan, ensure_ascii=False, indent=2)
# RAG Cross-Validation: Supplement review basis with retrieved documents
rag_section = ""
if state.rag_context:
rag_section = f"\n\n## RAG Retrieved Reference Documents (Review Basis)\n{state.rag_context}"
full = (
f"## Artifact to Evaluate\n{ev.artifact}\n\n"
f"## Real Data (For Verification, Do Not Use Content Outside Artifact as Basis)\n{scan_str}"
f"{rag_section}"
)
resp = self._llm.complete(
self._evaluator_prompt + "\n\n" + full,
stage="evaluator",
)
text = str(resp)
eval_issues = _parse_eval_issues(text)
print(f" 🔍 Reviewer found {len(eval_issues)} issues")
# 2) Programmatic Check (Every Round)
if self._verify_fn is not None:
# Construct dict-like state for verify_fn (compatible with langgraph-pse signature)
state_dict = {
"task_input": state.task_input,
"task_data": state.task_data,
"plan": state.plan,
"artifact": state.artifact,
"attempts": state.attempts,
"fictitious": state.fictitious,
"verified": state.verified,
"eval_issues": state.eval_issues,
"max_retries": state.max_retries,
"rag_context": state.rag_context,
}
prog_bad, ok = self._verify_fn(state_dict)
else:
prog_bad, ok = [], []
all_bad = list(prog_bad) + list(eval_issues)
state.attempts += 1
state.fictitious = all_bad
state.verified = ok
state.eval_issues = eval_issues
await ctx.store.set("state", state)
print(f"\n{'=' * 60}")
print(f" Check (Round {state.attempts}) — Programmatic Pass: {len(ok)}, "
f"Programmatic Issues: {len(prog_bad)}, Review Issues: {len(eval_issues)}")
for b in all_bad:
print(f" ❌ {b}")
# Decision: Pass → Stop, Otherwise → Fix
if not all_bad or state.attempts > state.max_retries:
if not all_bad:
print(" ✅ Check Passed")
else:
print(" ⚠️ Max retries reached, stopping correction")
return StopEvent(result={"artifact": state.artifact, "state": state})
return FixEvent(artifact=state.artifact, issues=all_bad)
This design has two key points:
- LLM review runs only in the first round—to prevent subsequent rounds' LLM reviewers from being influenced by previous correction results.
- Programmatic check runs every round—this is the true "hard check," such as verifying if code references exist or if function names can be grep'd in the source code.
Fix: Corrector
When the Evaluator finds issues, the process enters the Fix node:
@step
async def fix(self, ctx: Context, ev: FixEvent) -> EvaluatorEvent:
"""Fix: Correct artifact based on identified issues. RAG context injected into correction prompt to prevent fabrication."""
state: PSEState = await ctx.store.get("state")
scan = state.task_data.get("scan_result", {})
scan_str = json.dumps(scan, ensure_ascii=False, indent=2)
# RAG Context: Corrections are also based on retrieved real documents
rag_section = ""
if state.rag_context:
rag_section = (
"\n\n**RAG Reference Documents (Must Base Corrections On These)**:\n"
f"{state.rag_context}\n"
)
print(" 🔄 Auto-correcting...")
prompt = (
"The following artifact has issues identified by programmatic checks. Please correct.\n\n"
f"**Issue List (Must Fix)**:\n" + "\n".join(f"- {i}" for i in ev.issues) + "\n\n"
"**Real Data (Must Base Corrections On This, Replace Wrong Numbers With Real Values,\n"
"Do Not Fabricate Or Delete Numbers)**:\n"
f"{scan_str}\n"
f"{rag_section}\n"
"**Rules**:\n"
"1. Only correct errors pointed out in the issue list, replacing wrong numbers with correct values from real data\n"
"2. Do not delete any correct numbers or content; keep the rest unchanged\n"
"3. **Absolutely Prohibit Deleting Entire Work Experience Or Project Entries**—If data for an entry is wrong, correct the data rather than deleting the whole entry\n"
"4. **Absolutely Prohibit Replacing Entries With Placeholders Or Comments** (e.g., 'Note: Source document did not provide')—Should correct based on real data\n"
"5. If the issue list mentions duration expressions like '20 years': **Delete** the duration expression directly, do not replace with other year numbers\n"
"6. If the issue list mentions projects missing start/end times: Append time range in format `| YYYY.MM-YYYY.MM` after the company name in the project title,\n"
"inferring time from the employment period at that company\n"
"7. Output the corrected complete artifact, do not output explanations\n\n"
f"## Current Artifact\n{ev.artifact}"
)
resp = self._llm.complete(prompt, stage="fix")
fixed = str(resp)
state.artifact = fixed
state.eval_issues = []
await ctx.store.set("state", state)
return EvaluatorEvent(artifact=fixed, attempts=state.attempts)
Note that the Fix node also injects RAG context. This means that even during the correction phase, the LLM works under the constraints of real documents, rather than "freely improvising."
Section 3: RAG Grounding—Core Design of Shifting Defense Left
Comparison: Reactive vs. Proactive
Traditional multi-agent framework processes follow this flow:
generate → detect(hallucination) → fix → detect → fix → ...
This is a reactive loop—hallucinations have already been generated, and then detection and repair occur.
The llamaindex-pse process is:
retrieve → generate(grounded) → verify(residual)
This is a proactive strategy—providing real documents before generation, allowing the LLM to generate based on facts. verify_fn still runs, but its role degrades from "primary defense line" to "safety net."
Source Code Implementation: Intelligent Retrieval
The implementation of RAG retrieval is also sophisticated:
async def _retrieve_context(retriever, query: str, top_k: int = 5, reranker=None) -> str:
"""Retrieve relevant documents using LlamaIndex retriever, concatenated into a context string.
If the query is too long, it exceeds the embedding model's context length, so the first 200 characters are truncated as retrieval keywords.
If a reranker is provided, retrieval results are first reordered by Cross-Encoder.
"""
# Truncate short query: Embedding models have limited context windows (e.g., Ollama snowflake-arctic-embed2 is only 8K tokens)
short_query = query[:200] if len(query) > 200 else query
nodes = retriever.retrieve(short_query)
if not nodes:
return ""
# Rerank: Cross-Encoder reordering to improve retrieval accuracy
if reranker is not None:
try:
nodes = reranker.postprocess_nodes(nodes, query_str=short_query)
except Exception:
pass # Fallback to original sorting if reranking fails
# Take top_k nodes, sorted by score descending
sorted_nodes = sorted(nodes, key=lambda n: n.score or 0, reverse=True)[:top_k]
parts = []
for i, node in enumerate(sorted_nodes, 1):
score = f" (score={node.score:.2f})" if node.score is not None else ""
parts.append(f"[{i}]{score}\n{node.get_content()}")
return "\n\n".join(parts)
Here are several practical details:
- Query Truncation: Embedding models have limited context windows; long queries are truncated to the first 200 characters.
- Reranker Support: Optional Cross-Encoder reordering to improve retrieval accuracy.
- Score Annotation: Each retrieval result includes a score, facilitating debugging.
Why Is It Effective?
The root cause of hallucinations is the LLM "daydreaming." Providing it with real documents as context eliminates the need to fabricate.
In the Planner phase, market intelligence or JD information is retrieved; in the Specialist phase, resume source data or project metadata is retrieved. Both nodes see the same context, ensuring information consistency.
Section 4: LLM Client Design Bypassing LlamaIndex Enum Validation
Problem: Limitations of the Framework Abstraction Layer
LlamaIndex's OpenAI wrapper has a hidden limitation: it calls openai_modelname_to_contextsize() to validate model names. For non-OpenAI official third-party models (such as various compatible models), this throws an error:
ValueError: Unknown model type: example-model
Intuitive Approach: Modify LlamaIndex Source Code or Wait for Upstream Fix
But this means either intruding into framework source code or waiting for upstream support—neither is a good choice.
Practical Approach: Directly Use the Underlying SDK
The solution in llamaindex-pse is to bypass LlamaIndex's abstraction layer and directly use the OpenAI SDK:
class SimpleLLM:
"""Lightweight LLM based on openai SDK (Bypasses LlamaIndex OpenAI validation).
Provides only the chat() and complete() interfaces needed for PSE workflow.
Includes connection retry (up to 3 times, 5 seconds apart) to handle occasional gateway disconnections.
Automatically records token consumption to global token_stats.
"""
def __init__(self, model: str, api_key: str, base_url: str):
self._model = model
self._client = openai.OpenAI(api_key=api_key, base_url=base_url)
def _call_with_retry(self, fn, max_retries=3, delay=5):
"""API call with retry to handle occasional gateway disconnections."""
for attempt in range(max_retries):
try:
return fn()
except openai.APIConnectionError as e:
if attempt < max_retries - 1:
print(f" ⚠️ Connection failed ({attempt+1}/{max_retries}), retrying in {delay}s...")
time.sleep(delay)
else:
raise
def chat(self, messages: list[dict], stage: str = "chat") -> str:
"""Chat completion, returning assistant content."""
resp = self._call_with_retry(lambda: self._client.chat.completions.create(
model=self._model,
messages=messages,
timeout=180,
))
# Record token consumption
if resp.usage:
token_stats.record(
stage=stage,
prompt_tokens=resp.usage.prompt_tokens or 0,
completion_tokens=resp.usage.completion_tokens or 0,
model=self._model,
)
return resp.choices[0].message.content or ""
Why Is This Better?
- Compatibility: Supports any LLM compatible with the OpenAI protocol (such as various third-party models).
- Simplicity: Does not force adaptation to the framework's abstraction layer; directly calls the underlying SDK when needed.
- Built-in Retry: Handles occasional gateway disconnections, retrying up to 3 times.
- Token Statistics: Automatically records token consumption and costs for each call.
class TokenStats:
"""Token consumption statistics collector, including price calculation."""
# Model Pricing Table: Yuan / Million Tokens (CNY)
PRICING = {
"deepseek-chat": {"input": 1.0, "output": 2.0}, # DeepSeek V3 Official Price
"deepseek-reasoner": {"input": 4.0, "output": 16.0}, # DeepSeek R1
"agnes-2.0-flash": {"input": 0.5, "output": 1.5}, # Agnes Gateway (Estimated)
}
def __init__(self):
self.calls: list[dict] = []
self.total_prompt_tokens = 0
self.total_completion_tokens = 0
self.total_tokens = 0
self.total_cost_cny = 0.0
def record(self, stage: str, prompt_tokens: int, completion_tokens: int, model: str):
total = prompt_tokens + completion_tokens
# Calculate cost
pricing = self.PRICING.get(model, {"input": 0, "output": 0})
cost = (prompt_tokens * pricing["input"] + completion_tokens * pricing["output"]) / 1_000_000
self.calls.append({
"stage": stage,
"prompt_tokens": prompt_tokens,
"completion_tokens": completion_tokens,
"total_tokens": total,
"model": model,
"cost_cny": cost,
})
self.total_prompt_tokens += prompt_tokens
self.total_completion_tokens += completion_tokens
self.total_tokens += total
self.total_cost_cny += cost
Section 5: Sandbox Tools and Security Design
Problem: Security Risks of Multi-Agent Systems
Multi-agent systems often need to access the file system or execute commands, but this introduces security risks:
- Agents may be induced to execute destructive commands like
rm -rf. - Agents may access sensitive files outside the project without authorization.
Intuitive Approach: Completely Disable Tools or Let Users Handle It
But this limits agent capabilities or pushes security responsibilities onto users.
Practical Approach: Restricted Sandbox
llamaindex-pse provides two sandboxed tools:
def read_file(path: str) -> str:
"""Read file content. Parameter path is the file path (restricted within the project directory)."""
p = Path(path).resolve()
if not str(p).startswith(str(_PROJECT_ROOT)):
return f"[Error] Path exceeds project scope: {path}"
if not p.exists():
return f"[Error] File does not exist: {path}"
if not p.is_file():
return f"[Error] Not a file: {path}"
return p.read_text(encoding="utf-8")
read_file ensures agents can only read files within the project root directory.
# Dangerous Command Fragment Blacklist (Rejected upon match, reducing risk of induced destructive commands)
_DANGEROUS_PATTERNS = [
r"\brm\s+-rf\b", r"\brm\s+-fr\b", r"\brm\s+-r\b", r"\brm\s+-R\b",
r"\bmkfs\b", r"\bdd\b\s+if=", r":\(\)\s*\{", r"\bshutdown\b",
r"\breboot\b", r"\bhalt\b", r"\bpoweroff\b", r">\s*/dev/sd",
r"\bchmod\b\s+-R\s+777\s+/", r"\bchown\b\s+-R\s+.*\s+/",
r"curl\b[^\n]*\|\s*(sh|bash)", r"wget\b[^\n]*\|\s*(sh|bash)",
r"\bnc\b[^\n]*-e\b",
]
def run_bash(command: str) -> str:
"""Execute bash command and return output (Restricted Sandbox: Destructive commands prohibited, working directory restricted to project root)."""
for pat in _DANGEROUS_PATTERNS:
if re.search(pat, command):
return (
f"[Rejected] Command matched dangerous pattern ({pat}), intercepted by sandbox."
"Please perform destructive operations manually if necessary."
)
try:
result = subprocess.run(
command, shell=True, capture_output=True, text=True,
timeout=30, cwd=str(_PROJECT_ROOT),
)
return result.stdout + "\n" + result.stderr
except Exception as e:
return f"[Error] {e}"
run_bash has a dangerous command blacklist and working directory restriction, with a timeout set to 30 seconds.
Why Is This Better?
- Principle of Least Privilege: Agents can only access resources within the project.
- Dangerous Command Interception: Blacklist covers common destructive operations.
- Transparent Feedback: Clearly states the reason when intercepted, rather than failing silently.
Section 6: Practical Demo—Resume Customization Task
Scenario Setup
Assume we want to use llamaindex-pse to implement a resume customization task: starting from real resume source data, customize a resume for a target position.
Configuration Loading
All configurations are loaded from environment variables:
class Settings:
OPENAI_API_KEY: str = os.getenv("OPENAI_API_KEY", "")
OPENAI_MODEL: str = os.getenv("OPENAI_MODEL", "")
OPENAI_BASE_URL: str = os.getenv("OPENAI_BASE_URL", "")
PSE_MAX_RETRIES: int = int(os.getenv("PSE_MAX_RETRIES", "3"))
# Agnes Gateway (Compatible with OpenAI protocol, independent key / base_url / model)
AGNES_KEY: str = os.getenv("AGNES_KEY", "")
AGNES_BASE_URL: str = os.getenv("AGNES_BASE_URL", "")
AGNES_MODEL: str = os.getenv("AGNES_MODEL", "")
# Embedding Model (Used for RAG indexing)
EMBEDDING_PROVIDER: str = os.getenv("EMBEDDING_PROVIDER", "openai") # "openai" | "ollama"
EMBEDDING_API_KEY: str = os.getenv("EMBEDDING_API_KEY", os.getenv("OPENAI_API_KEY", ""))
EMBEDDING_BASE_URL: str = os.getenv("EMBEDDING_BASE_URL", os.getenv("OPENAI_BASE_URL", ""))
EMBEDDING_MODEL: str = os.getenv("EMBEDDING_MODEL", "")
Prompt Loading
Prompts are dynamically loaded from the task directory:
def load_prompt(name: str, task: str | None = None) -> str:
"""Load system prompt for a specific role.
Prioritize reading tasks/<task>/prompts/<name>.md; return empty string if not found.
"""
if task:
prompt_path = PROMPTS_DIR / task / "prompts" / f"{name}.md"
if prompt_path.exists():
return prompt_path.read_text(encoding="utf-8")
return ""
This design allows each task to have its own prompts while keeping the core framework unchanged.
Execution Flow
For the resume customization task, the execution flow is as follows:
- Planner retrieves market intelligence for the target position (salary range, skill requirements, etc.).
- Specialist retrieves the candidate's real resume data and generates a customized resume based on the plan and retrieval results.
- Evaluator runs programmatic checks: verifying if company names appear in the resume and if project details have a basis.
- If there are issues, the Fix node corrects based on real data, then returns to the Evaluator.
Throughout the entire process, the LLM always works under the constraints of real documents, reducing the possibility of hallucinations from the source.
Conclusion: Insights from Design Philosophy
The core value of llamaindex-pse lies not in implementing the PSE pattern—langgraph-pse and crewai-pse also implement it. Its unique contribution is using LlamaIndex's native Workflow + RAG capabilities to make "grounding against hallucinations" a first-class citizen of the framework, rather than an optional feature.
When RAG becomes part of the infrastructure, the design focus of multi-agent frameworks should shift from "how to orchestrate" to "how to ground." llamaindex-pse demonstrates a viable path:
- Event-driven Workflow makes control flow more declarative.
- RAG Precedence changes hallucination defense from post-hoc patching to pre-emptive prevention.
- Bypassing Framework Abstraction by directly calling the underlying SDK solves compatibility issues.
- Sandboxed Tools ensure agent operations stay within safe boundaries.
This design philosophy is worth borrowing for any task requiring "generating content based on real documents"—such as resume customization, technical reports, and investment analysis.
Source Code Navigation
| Module | File | Description |
|---|---|---|
| Core Workflow | workflow.py | PSE three-role Workflow implementation, including event definitions, RAG retrieval, and step decorators |
| LLM Client | model.py | SimpleLLM bypassing LlamaIndex enum validation, supporting various OpenAI-compatible models, with Token statistics |
| Sandbox Tools | tools.py | Restricted read_file and run_bash tools, including dangerous command blacklists |
| Configuration Management | config.py | Configuration class loaded from environment variables, supporting various LLM and Embedding backends |
| Prompt Loading | prompts.py | Loads system prompts from task directories |