一、我们要解决什么问题

企业场景中有一类高频需求：用户上传一份 PDF 或 Markdown 文档，然后围绕文档内容提问。听起来很简单，但实际做起来至少要回答三个工程问题：

1. 文档内容找不到答案怎么办？ 不能简单回复"不知道"，需要自动补充外部信息。
2. 回答基于什么证据？ 生产环境必须可审计，每条结论要能追溯到来源。
3. 流程中工具调用失败了怎么办？ 搜索引擎超时、模型返回异常，不能让整个流程崩溃。

这三个问题的共同特征是：它们不是"模型能力"问题，而是流程编排问题。这正是 LangGraph 的用武之地。

二、为什么选择 LangGraph 而非 LangChain 的链式调用

LangChain 的 Chain 和 LCEL 擅长处理线性流程：输入 → 处理 → 输出。但当流程中间出现以下需求时，链式结构就不够用了：

• 条件分支：验证通过走 A 路径，不通过走 B 路径
• 循环：验证不通过 → 补充搜索 → 重新回答 → 再验证
• 共享状态：多个节点需要读写同一份"工作记忆"
• 可观测性：需要记录每个节点的决策轨迹

LangGraph 用有向图的方式解决了这些问题：

Node（节点）= 一个处理步骤Edge（边）= 固定的执行顺序Conditional Edge（条件边）= 基于状态的动态路由State（状态）= 所有节点共享的数据结构

这使得"思考 → 行动 → 验证 → 可能回到行动"的 Agent 模式，可以用声明式的图来表达，而非用嵌套 if-else 和 while 来硬编码。

三、DocNexus 的整体架构

3.1 执行流程

START→ init（解析文档 → 文本切片 → 向量化建库）→ plan（分析问题 → 提取检索关键词）→ retrieve（向量检索文档切片）→ answer（基于证据生成回答草稿）→ verify（评估草稿充分性与置信度）├─ 充分 → finalize（结构化输出）→ END└─ 不足 → search（调用搜索引擎补充）→ answer → verify → ...

关键点在于 verify → search → answer → verify 这条循环边——它是"思考-行动-验证"闭环的图式表达。

3.2 工程分层

项目没有用单文件 all-in-one 的方式，而是按职责拆分为独立模块：

src/docnexus_graph/├── config.py # 配置与环境变量├── state.py# 状态数据结构定义├── io.py # 文档读取与切片├── retrieval.py# 向量检索层├── prompts.py# 提示词模板与 JSON 解析├── tools.py# 外部工具封装（搜索 + 重试）├── nodes.py# 业务节点实现├── graph.py# 图编排├── runner.py # CLI 执行入口├── api.py# FastAPI 服务接口└── __init__.py

每层只做一件事。比如你想把向量检索从"内存余弦相似度"替换为 FAISS，只需改 retrieval.py，不影响任何其他文件。这是将 Agent 从 PoC 推向生产的基础。

四、状态设计：Agent 的"工作记忆"

LangGraph 中，State 是所有节点的共享数据结构。每个节点读取 state，处理后返回 state 的增量更新。

class QAState(TypedDict, total=False):question: str# 用户问题document_path: str # 文档路径chunks: List[DocumentChunk]# 文档切片planned_terms: List[str] # 检索关键词retrieval_context: List[SourceEvidence] # 文档检索证据search_context: List[SourceEvidence]# 搜索引擎证据answer_draft: str# 回答草稿need_search: bool# 是否需要搜索补充confidence: int# 置信度 0-100missing_points: List[str]# 验证发现的缺口iteration: int # 当前搜索轮次max_iterations: int# 搜索上限trace: List[str] # 执行轨迹日志error: str # 最近一次错误

State 的设计原则：

• 可驱动：need_search + iteration + max_iterations 直接驱动条件边路由
• 可追溯：trace 记录每个节点的关键动作
• 可审计：retrieval_context 和 search_context 保留完整证据链
• 可扩展：后续加多轮对话记忆、权限字段，只需扩展 TypedDict

其中，证据块 SourceEvidence 的统一结构也值得注意：

