LangGraph PSE: 用状态机显式建模 PSE 三角色协作

作者:

🇬🇧 English Version

为什么需要显式的状态机?

多 Agent 框架有两种哲学:隐式流程(Agent 自由编排)和显式状态机(每一步的状态流转可观察、可调试、可控制)。

CrewAI 的 sequential 流程属于前者——Agent 按顺序执行,但你无法在运行时查看中间状态、无法在某个节点插入自定义逻辑、更无法精确控制重试路径。当任务变复杂(比如"生成→核对→修正"的循环),黑盒流程会变成调试噩梦。

LangGraph 的 StateGraph 提供了另一种可能:用代码显式定义图结构——节点、边、条件分支全部可见。langgraph-pse 正是基于这个能力,把 Planner → Specialist → Evaluator → Fix 的 PSE 模式变成了一幅可审查的架构图

这不是抽象描述,而是 graph.pybuild_graph() 的真实结构。整个流程是:

  1. START → planner → specialist → evaluator
  2. evaluator 判断结果:
    • 通过END
    • 有问题 → fix → 回到 evaluator(循环重试)

你可以用 LangGraph Studio 可视化这个图,看到每一步的状态快照。

核心创新:合并闸门(Merge Gate)

langgraph-pse 最独特的设计不在图结构,而在 Evaluator 节点

传统 PSE 框架的 Evaluator 只是一个 LLM 评审角色——让 LLM 自己判断自己产出的质量。这有两个致命问题:

  1. LLM 自判不可靠:LLM 倾向于给自己的产出打高分("自我偏好偏差")
  2. 主观问题导致死循环:LLM 每轮都能挑出新问题(措辞、风格、偏好),Fix 永远无法收敛

langgraph-pse 的解决方案是合并闸门——两个核查通道并行运行,结果合并:

def _make_evaluator_node(client, prompt: str, verify_fn: Optional[Callable]):
    """合并闸门:LLM 评审(仅首轮) + 程序化 verify_fn 硬核查(每轮)。

    - 首轮(attempts==0)调用 LLM 对照真实数据审查产物,输出问题清单;
      仅首轮跑 LLM,避免主观问题导致 fix 死循环。
    - 程序化 verify_fn 每轮都跑(确定性,比 LLM 自判可靠,防编造)。
    - 二者合并为 fictitious,驱动 fix 重试。
    """
    def evaluator(state: PSEState) -> dict:
        # 1) LLM 评审(仅首轮)
        eval_issues = []
        if attempts == 0 and prompt:
            # ... 调用 LLM 对照真实数据审查 ...

        # 2) 程序化核查(每轮)
        if verify_fn is not None:
            prog_bad, ok = verify_fn(state)

        all_bad = list(prog_bad) + list(eval_issues)  # 合并

关键决策:

LLM 评审 程序化 verify_fn
运行频率 仅首轮 每轮
能力 宏观判断(逻辑一致性、表述质量) 微观硬核查(数字对不对、引用存不存在)
风险 主观问题导致死循环 无(确定性输出)
收敛保障 有——只要 Fix 改对了,verify_fn 一定通过

仅首轮跑 LLM 是关键的收敛设计:首轮让 LLM 做宏观审查,后续轮次只由程序化核查驱动。这确保了 Fix 循环必然收敛——只要修正了所有硬核查问题,流程就会终止。

verify_fn:框架的灵魂扩展点

build_graph() 接受一个 verify_fn 参数,签名是 (state) -> (bad: list, ok: list)。这是框架区别于 crewai-pse(专注文章生成)的关键——langgraph-pse 从设计之初就是通用质量闭环框架,verify_fn 是任务适配的灵魂。

CRM-QA 任务展示了真实的 verify_fn 实现——核对报告中引用的数字是否与扫描结果一致

def _verify_report(report: str, scan_json: dict) -> tuple[list, list]:
    """报告里引用的数字必须能在扫描结果里找到。

    设计原则:宁可漏报也不误报——只要报告在合理范围内
    出现了真实值即视为通过,避免 LLM 的自然语言表述
    (如「约 6%」「占 39%」)被误判为幻觉而让 fix 死循环。
    """
    for f in findings:
        check = f["check"]
        target = f["count"]
        # 在 Markdown 表格行中定位 check,提取数字,比对
        if val == target:
            ok.append(f"{check} = {target}")
        else:
            bad.append(f"{check} 报告写 {val} 但真实为 {target}")

这个 verify_fn 的"宁可漏报也不误报"原则很重要——如果 LLM 写"约 6%"而真实值是 5.8%,不应该判定为错误。这种对自然语言表述的宽容度,是 LLM 时代程序化核查的设计智慧。

状态透明:PSEState 的数据流

PSE 框架的核心挑战是状态在三个角色间流转时的可观测性langgraph-pseTypedDict 显式定义每一个状态字段:

class PSEState(TypedDict, total=False):
    task_input: str          # 输入:任务上下文
    task_data: dict          # 输入:额外数据(如扫描结果),供 verify_fn 使用
    plan: str                # Planner 输出的执行规划
    artifact: str            # Specialist 输出的产物(报告 / 修正数据等)
    attempts: int            # 控制流:当前重试次数
    fictitious: list         # 控制流:问题列表(程序化 + 评审合并)
    verified: list           # 控制流:通过项
    eval_issues: list        # 控制流:LLM 评审问题(仅首轮)
    max_retries: int         # 控制流:最大重试次数

