引言：多智能体的"幻觉困境"
你有没有遇到过这样的场景：让多智能体系统根据候选人简历为目标岗位定制一份简历，结果生成的内容里出现了候选人从未在真实简历中提过的公司名、项目细节，甚至是编造的任职时间？
这不是你的错——这是大语言模型的"天性"。当 LLM 被要求"创造"内容时，它倾向于用训练数据中的模式来填补信息空白，哪怕这些信息与事实完全不符。
大多数多智能体框架采用的是反应式方案：先生成内容，再检测幻觉，最后尝试修复。但问题是：幻觉已经在产物中产生了，修复往往只是打补丁。
llamaindex-pse 采取了不同的思路：把 RAG 检索放在生成之前。让 Planner 和 Specialist 在产出任何内容之前，先从知识库检索真实文档作为上下文。这样，智能体从源头就是"接地"的——它不需要编造，因为真实数据就在手边。
第一节：为什么选择 LlamaIndex Workflow 而非 StateGraph？
直觉做法：状态图 + 条件边
如果你熟悉多智能体编排框架，可能会想到 LangGraph 的 StateGraph。它用显式的条件边来控制流程分支：
START → node_a → conditional_edge ├─ 是 → node_b → ... └─ 否 → node_c → ...
每次流程分支都需要手动编写判断逻辑——Evaluator 通过后走哪条边、有问题时走哪条边，代码里到处是条件判断。
实际做法：事件驱动 + @step 装饰器
llamaindex-pse 使用了 LlamaIndex 的 Workflow，通过类型化的事件和 @step 装饰器来定义节点：
class PlannerEvent(Event):
"""触发 Planner 步骤。"""
task_input: str = ""
class SpecialistEvent(Event):
"""触发 Specialist 步骤。"""
task_input: str = ""
plan: str = ""
class EvaluatorEvent(Event):
"""触发 Evaluator 步骤。"""
artifact: str = ""
attempts: int = 0
class FixEvent(Event):
"""触发 Fix 步骤。"""
artifact: str = ""
issues: list = field(default_factory=list)
每个 @step 装饰的方法接收一个输入事件，返回下一个事件。节点返回什么事件，流程就走哪条路。
为什么更好？
- 重试循环天然表达：Evaluator 返回
FixEvent或StopEvent，无需手动条件边 - 状态集中且类型安全：用 dataclass 封装所有状态
@dataclass
class PSEState:
"""PSE Workflow 状态（dataclass，通过 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 检索上下文（planner / specialist 自动填充）
rag_context: str = ""
所有节点共享同一个 state 对象，通过 ctx.store 读写，避免了隐式的全局状态或复杂的状态传递机制。
- 代码更声明式：控制流即代码结构，一目了然
第二节：PSE 三角色的分工与协作
Planner：情报收集者
Planner 的职责是理解任务，产出执行规划。它的独特之处在于可选的 RAG 增强：
@step
async def planner(self, ctx: Context, ev: PlannerEvent) -> SpecialistEvent:
"""Planner：RAG 检索市场/JD 情报 + 读取上下文 → 产出执行规划。"""
# RAG 增强：优先用 planner_retriever（市场/JD 情报），fallback 到通用 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 = "市场情报" if self._planner_retriever else "参考文档"
rerank_tag = " (reranked)" if self._reranker else ""
print(f" 📚 RAG 检索{label}{rerank_tag} {len(rag_ctx)} 字")
full_input = ev.task_input
if rag_ctx:
full_input += f"\n\n## 检索到的参考文档\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"✅ 规划已完成 ({len(plan)} 字)")
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)
注意这里的关键设计：Planner 不是直接开始规划，而是先检索相关文档。对于简历定制任务，它会检索目标岗位的市场情报；对于技术文章生成，它会检索项目源代码的元数据。
Specialist：执行者
Specialist 负责把规划展开为最终产物。它同样享有 RAG 增强：
@step
async def specialist(self, ctx: Context, ev: SpecialistEvent) -> EvaluatorEvent:
"""Specialist：RAG 检索简历源数据 + 把规划展开为最终产物。"""
full = (ev.task_input + "\n\n## 执行规划\n" + ev.plan) if ev.plan else ev.task_input
# RAG 增强：specialist 用 retriever（简历源数据），如 planner 已检索则复用
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 检索简历源数据{rerank_tag} {len(rag_ctx)} 字")
state.rag_context = rag_ctx
await ctx.store.set("state", state)
if rag_ctx:
full += f"\n\n## 检索到的参考文档（产物必须基于此，禁止编造）\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 未输出任何内容")
state.artifact = artifact
await ctx.store.set("state", state)
return EvaluatorEvent(artifact=artifact, attempts=state.attempts)
这里有一个重要的设计细节：如果 Planner 已经检索了上下文，Specialist 会复用。这避免了重复检索，同时也保证了两个节点看到的信息是一致的。
Evaluator：合并闸门
Evaluator 是 PSE 模式中最重要的创新点——它是一个合并闸门，结合了两层检查：
@step
async def evaluator(self, ctx: Context, ev: EvaluatorEvent) -> FixEvent | StopEvent:
"""Evaluator（合并闸门）：LLM 评审(仅首轮) + 程序化 verify_fn 硬核查(每轮)。"""
state: PSEState = await ctx.store.get("state")
# 1) LLM 评审（仅首轮）
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 交叉验证：用检索到的文档补充评审依据
rag_section = ""
if state.rag_context:
rag_section = f"\n\n## RAG 检索到的参考文档（评审依据）\n{state.rag_context}"
full = (
f"## 待评估的产物\n{ev.artifact}\n\n"
f"## 真实数据（供核对，禁止以产物之外的内容为依据）\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" 🔍 评审员发现问题 {len(eval_issues)} 项")
# 2) 程序化核查（每轮）
if self._verify_fn is not None:
# 构造类 dict state 给 verify_fn（兼容 langgraph-pse 签名）
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" 核查 (第{state.attempts}次) — 程序化通过 {len(ok)} 项, "
f"程序化问题 {len(prog_bad)} 项, 评审问题 {len(eval_issues)} 项")
for b in all_bad:
print(f" ❌ {b}")
# 决定：通过 → Stop，否则 → Fix
if not all_bad or state.attempts > state.max_retries:
if not all_bad:
print(" ✅ 核查通过")
else:
print(" ⚠️ 达到最大重试次数，停止修正")
return StopEvent(result={"artifact": state.artifact, "state": state})
return FixEvent(artifact=state.artifact, issues=all_bad)
这个设计有两个关键点：
- LLM 评审只在首轮运行——避免后续轮次中 LLM 评审员被之前的修正结果影响
- 程序化核查每轮都运行——这是真正的"硬核查"，比如检查代码引用是否存在、函数名是否在源码中 grep 到
Fix：修正者
当 Evaluator 发现问题时，流程进入 Fix 节点：
@step
async def fix(self, ctx: Context, ev: FixEvent) -> EvaluatorEvent:
"""Fix：按核查出的问题修正产物。RAG 上下文注入修正提示，防止凭空编造。"""
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 上下文：修正时也基于检索到的真实文档
rag_section = ""
if state.rag_context:
rag_section = (
"\n\n**RAG 参考文档（修正时必须以此为准）**:\n"
f"{state.rag_context}\n"
)
print(" 🔄 自动修正中...")
prompt = (
"以下产物被程序化核查发现问题，请修正。\n\n"
f"**问题清单（必须修复）**:\n" + "\n".join(f"- {i}" for i in ev.issues) + "\n\n"
"**真实数据（修正时必须以此为准，把错误数字改为真实值，"
"不得编造也不得删除数字）**:\n"
f"{scan_str}\n"
f"{rag_section}\n"
"**规则**:\n"
"1. 仅修正问题清单中指出的错误，将错误数字改为真实数据中的正确值\n"
"2. 不要删除任何正确的数字或内容，保持其余部分不变\n"
"3. **绝对禁止删除整段工作经历或项目经历**——如果某段经历的数据有误，修正数据而非删除整段\n"
"4. **绝对禁止将经历替换为占位符或注释**（如'注：源文档未提供'）——应基于真实数据修正\n"
"5. 如果问题清单提到'20年'等年限表述：**直接删除**该年限表述，不要替换为其他年份数字\n"
"6. 如果问题清单提到项目缺少起止时间：在项目标题的公司名后追加`| YYYY.MM-YYYY.MM`格式的时间范围，"
"时间从该公司的任职期间推断\n"
"7. 输出修正后的完整产物，不输出解释\n\n"
f"## 当前产物\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)
注意 Fix 节点也注入了 RAG 上下文。这意味着即使是修正阶段，LLM 也是在真实文档的约束下工作，而不是"自由发挥"。
第三节：RAG 接地——防线左移的核心设计
对比：反应式 vs 主动式
传统的多智能体框架的流程是：
generate → detect(hallucination) → fix → detect → fix → ...
这是一个反应式循环——幻觉已经产生，然后去检测和修复。
llamaindex-pse 的流程是：
retrieve → generate(grounded) → verify(residual)
这是一个主动式策略——先生成前就提供真实文档，让 LLM 基于事实生成。verify_fn 仍然运行，但它的角色从"主要防线"降级为"安全网"。
源码实现：智能检索
RAG 检索的实现也很精巧：
async def _retrieve_context(retriever, query: str, top_k: int = 5, reranker=None) -> str:
"""用 LlamaIndex retriever 检索相关文档，拼接为上下文字符串。
query 过长时会超出 embedding 模型上下文长度，因此截取前 200 字作为检索关键词。
如果提供 reranker，检索结果会先经过 Cross-Encoder 重排序。
"""
# 截取短查询：embedding 模型上下文有限（如 Ollama snowflake-arctic-embed2 仅 8K tokens）
short_query = query[:200] if len(query) > 200 else query
nodes = retriever.retrieve(short_query)
if not nodes:
return ""
# Rerank：Cross-Encoder 重排序，提升检索精度
if reranker is not None:
try:
nodes = reranker.postprocess_nodes(nodes, query_str=short_query)
except Exception:
pass # rerank 失败时 fallback 到原始排序
# 取 top_k 个节点，按 score 降序
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)
这里有几个实用细节：
- 查询截断：embedding 模型的上下文窗口有限，长查询会被截取前 200 字
- Reranker 支持：可选的 Cross-Encoder 重排序，提升检索精度
- 分数标注：每个检索结果附带 score，方便调试
为什么有效？
幻觉产生的根源是 LLM 在"空想"。给它真实文档作为上下文，它就不需要编造。
在 Planner 阶段，检索市场情报或 JD 信息；在 Specialist 阶段，检索简历源数据或项目元数据。两个节点都能看到相同的上下文，保证了信息的一致性。
第四节：绕过 LlamaIndex 枚举校验的 LLM 客户端设计
问题：框架抽象层的限制
LlamaIndex 的 OpenAI 封装有一个隐藏的限制：它会调用 openai_modelname_to_contextsize() 来校验模型名称。对于非 OpenAI 官方的第三方模型（如各类兼容模型），这会报错：
ValueError: Unknown model type: example-model
直觉做法：修改 LlamaIndex 源码或等待上游修复
但这意味着你要么侵入框架源码，要么等待上游支持——都不是好选择。
实际做法：直接使用底层 SDK
llamaindex-pse 的解决方案是绕过 LlamaIndex 的抽象层，直接使用 OpenAI SDK：
class SimpleLLM:
"""基于 openai SDK 的轻量 LLM（绕开 LlamaIndex OpenAI 校验）。
仅提供 PSE workflow 需要的 chat() 和 complete() 接口。
内置连接重试（最多 3 次，间隔 5 秒），应对网关偶发断连。
自动记录 token 消耗到全局 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 调用，应对网关偶发断连。"""
for attempt in range(max_retries):
try:
return fn()
except openai.APIConnectionError as e:
if attempt < max_retries - 1:
print(f" ⚠️ 连接失败 ({attempt+1}/{max_retries})，{delay}s 后重试...")
time.sleep(delay)
else:
raise
def chat(self, messages: list[dict], stage: str = "chat") -> str:
"""Chat completion，返回 assistant 内容。"""
resp = self._call_with_retry(lambda: self._client.chat.completions.create(
model=self._model,
messages=messages,
timeout=180,
))
# 记录 token 消耗
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 ""
为什么更好？
- 兼容性：支持任何 OpenAI 兼容协议的 LLM（如各类第三方模型）
- 简单直接：不强行适配框架的抽象层，需要时直接调用底层 SDK
- 内置重试：应对网关偶发断连，最多重试 3 次
- Token 统计：自动记录每次调用的 token 消耗和费用
class TokenStats:
"""Token 消耗统计收集器，含价格计算。"""
# 模型定价表：元 / 百万 tokens（人民币）
PRICING = {
"deepseek-chat": {"input": 1.0, "output": 2.0}, # DeepSeek V3 官方价
"deepseek-reasoner": {"input": 4.0, "output": 16.0}, # DeepSeek R1
"agnes-2.0-flash": {"input": 0.5, "output": 1.5}, # Agnes 网关（估算）
}
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
# 计算费用
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
第五节：沙箱工具与安全性设计
问题：多智能体的安全风险
多智能体系统通常需要访问文件系统或执行命令，但这带来了安全风险：
- 智能体可能被诱导执行
rm -rf等破坏性命令 - 智能体可能越权访问项目外的敏感文件
直觉做法：完全禁用工具或让用户自己处理
但这会限制智能体的能力，或者把安全责任推给用户。
实际做法：受限沙箱
llamaindex-pse 提供了两个沙箱化工具：
def read_file(path: str) -> str:
"""读取文件内容。参数 path 为文件路径（限定在项目目录内）。"""
p = Path(path).resolve()
if not str(p).startswith(str(_PROJECT_ROOT)):
return f"[错误] 路径超出项目范围: {path}"
if not p.exists():
return f"[错误] 文件不存在: {path}"
if not p.is_file():
return f"[错误] 不是文件: {path}"
return p.read_text(encoding="utf-8")
read_file 确保智能体只能读取项目根目录内的文件。
# 危险命令片段黑名单（命中即拒绝，降低被诱导执行破坏命令的风险）
_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:
"""执行 bash 命令并返回输出（受限沙箱：禁止破坏性命令，工作目录限定在项目根内）。"""
for pat in _DANGEROUS_PATTERNS:
if re.search(pat, command):
return (
f"[拒绝] 命令命中危险模式（{pat}），已被沙箱拦截。"
"如需执行破坏性操作请人工进行。"
)
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"[错误] {e}"
run_bash 有危险命令黑名单和工作目录限制，超时设为 30 秒。
为什么更好？
- 最小权限原则：智能体只能访问项目内的资源
- 危险命令拦截：黑名单覆盖常见的破坏性操作
- 透明反馈：被拦截时明确告知原因，而不是静默失败
第六节：实战演示——简历定制任务
场景设定
假设我们要用 llamaindex-pse 实现一个简历定制任务：从真实简历源数据出发，为目标岗位定制一份简历。
配置加载
所有配置从环境变量加载：
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 网关（与 DeepSeek 同 OpenAI 兼容协议，独立 key / base_url / 模型）
AGNES_KEY: str = os.getenv("AGNES_KEY", "")
AGNES_BASE_URL: str = os.getenv("AGNES_BASE_URL", "")
AGNES_MODEL: str = os.getenv("AGNES_MODEL", "")
# Embedding 模型（RAG 索引用）
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", "")
提示词加载
提示词从任务目录动态加载：
def load_prompt(name: str, task: str | None = None) -> str:
"""加载指定角色的系统提示词。
优先读 tasks/<task>/prompts/<name>.md，找不到返回空串。
"""
if task:
prompt_path = PROMPTS_DIR / task / "prompts" / f"{name}.md"
if prompt_path.exists():
return prompt_path.read_text(encoding="utf-8")
return ""
这种设计让每个任务都可以有自己的提示词，而核心框架保持不变。
运行流程
对于简历定制任务，运行流程如下：
- Planner 检索目标岗位的市场情报（薪资范围、技能要求等）
- Specialist 检索候选人的真实简历数据，基于规划和检索结果生成定制简历
- Evaluator 运行程序化核查：检查公司名称是否在简历中出现、项目细节是否有依据
- 如果有问题，Fix 节点基于真实数据修正，然后回到 Evaluator
整个过程，LLM 始终在真实文档的约束下工作，从源头上减少了幻觉的可能性。
收尾：设计哲学的启示
llamaindex-pse 的核心价值不在于实现了 PSE 模式——langgraph-pse 和 crewai-pse 也实现了。它的独特贡献在于用 LlamaIndex 原生的 Workflow + RAG 能力，把"接地防幻觉"从可选特性变成了框架的一等公民。
当 RAG 成为基础设施的一部分时，多智能体框架的设计重心应该从"如何编排"转向"如何接地"。llamaindex-pse 证明了一种可行的路径：
- 事件驱动的 Workflow 让控制流更声明式
- RAG 前置 让幻觉防护从事后修补变为事前预防
- 绕过框架抽象 直接调用底层 SDK 解决了兼容性问题
- 沙箱化工具 确保了智能体的操作在安全边界内
对于任何需要"基于真实文档生成内容"的任务——简历定制、技术报告、投资分析——这种设计哲学都值得借鉴。
源码导航
| 模块 | 文件 | 说明 |
|---|---|---|
| 核心工作流 | workflow.py | PSE 三角色的 Workflow 实现，包含事件定义、RAG 检索、步骤装饰器 |
| LLM 客户端 | model.py | 绕过 LlamaIndex 枚举校验的 SimpleLLM，支持各类 OpenAI 兼容模型，含 Token 统计 |
| 沙箱工具 | tools.py | 受限的 read_file 和 run_bash 工具，含危险命令黑名单 |
| 配置管理 | config.py | 从环境变量加载的配置类，支持多种 LLM 和 Embedding 后端 |
| 提示词加载 | prompts.py | 从任务目录加载系统提示词 |
发表回复