class SourceEvidence(TypedDict, total=False):source_id: str # "D1"=文档片段，"S1"=搜索结果source_type: str # "document" 或 "search"title: str # 标题location: str# 文档字符范围或 URLscore: float # 相似度/匹配权重content: str # 片段正文

所有证据——无论来自文档还是搜索引擎——统一为同一结构，这样 answer 和 verify 节点无需关心证据来源于哪个工具。

五、节点实现：逐个拆解

5.1 init 节点：文档解析与向量化

init 是流程的第一个节点，完成三件事：读文档、切片、建向量索引。

def node_init(state: QAState, retriever: VectorRetriever) -> Dict[str, Any]:if state.get("chunks"):return {"trace": _trace(state, "init:文档已初始化，复用缓存切片")}path = state["document_path"]raw_text = read_document_text(path)chunk_size = state.get("chunk_size", 1200)overlap = state.get("chunk_overlap", 180)chunks = split_text(raw_text, size=chunk_size, overlap=overlap)retriever.build(chunks) # 关键：向量化入库return {"chunks": chunks,"iteration": 0,"need_search": False,"confidence": 0,"trace": _trace(state, f"init:解析完成，共 {len(chunks)} 段"),# ... 重置各项字段}

切片策略：滑动窗口方式。每个 chunk 1200 字符，窗口步进 size - overlap，保证相邻切片有 180 字符的重叠区。重叠的目的是避免关键语句恰好被切断。

def split_text(text: str, size: int, overlap: int) -> List[DocumentChunk]:step = max(size - overlap, 150)while cursor < len(clean_text):end = min(cursor + size, len(clean_text))chunk_text = clean_text[cursor:end].strip()if chunk_text:chunks.append({"chunk_id": f"D{index}", "text": chunk_text, "start": cursor, "end": end})index += 1cursor += stepreturn chunks

每个切片都带上 start/end 字符位置和全局唯一 chunk_id（D1、D2...），后续来源追溯直接使用这个 ID。

5.2 plan 节点：Think 阶段

plan 不是直接回答问题，而是让 LLM 先分析问题、提取高价值检索词：

def node_plan(state: QAState, llm: ChatOpenAI) -> Dict[str, Any]:prompt = ChatPromptTemplate.from_messages([("system", PLAN_TEMPLATE),("human", "问题：{question}"),])response = llm.invoke(prompt.format_prompt(question=state["question"]).to_messages())payload = extract_json(response.content)planned_terms = payload.get("focus_terms", [])# ... 校验与回退if not planned_terms:planned_terms = extract_keywords(state["question"])# 规则回退return {"planned_terms": planned_terms, "trace": ...}

这里有一个细节：当 LLM 返回的 JSON 解析失败时，不会抛异常，而是回退到基于规则的关键词提取（正则匹配中英文词汇 + 停用词过滤）。这种"主策略 + 回退策略"的模式在生产系统中很重要。

5.3 retrieve 节点：向量检索

retrieve 用 plan 阶段生成的关键词拼成查询串，交给向量检索器做语义匹配：

def node_retrieve(state: QAState, retriever: VectorRetriever) -> Dict[str, Any]:query_terms = state.get("planned_terms", [])if not query_terms:query_terms = extract_keywords(state["question"])query = " ".join(query_terms)results = retriever.search(query, top_k=state.get("top_k", 4))# 将结果转为统一的 SourceEvidence 格式context = [{"source_id": item["chunk_id"],"source_type": "document","title": f"文档片段 {item.get('chunk_id')}","score": float(item.get("score", 0.0)),"content": item["text"],}for item in results]return {"retrieval_context": context, "trace": ...}

5.4 answer 节点：基于证据生成回答

answer 节点合并文档证据（retrieval_context）和搜索证据（search_context），交给 LLM 生成结构化草稿：

ANSWER_TEMPLATE = ("你是企业知识问答助手。仅基于候选证据作答，要求给出结构化中文答案：n""1) 一句话结论；2) 分点事实依据；3) 风险与边界。n""每条事实必须标注来源标签，例如 [D1]、[S2]。")

提示词要求模型在回答中标注 [D1]、[S2] 等来源标签，这是来源追溯的关键机制——不是回答完再贴来源，而是在生成过程中就要求逐条标注。

5.5 verify 节点：自我评估

