让智能体从源头不说谎：用 LlamaIndex Workflow 构建 RAG 接地的 PSE 多智能体框架

🇬🇧 English Version

引言：多智能体的"幻觉困境"

你有没有遇到过这样的场景：让多智能体系统根据候选人简历为目标岗位定制一份简历，结果生成的内容里出现了候选人从未在真实简历中提过的公司名、项目细节，甚至是编造的任职时间？

这不是你的错——这是大语言模型的"天性"。当 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 装饰的方法接收一个输入事件，返回下一个事件。节点返回什么事件，流程就走哪条路。

为什么更好？

1. 重试循环天然表达：Evaluator 返回 FixEvent 或 StopEvent，无需手动条件边
2. 状态集中且类型安全：用 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 读写，避免了隐式的全局状态或复杂的状态传递机制。

3. 代码更声明式：控制流即代码结构，一目了然

第二节：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)

这个设计有两个关键点：

1. LLM 评审只在首轮运行——避免后续轮次中 LLM 评审员被之前的修正结果影响
2. 程序化核查每轮都运行——这是真正的"硬核查"，比如检查代码引用是否存在、函数名是否在源码中 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)

这里有几个实用细节：

1. 查询截断：embedding 模型的上下文窗口有限，长查询会被截取前 200 字
2. Reranker 支持：可选的 Cross-Encoder 重排序，提升检索精度
3. 分数标注：每个检索结果附带 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 ""

为什么更好？

1. 兼容性：支持任何 OpenAI 兼容协议的 LLM（如各类第三方模型）
2. 简单直接：不强行适配框架的抽象层，需要时直接调用底层 SDK
3. 内置重试：应对网关偶发断连，最多重试 3 次
4. 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 秒。

为什么更好？

1. 最小权限原则：智能体只能访问项目内的资源
2. 危险命令拦截：黑名单覆盖常见的破坏性操作
3. 透明反馈：被拦截时明确告知原因，而不是静默失败

第六节：实战演示——简历定制任务

场景设定

假设我们要用 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 ""

这种设计让每个任务都可以有自己的提示词，而核心框架保持不变。

运行流程

对于简历定制任务，运行流程如下：

1. Planner 检索目标岗位的市场情报（薪资范围、技能要求等）
2. Specialist 检索候选人的真实简历数据，基于规划和检索结果生成定制简历
3. Evaluator 运行程序化核查：检查公司名称是否在简历中出现、项目细节是否有依据
4. 如果有问题，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	从任务目录加载系统提示词