三个层次一目了然:

  • 输入层task_input + task_data——任务进入框架的数据
  • 产物层plan + artifact——Planner 的规划、Specialist 的产出
  • 控制流层attemptsfictitiousverifiedeval_issues——驱动重试和终止

对比隐式流程框架,你不需要翻日志去猜测"LLM 到底输出了什么",每个字段在任何时刻的值都是可查询的。

三重沙箱:工具链的安全纵深

langgraph-pse 的工具链不是简单的"给 Agent 一个 bash",而是三层纵深防御

第一层:路径沙箱(read_file)

@tool("read_file")
def read_file(path: str) -> str:
    p = Path(path).resolve()
    if not str(p).startswith(str(_PROJECT_ROOT)):
        return f"[错误] 路径超出项目范围: {path}"

Agent 只能读项目目录内的文件,任何 ../../etc/passwd 的尝试都会被拦截。

第二层:命令黑名单(run_bash)

_DANGEROUS_PATTERNS = [
    r"\brm\s+-rf\b", r"\bmkfs\b", r"\bdd\b\s+if=",
    r":\(\)\s*\{",                   # fork bomb
    r"curl\b[^\n]*\|\s*(sh|bash)",  # pipe to shell
    r"\bnc\b[^\n]*-e\b",            # reverse shell
    # ... 共 14 条模式
]

覆盖了 rm -rf、fork bomb、curl | sh、reverse shell 等常见攻击模式。

第三层:SQL 只读沙箱(query_crm)

@tool("query_crm")
def query_crm(sql: str) -> str:
    if not sql.strip().lower().startswith("select"):
        return "[拒绝] 仅允许 SELECT 查询。"
    if ";" in sql.strip().rstrip(";"):
        return "[拒绝] 仅允许单条 SELECT,不支持多语句。"
    con = sqlite3.connect(f"file:{db}?mode=ro&immutable=1", uri=True)

三道防线:非 SELECT 拒绝、多语句拒绝、SQLite immutable 模式(即使绕过前两道也无法写入)。

实战:CRM 数据质量看门狗

langgraph-pse 当前跑在 personal-crm 项目里,做数据质量看门狗。这个场景和"写文章"完全不同——产物是结构化的 QA 报告,核查的是数字准确性

运行方式:

# 确定性扫描(零成本,无需 API key)
python tasks/crm-qa/run.py --db /path/to/crm.db

# 用 LLM 生成自然语言 QA 报告 + 数字核对
python tasks/crm-qa/run.py --db /path/to/crm.db --llm

两种模式的设计哲学:

  • 默认模式:纯程序化扫描,输出 JSON findings,零 LLM 成本。适合 CI/CD 定时检查。
  • –llm 模式:在扫描结果基础上,让 Specialist 写自然语言报告,verify_fn 核对报告中引用的数字与扫描结果是否一致。适合人工审阅。

这个"确定性扫描 + LLM 报告 + 程序化数字核对"的组合,体现了 langgraph-pse 的设计核心:LLM 负责表达,程序负责验证

build_graph():可组合的图构建

build_graph() 是框架的入口,通过参数组合而非继承来定制行为:

graph = build_graph(
    task="crm-qa",              # 加载 tasks/crm-qa/prompts/*.md
    verify_fn=_verify_state,    # 注入数字核查
    use_planner=True,           # 是否需要规划阶段
    max_retries=3,              # 最大修正轮次
    provider="deepseek",        # LLM 后端
)

关键设计决策:

  • use_planner=False:简单任务可以跳过 Planner,直接让 Specialist 执行。避免给"查一个数字"这类任务加上不必要的规划开销。
  • verify_fn=None:如果不需要程序化核查,Evaluator 退化为纯 LLM 评审(首轮)。
  • tools=None:默认注入 read_file + run_bash,可以自定义工具列表。

这种"参数即配置"的风格比继承体系更轻量——你不需要写一个 CRMQAGraph(StateGraph) 子类,传几个参数就行。

源码导航

模块 文件 说明
状态机核心 graph.py PSEState、节点构建、合并闸门、图编译
三重沙箱工具 tools.py read_file + run_bash + query_crm + crm_qa_scan
配置管理 config.py 多 Provider 环境变量
模型创建 model.py ChatOpenAI + 指数退避重试
提示词加载 prompts.py 动态加载 tasks//prompts/*.md
CRM-QA 入口 run.py 扫描 + LLM 报告 + verify_fn 数字核对
QA 扫描器 qa_scan.py 确定性数据质量扫描

快速开始

# 安装依赖
uv sync

# 配置环境变量(参考 .env.example)
cp .env.example .env
# 编辑 .env,填入 OPENAI_API_KEY / OPENAI_BASE_URL / OPENAI_MODEL
# 或 AGNES_KEY / AGNES_BASE_URL / AGNES_MODEL

# 运行确定性扫描(零成本)
python tasks/crm-qa/run.py --db /path/to/crm.db

# 用 LLM 生成 QA 报告 + 数字核对
python tasks/crm-qa/run.py --db /path/to/crm.db --llm

评论

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注

首页 简历 关于 隐私政策

© 2026 Erishen
沪ICP备2024079226号-1   沪公网安备31010502007082号