verify 是整个流程中最关键的决策节点。它让 LLM 扮演"答案验证器"角色，评估草稿是否被证据充分支撑：

VERIFY_TEMPLATE = ("你是答案验证器。请判断草稿是否能被来源证据完整支撑。"'只返回 JSON：{"need_search":true/false,"confidence":0到100,''"missing_points":["缺少哪部分证据"],"rationale":"一句话判断"}')

返回的 JSON 被解析后写入 state：

need_search = bool(payload.get("need_search", False))confidence = max(0, min(100, int(payload.get("confidence", 0))))missing_points = [str(item) for item in missing][:6]

confidence 被钳位到 [0, 100]，missing_points 限制最多 6 条——这些都是防御性处理，避免 LLM 输出异常时让下游逻辑失控。

5.6 search 节点：外部工具调用

当 verify 判定 need_search=true 时，条件边将流程路由到 search 节点：

def node_search(state: QAState) -> Dict[str, Any]:if state.get("iteration", 0) >= state.get("max_iterations", 0):return {"need_search": False, "trace": _trace(state, "search:达到上限，停止搜索")}query = state["question"]if state.get("missing_points"):query = f"{query} {' '.join(state['missing_points'])}"try:rows = search_with_retry(query=query, top_k=...)# 转为 SourceEvidence 格式，source_id 从 "S1" 开始编号...except Exception as e:return {"iteration": state.get("iteration", 0) + 1, "need_search": False, "error": str(e)}

三个值得注意的设计：

1. 迭代上限：iteration >= max_iterations 时强制停止，避免无限循环
2. 查询增强：把 missing_points（验证发现的缺口）拼入搜索关键词，让搜索更精准
3. 异常兜底：搜索失败不会让流程崩溃，而是记录错误、递增迭代计数、正常流转到下一个节点

5.7 finalize 节点：结构化输出

finalize 收集所有证据和草稿，生成两种格式的输出：

• Markdown 报告：便于人直接阅读
• JSON payload：便于系统下游接入

payload = {"project": PROJECT_NAME,"question": state["question"],"document": Path(state["document_path"]).name,"answer": state.get("answer_draft", ""),"confidence": state.get("confidence", 0),"sources": [{"source_id": ..., "type": ..., "title": ..., "score": ...} for ...],"trace": state.get("trace", []),"error": state.get("error", ""),}

六、向量检索的实现细节

项目没有引入 FAISS 或 Chroma 等外部向量数据库，而是用 numpy 实现了一个轻量的内存向量检索器。这个决定是刻意的：减少外部依赖，让读者聚焦 LangGraph 本身。

6.1 建索引

class VectorRetriever:def build(self, chunks: List[DocumentChunk]) -> None:texts = [chunk["text"] for chunk in chunks]matrix = np.array(self.embedder.embed_documents(texts), dtype=np.float32)self.vectors = matrixself.chunks = chunksself.norms = np.linalg.norm(matrix, axis=1)

embed_documents 一次性将所有切片文本向量化。预计算每个向量的 L2 范数（norms），用于后续余弦相似度的分母。

6.2 检索

def search(self, query: str, top_k: int) -> List[DocumentChunk]:q = np.array(self.embedder.embed_query(query), dtype=np.float32)q_norm = float(np.linalg.norm(q))norms = np.where(self.norms == 0, 1e-9, self.norms)# 防除零scores = self.vectors.dot(q) / (norms * q_norm) # 余弦相似度rank = np.argsort(scores)[::-1][:top_k]result = []for idx in rank:value = float(scores[int(idx)])if value <= 0:continue # 过滤负相关chunk = dict(self.chunks[int(idx)])chunk["score"] = valueresult.append(chunk)return result

余弦相似度 = 向量内积 / (两个向量的 L2 范数之积)。取 Top-K 后还有一步过滤：score <= 0 的不返回。这意味着如果文档跟问题完全不相关，检索结果可以是空的——这时 verify 节点会触发搜索补充。

6.3 为什么不直接用关键词匹配

关键词匹配的致命问题是词汇鸿沟：

• 问题："项目的核心目标"
• 文档中写的是："本系统的主要设计意图"

