为什么需要显式的状态机?
多 Agent 框架有两种哲学:隐式流程(Agent 自由编排)和显式状态机(每一步的状态流转可观察、可调试、可控制)。
CrewAI 的 sequential 流程属于前者——Agent 按顺序执行,但你无法在运行时查看中间状态、无法在某个节点插入自定义逻辑、更无法精确控制重试路径。当任务变复杂(比如"生成→核对→修正"的循环),黑盒流程会变成调试噩梦。
LangGraph 的 StateGraph 提供了另一种可能:用代码显式定义图结构——节点、边、条件分支全部可见。langgraph-pse 正是基于这个能力,把 Planner → Specialist → Evaluator → Fix 的 PSE 模式变成了一幅可审查的架构图:
这不是抽象描述,而是 graph.py 中 build_graph() 的真实结构。整个流程是:
- START → planner → specialist → evaluator
- evaluator 判断结果:
- 通过 → END
- 有问题 → fix → 回到 evaluator(循环重试)
你可以用 LangGraph Studio 可视化这个图,看到每一步的状态快照。
核心创新:合并闸门(Merge Gate)
langgraph-pse 最独特的设计不在图结构,而在 Evaluator 节点。
传统 PSE 框架的 Evaluator 只是一个 LLM 评审角色——让 LLM 自己判断自己产出的质量。这有两个致命问题:
- LLM 自判不可靠:LLM 倾向于给自己的产出打高分("自我偏好偏差")
- 主观问题导致死循环: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-pse 用 TypedDict 显式定义每一个状态字段:
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 的产出 - 控制流层:
attempts、fictitious、verified、eval_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/ |
| 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
发表回复