AI智能体的结构化约束：ResolveFlow流程内部机制
阐述了基于LangGraph的智能体如何通过代码级检查而非提示指令来实现推理与执行的分离，同时介绍了在此过程中出现的一个检索错误。
大多数代理型人工智能演示都遵循相同的基本模式：模型决定一个动作后立即执行它。某个工具会被附加到提示语中，模型调用该工具，而工具会在无需进一步检查的情况下运行。在简短的演示视频中这可能看起来很可信，但恰恰是这种设计让那些担心让自主系统处理重要事务的人感到担忧——因为通常在做出正确诊断与对实时系统造成有害影响之间，唯一的保障往往只是提示语中的一句警告。
此处描述的项目正是为避免依赖单一的警示指令而设计的。
ResolveFlow可以接收GitHub问题链接，收集相关证据，将问题分类，然后根据类别执行三种操作之一：采取固定的、不可更改的解决方案，基于收集到的证据启动由大语言模型驱动的调查，或将问题直接转交给人工审核员。它并非采用模型触发工具调用的单一循环结构，而是以一个核心理念为基础构建的LangGraph状态机：
推理与执行的分离源于系统的设计方式，而非遵循某种约定。
这意味着并非简单地指令模型先向他人咨询即可。相反，在人类看到结果之前，会先进行第二次独立的LLM调用，对第一个模型的诊断结果进行审查与评估。而在代码库中唯一被允许向GitHub写回数据的函数，会通过其自身的逻辑来验证是否存在明确的批准标志——这并非因为系统预期会以那种方式路由调用，而是因为即便未来有代码修改建立了跳过常规步骤的直接路径，该函数在没有该标志的情况下也会拒绝执行。
本指南的其余部分将介绍如何利用实际实现来构建该系统，以及开发过程中出现的一个错误。这个错误带来了一个重要的教训：指向真实源文档的模型与指向确实与当前问题相关的源文件的模型并非同一回事。
处理流程的结构
共有六个阶段按顺序执行，其中只有一个阶段被允许修改处理流程之外的任何内容：
- fetch_evidence — 实际调用 GitHub REST API，获取问题的正文、评论线程以及所有 CI 检测结果。
- normalize_evidence — 对这些调用返回的未处理 JSON 进行检查并转换为带类型的 IssueEvidence 对象。
- classify — 一个轻量级的规则驱动步骤，不涉及任何 LLM 调用。
以下章节将详细阐述其中最重要的部分。
分类：刻意不使用LLM调用
由于大型语言模型随时可用，人们很容易倾向于让所有决策都通过它来做出，即便那些并不需要此类推理的决策也不例外。分类步骤决定了某个问题在多大程度上可以被允许走那条成本高昂、风险较高的路径——即基于模型的诊断、数据检索以及最终的写入操作。正因为具有这种筛选作用，该步骤本身就需要成本低廉、处理速度快，并且结果完全可预测：
def classify(state: GraphState) -> dict:
if state["evidence"].has_failing_ci:
return {"classification": "deterministic"}
elif state["evidence"].is_information_sparse:
return {"classification": "ai_investigation"}
else:
return {"classification": "human_review"}
该步骤会产生三种可能的结果，每种结果都需要不同程度的后续信任：
- 确定性结果——测试失败的信号本身就是明确且机械性的，无需进一步调查；该问题只需被标记后转交给维护人员即可。
值得注意的是，备用方案是human_review而非ai_investigation。每当系统无法判断情况时，它不会试图即兴给出一个聪明的答案。
诊断方式：先检索，再依据结构化方案处理，而非自由文本
一旦问题被路由到ai_investigation分支，generate_diagnosis步骤就会搜索一个Pinecone索引，该索引包含约2,750个片段，这些片段来自四个不同仓库中的真实已解决问题（facebook/react、langchain-ai/langchain、microsoft/terminal以及vercel/next.js）。随后，这些检索到的片段会被发送给大语言模型，作为调用时的辅助上下文：
def generate_diagnosis(state: GraphState) -> dict:
evidence = state["evidence"]
query = f"{evidence.title}\n\n{evidence.body}"
snippets = retrieve_evidence(query, k=3)
snippet_block = "\n\n".join(
f"[{s['id']}] (relevance: {s['score']:.2f}) {s['text']}" for s in snippets
)
prompt = (
f"Issue: {evidence.title}\n{evidence.body}\n\n"
f"Comments:\n{chr(10).join(evidence.comments) or '(none)'}\n\n"
f"Evidence snippets (cite by id in square brackets):\n{snippet_block}"
)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_llm = llm.with_structured_output(Diagnosis)
diagnosis = structured_llm.invoke([("system", _SYSTEM_PROMPT), ("human", prompt)])
return {
"diagnosis": diagnosis,
"retrieved_ids": [s["id"] for s in snippets],
"retrieved_scores": {s["id"]: s["score"] for s in snippets},
}
这个诊断步骤中有几个细节值得仔细研究。
首先，输出的结构并非事后提取的——而是在一开始就确定的。诊断功能被定义为一个Pydantic模型：
class Diagnosis(BaseModel):
root_cause: str
severity: Literal["low", "medium", "high"]
missing_info: list[str] = Field(default_factory=list)
recommended_next_steps: list[str]
citations: list[str] = Field(
default_factory=list,
description="IDs of retrieved evidence/doc snippets that support each claim above",
)
通过调用 .with_structured_output(Diagnosis)，模型在生成答案时会被强制遵循特定的结构。之后不会使用正则表达式从一段文字中提取根本原因——一旦调用成功，返回的已经是结构化的对象，而非需要人工解读的文本。
其次，引用并非关乎语气或自信程度，而必须是真实的标识符。系统提示明确要求所有引用都必须与所提供的代码片段ID相匹配；不得编造内容，也不得将与代码片段实际不支持的论点关联起来。这一约束本身似乎就已足够，但实际上并非如此——这正是需要后续处理步骤的原因。
独立审核：模型负责解释，代码负责决策
这可以说是整个系统中最重要的架构设计选择。
independent_review节点会发起第二个完全独立的ChatOpenAI调用，该调用拥有自己的提示词，且与用于生成诊断结果的调用没有共享上下文。它的任务就是对那个诊断结果进行评估。
不过关键在于：是否批准或升级的决定从来不由模型来做出。实际上是由普通Python代码计算出的三个布尔值来决定的，而大语言模型的输出则被简化为人类可读的评论，这些评论实际上并不依赖于任何批准逻辑。
def independent_review(state: GraphState) -> dict:
evidence = state["evidence"]
diagnosis = state["diagnosis"]
retrieved_ids = set(state.get("retrieved_ids", []))
retrieved_scores = state.get("retrieved_scores", {})
groundedness_ok = bool(diagnosis.citations) and all(
citation_id in retrieved_ids
and retrieved_scores.get(citation_id, 0.0) >= MIN_RELEVANCE_SCORE
for citation_id in diagnosis.citations
)
risk_ok = diagnosis.severity in _ALLOWED_SEVERITIES # {"low", "medium"}
permission_ok = True # comment/label are the only writes available today
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
reasoning = llm.invoke(
[("system", _SYSTEM_PROMPT), ("human", prompt)]
).content # human-readable critique — not what the gate checks
outcome = "approve" if (groundedness_ok and risk_ok and permission_ok) else "escalate_to_human"
return {"review_result": ReviewResult(
outcome=outcome,
groundedness_ok=groundedness_ok,
risk_ok=risk_ok,
permission_ok=permission_ok,
reasoning=reasoning,
)}
任何可靠的“大语言模型作为裁判”系统都需要遵循这一通用模式：模型可以解释自己的决策，但最终的决定是由代码来做出的。
如果你让一个模型来评估另一个模型的工作成果，然后仅仅信任它给出的评价结果，那么实际上你就构建了一个可靠性受制于你试图验证的那一因素的系统。
在这种设计中，大语言模型的输出虽可作为给人类读者的有用叙述，但真正起决定作用的审核机制不会因为具有说服力的措辞而被诱导出错误结果，因为它并不会解析模型所写的任何内容来做出判断。
还有一个值得注意的细节：高严重程度从未出现在_ALLOWED_SEVERITIES中。任何被标记为高严重程度的诊断结果都会自动转交给人类处理，无论其依据的多充分。正确与安全到可以自动批准并非同一回事。
缺陷：“有依据”并不等同于“相关”
这正是实际操作中变得真正棘手的地方。
最初的groundedness_ok检查仅用于验证引文ID是否与检索结果中的某个条目匹配——即真实的ID，而非伪造的ID。从理论上看，这似乎是个合理的检查方式。但实际上并不够用。
在针对40个问题的小型语料库进行测试时，一个真正为空内容的React问题被送入了处理流程（facebook/react#36932，“experimental_taintUniqueValue在处理大型二进制值时会抛出RangeError错误”）。检索结果返回了三个片段，它们都是有效的且都被正确识别——但其中没有任何一个与这个特定漏洞有关。即便基于如此有限的信息，模型仍然得出了一个看似有根据且具体的错误诊断：它声称存在“React DevTools扩展兼容性问题”。所有的引用都顺利通过了真实性测试。但这个诊断本身依然毫无用处。
解决办法是不再将“被检索到”视为“相关”的代名词。代码文件tools/retrieval.py已被更新，现在每个片段在返回文本的同时还会附带其余弦相似度得分：
def retrieve_evidence(query: str, k: int = 3) -> list[dict]:
results = vector_store.similarity_search_with_score(query, k)
return [
{"id": doc.metadata["id"], "text": doc.page_content,
"source": doc.metadata["source"], "score": score}
for doc, score in results
]
我随后查看了实际相似度分数在真实索引中的表现，而非凭猜测设定阈值。那些与语料库中真实内容高度匹配的查询，其相似度分数在0.53到0.63之间；而针对四个数据源中完全不存在的内容的查询，相似度分数则在0.21到0.22之间。基于这一差异，independent_review.py现在设定了一个严格的最低标准：所有被引用的片段都必须满足MIN_RELEVANCE_SCORE = 0.35这一要求。该阈值被刻意设定在差距的较高一侧，而非中间位置，因为让一个正确的诊断结果被误判为需要升级的情况，其代价远小于让一个错误的诊断结果被当作正确通过。
在修正评分逻辑的同时，检索语料库本身也需要扩大规模。最初它仅包含来自一个存储库的40篇文档，生成了大约130个片段——这些片段的长度过短，以至于现实世界中的随机问题往往根本找不到真正匹配的主题内容可供检索。后来该语料库扩展到从四个实际存储库中获取的约600篇文档，从而产生了近2,750个片段。
随着更大规模数据集的引入，同样的 taintUniqueValue 问题现在会同时影响两份真实的近似重复报告，并能准确定位根本原因：在处理大型缓冲区时，String.fromCharCode.apply 方法超出了 JavaScript 引擎允许的参数数量上限。值得注意的是，由于该问题的严重程度被评定为高，independent_review 仍会将其升级以便人工审查——按照设计，无论诊断的准确性如何，严重程度的问题都会被升级处理。准确性并不能免除通过风险审查的步骤。
这里更重要的启示是：引用一个真实存在的、可检索到的块标识符确实是必要条件，但并非充分条件。声称“模型引用了某些内容”与声称“模型引用的内容是真实且与该漏洞相关的”并非同一回事，而只验证前者情况的系统虽然看似谨慎，实际上只是在机械地处理无用信息。
审批步骤是真正的暂停，而非仅仅是表面上的加载状态
迄今为止所描述的每个阶段都仅提出相应的操作建议。在流程中的某个特定节点之前，没有任何内容会被写入 GitHub。所有上游环节——分类、诊断、审核——都仅负责提出建议；直到这一步骤，才会有内容被写入 GitHub。
await_approval节点会使用完全相同的逻辑来生成即将发布的确切评论（以及在适用情况下的精确标签），无论该问题是通过确定性分支处理的，还是来自已通过审核的ai_investigation诊断结果。之后它会调用LangGraph的interrupt()函数：
def await_approval(state: GraphState) -> dict:
proposed_action = _build_proposed_action(state)
approved = interrupt({
"classification": state["classification"],
"proposed_action": proposed_action,
})
return {"proposed_action": proposed_action, "approved": bool(approved)}
调用interrupt()并不仅仅是在进程空闲时显示“等待批准”的旋转图标——它实际上会暂停图表的运行。若要稍后恢复该进程的运行，就需要通过完全独立的请求来找到同一条被暂停的线程，并使用与最初运行时相同的thread_id，通过类似result = await compiled_graph.ainvoke(Command(resume=True), config)的调用从暂停处继续执行。要使这一操作能够实现，图表的状态必须能够在两次无关的HTTP请求之间保持不变，这就排除了依赖普通进程内存进行存储的可能性。
只有当该简历信息生成后，execute节点才会运行。这里有两点值得注意。首先，权限检查是在节点自身的代码中实现的，而非仅通过图中的边来表达——因此即便未来有重构意外地在execute节点直接添加了快捷边，这一检查依然能够捕获到问题。其次，execute节点会原封不动地发布state[“proposed_action”]中的内容；它不会在之后重新生成相关注释。最终发布的内容正是经过人工确认的内容。
def execute(state: GraphState) -> dict:
if not state.get("approved"):
raise PermissionError("execute() called without explicit approval")
evidence = state["evidence"]
action = state["proposed_action"]
token = state.get("github_token")
result = {
"comment": post_comment(evidence.repo, evidence.issue_number,
action["comment"], token=token)
}
if action.get("label"):
try:
result["label"] = add_label(evidence.repo, evidence.issue_number,
action["label"], token=token)
except requests.HTTPError as exc:
result["label_error"] = str(exc)
return {"execution_result": result}
为何暂停运行的存储后端比看上去更重要
最初的实现依赖于SqliteSaver，它将状态信息保存到后端服务器本地的文件中。这种设置在开发者的机器上可以正常运行。
它在生产环境中的故障表现很容易被忽视：在 Render 这类免费级托管平台上，磁盘存储在重启后不会保持不变。进程会在一段时间无操作后关闭，而在下一个请求到来时会在一个全新的容器中重新启动。
具体来说就是这样的：某个任务执行到 await_approval 步骤后会暂停，等待人工操作。在有人点击批准之前，免费实例就会进入空闲状态并停止运行。接下来的请求会启动一个带有完全空白数据库的新容器。当 Command(resume=...) 被执行时，已经没有内容可以继续运行——那个被暂停的线程的所有历史记录都在没有任何错误或警告的情况下消失了。
解决方案是 AsyncPostgresSaver，它依托于一个独立的 Postgres 数据库（本示例中使用的是 Neon），该数据库与应用程序运行的容器无关：
async with (
AsyncPostgresSaver.from_conn_string(DATABASE_URL, serde=get_serde()) as saver,
AsyncConnectionPool(DATABASE_URL, open=False,
check=AsyncConnectionPool.check_connection) as pool),
):
即便容器被销毁并重新创建，只要 DATABASE_URL 仍然指向同一个数据库，所有暂停的线程都会保留下来。interrupt() 的暂停与恢复机制完全不变——只有暂停状态的实际存储位置发生了变化。
需要特别指出的是：审批通过后的操作是以审批人的身份进行的，而非使用某种共享的部署凭证。该在线应用允许任何 GitHub 用户登录，此后该次操作的所有读取或写入行为都会使用该用户的 OAuth 令牌。像 post_comment(evidence.repo, evidence.issue_number, action["comment"], token=token) 这样的调用会使用审批人的令牌，而非负责部署应用的人的令牌。
这不仅仅关乎身份验证的安全性。它将“是审批人发布了内容”这一说法转变为 GitHub 可以通过查看评论作者来确认的事实，而非仅由应用界面自行宣称的内容。
它还让 GitHub 现有的权限系统无需任何额外代码即可实现有效的管控：仅浏览自己不拥有的仓库的人可以留下评论，但添加标签则需要对该仓库拥有处理权限或写入权限。execute.py 能够优雅地处理这种情况——添加标签的失败操作会被视为部分成功，而不会导致整个请求失败。
无论由谁触发，对大语言模型和嵌入模型的调用仍然使用部署者的 API 密钥来执行，这正是为何要设置按用户计算的每日速率限制以控制成本的原因。
有一个需要明确区分的点：回归评估与能力评估测试的是完全不同的内容，用相同的标准来评判二者是很容易犯的错误。classify()中的路由优先级、independent_review()中的门控逻辑以及execute()中的权限检查，每种都有唯一正确的行为方式，这类测试所能接受的通过率只能是100%，没有商量余地。无论运行多少次，只要安全门控被绕过一次，那就是严重的故障——绝不能通过取平均值来掩盖这一事实。
那个标准与判断generate_diagnosis能否生成优质诊断结果完全不同。后者的评估由模型来完成，确实会随着时间逐步改进，而且根本不可能真正达到100%。将这两种检查视为处于同一评价标准之下是一个常见的误区——“基本”起作用的防护机制实际上根本算不上真正的防护机制。
当前的真实情况
与其夸大其词，不如如实陈述，以下就是事情的客观现状，而非推销话术：
证据收集通过真实的 GitHub REST 请求完成，并被转换为结构化对象。分类基于规则且具有确定性，不涉及大型语言模型。诊断结果来自实际的 OpenAI 请求，该请求会生成结构化输出，其数据来源于每周重新导入的约 2,750 个数据块的 Pinecone 检索结果。独立审核也是通过另一次 OpenAI 请求实现的，但真正起决定作用的是代码而非模型自身的意见。人工审批环节通过真实的 interrupt() 暂停功能实现，随后可通过 Command(resume=...) 继续处理，并对数据从开头到结尾进行逐字节验证。前端和后端分别部署在 Vercel 和 Render 上，完全采用异步处理方式，数据会保存到 Postgres 数据库中。GitHub OAuth 允许任何访问者使用自己的账户登录，后续操作将以该用户的身份执行，同时设有每日使用限额。评估套件包含用于检测安全机制退化的回归测试。
最后这两个不足之处并非被隐瞒——它们被列入发展路线图，是因为它们确实是接下来最值得开发的重点，而非因为被忽视了。
值得应用到自己项目中的经验
构建一个能够在真实系统中运行的智能体，会体现出一些超越这个特定GitHub问题处理工具的通用原则：
应将推理与执行放在不同的代码路径中，而不仅仅是不同的提示语中。像“行动前务必先询问”这样的指令仍然只是一种行为，而行为总会在最需要其保持稳定的时刻出现故障。只有那种在未设置验证标志时绝对拒绝运行的函数，才是真正的界限，而非建议。
当你说某个审核步骤是“独立的”时，必须确保这一点确实成立：要有独立的调用、没有共享的追踪信息，且判定结果是由代码生成而非模型自身生成的文本。模型可以解释其推理过程，但不应由其自己来评分。
不要用“模型指向了真实的内容”来替代“模型指向了相关的内容”。在确定相似度阈值之前，应实际针对真实查询来衡量检索质量。
如果人工干预的暂停时间对您的设计至关重要，那么请明确测试在人工人员响应之前流程重新启动时会发生什么。内存中的状态和临时磁盘在本地测试中可能看起来正常，但会在最容易造成严重后果的地方出现故障。
最后，要明确区分哪些部分真正已完成，哪些仍只是临时替代方案。一个坦诚说明自身缺点的状态表，比那些暗指一切都已经完成的README文件更能赢得信任。
相关阅读
- Agentic AI Explained: From Language Models to Autonomous Agents — 详细介绍了大语言模型如何通过工具、内存、规划、多智能体架构以及MCP集成等方式演变为自主智能体系统。