两者语义相同，但关键词没有交集。向量检索通过将文本映射到语义空间，天然解决了同义表达、长短句变形的问题。而且切换向量检索不改变图的结构——retrieve 节点的输入输出不变，只是内部实现从"字符串匹配"变成了"向量运算"。

七、图的编排：声明式定义执行拓扑

所有节点和边的组装在 graph.py 中完成：

def build_graph(llm: ChatOpenAI, embedding_model: OpenAIEmbeddings) -> StateGraph:retriever = VectorRetriever(embedding_model)workflow = StateGraph(QAState)workflow.add_node("init", lambda state: node_init(state, retriever))workflow.add_node("plan", lambda state: node_plan(state, llm))workflow.add_node("retrieve", lambda state: node_retrieve(state, retriever))workflow.add_node("answer", lambda state: node_answer(state, llm))workflow.add_node("verify", lambda state: node_verify(state, llm))workflow.add_node("search", node_search)workflow.add_node("finalize", node_finalize)workflow.add_edge(START, "init")workflow.add_edge("init", "plan")workflow.add_edge("plan", "retrieve")workflow.add_edge("retrieve", "answer")workflow.add_edge("answer", "verify")workflow.add_conditional_edges("verify",route_after_verify,{"search": "search", "finalize": "finalize"},)workflow.add_edge("search", "answer")workflow.add_edge("finalize", END)return workflow.compile()

条件路由函数 route_after_verify 是整个图中唯一的决策逻辑：

def route_after_verify(state: QAState) -> str:if state.get("need_search") and state.get("iteration", 0) < state.get("max_iterations", 0):return "search"return "finalize"

两个条件同时满足时走 search，否则走 finalize。max_iterations 是安全阀，防止 verify 持续判定为"不足"而导致无限循环。

八、工具调用的失败重试

搜索引擎的网络调用在生产环境中不可靠。search_with_retry 实现了指数退避重试：

def search_with_retry(query: str, top_k: int = 4, max_retries: int = 2) -> List[Dict[str, str]]:last_error = Nonefor attempt in range(max_retries + 1):try:from duckduckgo_search import DDGSwith DDGS() as ddgs:items = list(ddgs.text(query, max_results=top_k))return [{"title": ..., "url": ..., "snippet": ...} for item in items]except Exception as e:last_error = eif attempt >= max_retries:breaktime.sleep(0.8 * (2 ** attempt)) # 0.8s, 1.6sraise RuntimeError(f"搜索引擎调用失败：{last_error}")

等待时间递增：第一次失败等 0.8 秒，第二次等 1.6 秒。即使最终仍然失败，上层 node_search 也会捕获异常并带着 error 信息正常流转，而不是让整个图崩溃。

九、来源追溯：不是锦上添花，而是生产必需

DocNexus 的追溯机制贯穿三个层面：

层面一：证据编号

每个证据块在产生时就被分配全局唯一 ID：

• 文档切片：D1、D2、D3...（在 split_text 中生成）
• 搜索结果：S1、S2、S3...（在 node_search 中编号，且基于已有数量递增）

层面二：回答中的内联引用

通过提示词约束，LLM 在生成回答时必须标注来源：

每条事实必须标注来源标签，例如 [D1]、[S2]。

这样生成的回答类似：

本项目的核心目标是构建企业级文档问答系统 [D1]，支持 PDF 和 Markdown 两种格式 [D2]，类似的方案可参考 RAG 范式 [S1]。

层面三：结构化来源清单

finalize 节点在 JSON payload 中输出完整的来源列表：

{"sources": [{"source_id": "D1", "type": "document", "title": "文档片段 D1", "location": "0-1200", "score": 0.87},{"source_id": "S1", "type": "search", "title": "RAG 最佳实践", "location": "https://...", "score": 1.0}]}

审核人可以通过 source_id 快速定位到原始证据，验证回答的可信度。

十、执行轨迹（Trace）：每一步都有据可查

每个节点在返回 state 增量时都会追加一条 trace 记录：

def _trace(state: QAState, line: str) -> List[str]:return state.get("trace", []) + [line]

最终输出的 trace 类似：

init:解析完成，共 12 段plan:提取检索词 ['核心目标', '设计意图', '系统功能']retrieve:向量检索到 4 条文档证据answer:生成草稿，来源=4，长度=342verify:need_search=True, confidence=55, missing=2search:新增 4 条外部证据，round=1answer:生成草稿，来源=8，长度=520verify:need_search=False, confidence=85, missing=0

这条轨迹是 debug 和评测的核心依据。你能清楚看到：第一次验证置信度只有 55 且有 2 条缺口，触发了搜索补充；第二轮加入搜索证据后置信度提升到 85，不再需要搜索，流程收敛。

十一、FastAPI 服务化

项目提供了 api.py 将整个 Agent 封装为 HTTP 服务：

@app.post("/api/ask", response_model=AskResponse)def ask(payload: AskRequest) -> AskResponse:result = run_once(document_path=payload.document,question=payload.question,model=payload.model or OPENAI_MODEL,temperature=payload.temperature,max_search_rounds=payload.search_rounds,chunk_size=payload.chunk_size,chunk_overlap=payload.chunk_overlap,top_k=payload.top_k,embedding_model=payload.embedding_model or OPENAI_EMBEDDING_MODEL,)return AskResponse(status="ok", markdown=..., payload=...)

请求体通过 Pydantic BaseModel 做参数校验（类型、范围），返回体包含 markdown（人读）和 payload（机读）两种格式。健康检查 GET /health 可直接接入监控探针。

启动方式：

python run_api_server.py# 等效于：uvicorn src.docnexus_graph.api:app --host 0.0.0.0 --port 8000

十二、提示词工程中的防御性设计

与 LLM 交互时，最常见的问题是模型不按要求返回 JSON。DocNexus 的应对策略是：

宽容解析

def extract_json(text: str) -> dict:start = text.find("{")end = text.rfind("}")if start == -1 or end == -1 or end <= start:return {}try:return json.loads(text[start : end + 1])except Exception:return {}

不要求模型的响应"只有 JSON"——允许前后有解释文字，只提取第一个 {...} 块。解析失败返回空字典而非抛异常。

回退策略

plan 节点中，如果 LLM 返回的 focus_terms 为空或解析失败，回退到基于规则的关键词提取：

if not planned_terms:planned_terms = extract_keywords(state["question"])

verify 节点中，confidence 字段做类型转换保护：

try:confidence = int(raw_confidence)except Exception:confidence = 0confidence = max(0, min(100, confidence))

这些看似琐碎的处理，决定了系统在面对 LLM 输出不稳定时能否正常运行。

十三、一些设计权衡

为什么不用 LangChain 的 Tool Calling

LangGraph 本身支持 ToolNode，但本项目选择让 node_search 直接调用搜索函数，而非走 LLM 的 function calling 通道。原因是：在这个场景里，"是否搜索"的决策已经由 verify 节点的条件边完成了，不需要 LLM 再做一次工具选择。减少一次 LLM 调用 = 减少延迟和成本。

为什么用内存向量库而非 FAISS

目标是让项目"开箱即用"，不增加额外安装步骤。numpy 是 Python 生态的标配，而 FAISS 在某些平台上安装有兼容性问题。内存向量库完全足以处理单文档（几十到几百个切片）的场景。需要升级时只需替换 retrieval.py。

为什么 verify 用 LLM 而非规则

规则验证（比如"回答中是否包含来源标签"）能覆盖形式检查，但无法判断"回答的论据是否与证据矛盾"或"是否遗漏了问题的某个方面"。这类语义层面的评估，目前只有 LLM 能做。代价是多一次 API 调用，但在文档问答场景中准确性优先于延迟。

十四、总结与扩展方向

DocNexus GraphRAG Agent 的核心价值不在于代码量（核心逻辑约 300 行），而在于它把 Agent 的关键能力用 LangGraph 的语言做了清晰表达：

如果你要把这个项目推向生产，建议优先做这三件事：

1. 持久化向量索引：用 FAISS 或 Chroma 替换内存检索，支持大文档和多文档
2. 受控搜索源：把 DuckDuckGo 替换为企业内网搜索 API，满足合规要求
3. 评测闭环：建立问答评测集，统计 confidence 分布与命中率，持续校准提示词