企业官网建设流程全解析

上个月我们组在搞一个“模拟项目X”——一个面向业务部门的企业内部咨询Agent。调研了一圈之后，我最终把技术栈锁定在 LangGraph 配合 Elasticsearch 上，做出来的效果还挺稳。先说结论：LangGraph 负责把 Agent 编排成一张可控的图，ES 负责给 Agent 塞知识、存记忆、留日志，两者分工完全不同，但合在一起，恰好解决了人机交互 Agents 最头疼的三件事：流程不可控、记忆不持久、回答不可信。

这篇文章我按自己真实的搭建过程来写。整个过程包含思路拆解、系统设计、核心代码、踩坑记录，适合已经用过 LangChain 但想进阶到生产级 Agent 的工程师，也适合想用 ES 当记忆层和数据检索层的后端同学。

1. 为什么用“图”编排人机交互Agents：LangGraph与Elasticsearch的角色拆解

1.1 从“剧情循环”到“状态图”，思维转换是第一道坎

很多人第一步是用 LangChain 的 AgentExecutor 写“LLM 自己决定调不调工具”的循环。demo 跑起来很容易，到了生产级就难受：LLM 可能在一个工具里来回调、重复调用、甚至走着走着就偏离原意。你想在中间插一道人工审批，AgentExecutor 也不是不能做，但状态分散在会话上下文里，恢复、回滚都很别扭。

LangGraph 的底层思路完全不同。它把 Agent 的每一步都抽象成图里的节点，节点之间靠边连接，条件边决定流向。整个过程是显式的：用户消息进来，先进检索节点，再进生成节点，生成完进入审批节点，审批通过才进执行节点。每一步都可以被打断、被观察、被恢复。用生活里的话说，传统 Agent 像一个“自由人”来回跑腿，你只能隔着窗户喊话；LangGraph 像一条流水线，每个工位只干一件事，质检节点能随时拦截不良品。

这个转变对我影响最大的是心态：我不再“猜 LLM 下一步想干嘛”，而是把流程的骨架先定死，再在节点内部给 LLM 足够的自由度。实践下来，稳定性提升非常明显。

1.2 ES在Agent架构里的三重身份

确定了用图编排之后，信息层选型就顺理成章。Elasticsearch 在这个架构里同时干了三件事：

为什么不单独搞一套向量数据库再加一套 Redis？对中小团队来说，ES 一套集群就能同时处理全文搜索、近实时向量检索、结构化查询、生命周期管理，运维成本低不少。在 Agent 场景里，文本检索和向量检索混用是常态，ES 的混合检索能力虽然不如专用向量库极致，但胜在“一个顺手的地方把所有信息都办了”。

2. 系统整体设计：一个带RAG与人工审批的咨询Agent长什么样

2.1 业务场景假设

我拿模拟项目X来举例：这是一个“工单咨询与处理Agent”。用户在小程序里提问，例如：

我们部门想申领一批测试设备，总预算大概八万，需要走什么流程？

Agent 的任务是：从知识库召回“设备采购流程”相关文档，通过工具查询 ES 里的“预算审批规则”业务索引，生成一个包含步骤、材料清单、风险提示的答复草稿。如果预算超过五万元，则必须转人工审批；审批通过的草稿才能推送给用户，否则自动返回修改建议。

这个场景天然需要“人机交互”——不是简单的多轮闲聊，而是机器生成内容、人来做关键决策。

2.2 为什么选LangGraph而不是纯LangChain AgentExecutor

对比下来，LangGraph 带来的几个核心优势正好打在业务痛点上：

• 可中断与恢复：审批节点可以暂停整个图的执行，等人给出“通过”或“修改意见”后，再继续往下走。这个能力在 AgentExecutor 里需要自己造轮子，在 LangGraph 里是原生支持的。
• 状态显式化：所有节点共享一个 State 对象，谁写入什么字段一目了然。排查问题时直接看 State 里每一层写入的结果，不用翻模型调用链。
• 条件边灵活：根据“是否超过金额阈值”“检索结果是否为空”等业务条件，动态决定下一步走哪个分支，比在工具循环里硬编码规则干净得多。

2.3 为什么选Elasticsearch而不是独立向量库加Redis

我仔细评估过专用向量库方案：Milvus 或 pgvector 配 Redis 存缓存。后来还是选了 ES。原因很实际：

• 我们已经有 ES 集群，不需要额外引一个高可用依赖；
• 业务数据不只有向量，还有大量精确过滤条件（比如“仅看华东区”“状态为生效中”）。ES 的 term 查询、range 查询和向量检索可以放在同一个查询里完成，而专用向量库通常要先把过滤字段塞进元数据，再配合标量过滤，查询逻辑反而更绕。
• 混合检索需要“全文+向量”分数融合，ES 在同一个请求里就能返回两类命中，不用自己分两次请求再客户端合并（当然自己合并也行，后面我会给代码）。

2.4 整体架构图

整个系统分四层：

3. 核心实现：State、节点、条件边与ES混合检索代码

3.1 先定义State：全局黑板

LangGraph 里所有节点都读写同一个 State。我用 TypedDict 定义，messages 字段特殊一点，必须用add_messages这个 reducer，这样多个节点往 messages 里追加内容时不会互相覆盖，而是按顺序累积。其他字段默认是“后写覆盖前写”。

from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): # 对话消息，自动累积追加 messages: Annotated[list, add_messages] # 用户最新问题 query: str # 检索命中的知识片段 retrieved_docs: list[dict] # 业务查询结果 business_data: list[dict] # 模型生成的回复草稿 draft_reply: str # 是否需要人工审批 need_human_approval: bool # 审批/执行结果 execution_result: dict

3.2 检索节点：ES混合检索

检索节点是 Agent 的“耳鼻”，负责把用户问题转成两种信号：BM25 关键词命中 + 向量语义命中。我用的 embedding 模型是开源系列里常见的那几款，维度以模型实际输出为准。下面给出 ES 索引 mapping 和检索函数。

ES 索引 mapping 里关键字段是 content 和 content_embedding，一个给传统全文检索用，一个给 kNN 用：

{ "mappings": { "properties": { "title": {"type": "text"}, "content": {"type": "text"}, "category": {"type": "keyword"}, "content_embedding": { "type": "dense_vector", "dims": 1024, "index": true, "similarity": "cosine" }, "updated_at": {"type": "date"} } } }

检索节点的代码：

def embed_text(model, text): # 基于实际部署的embedding模型，返回归一化的向量 return model.encode(text).tolist() def hybrid_search(es_client, index, query_text, query_embedding, size=5): query = { "bool": { "should": [ {"match": {"content": {"query": query_text, "boost": 1.0}}} ], "minimum_should_match": 1 } } knn = { "field": "content_embedding", "query_vector": query_embedding, "num_candidates": 50, "boost": 1.0 } # 注意：8.12及以上版本可以直接传 query 和 knn 参数 resp = es_client.search( index=index, query=query, knn=knn, size=size, source=False, fields=["title", "content", "category"] ) return resp["hits"]["hits"]

传入的num_candidates不是最终返回数，而是 HNSW 近邻搜索阶段先捞出来的候选数量，越大越准、越慢。我一般取最终返回数的 10 倍左右，比如返回 5 条就设 50。

3.3 把一次查询写成完整节点

这里演示一个完整的节点函数，它会提取用户问题、生成向量、调混合检索、把文档写入 state。

def retrieve_node(state: AgentState, es_client, embed_model) -> dict: latest_user_message = state["messages"][-1].content state["query"] = latest_user_message q_vec = embed_text(embed_model, latest_user_message) hits = hybrid_search( es_client, index="kb_docs_v3", query_text=latest_user_message, query_embedding=q_vec, size=5 ) docs = [] for hit in hits: fields = hit["fields"] docs.append({ "title": fields["title"][0], "content": fields["content"][0], "category": fields["category"][0], "score": hit["_score"] }) return {"retrieved_docs": docs}

为什么要在节点里手动把_score拿出来？因为后面生成节点要判断“这次检索到底靠不靠谱”。如果最高分都低于某个阈值，说明知识库里可能没有用户问的东西，此时与其硬编答案，不如让 Agent 说“我暂时没有查到相关资料，已转给人工处理”，避免一本正经地胡说。

3.4 人机交互Agents的审批中断机制

到了最关键的部分：人工审批。LangGraph 的人机交互有两种常用方式，一种是编译图时用interrupt_before指定在哪个节点前打断，一种是在节点内部用interrupt()函数挂起。我用的是带Command方案的第二种，因为业务上需要带上审批上下文。

from langgraph.types import Command, interrupt def approval_node(state: AgentState) -> AgentState: user_decision = interrupt({ "type": "approval_request", "draft_reply": state["draft_reply"], "hint": "确认回复内容是否正确，请回复 approved 或 rejected" }) if user_decision.get("action") == "approved": return {"execution_result": {"status": "approved"}} return {"execution_result": {"status": "rejected", "reason": user_decision.get("reason")}}

执行时是这样的：

config = {"configurable": {"thread_id": "session-user-001"}} # 第一次调用，图会停在 approval_node 里 result = graph.invoke( {"messages": [{"role": "user", "content": "申请八万设备预算，怎么走流程？"}]}, config ) # 前端拿到 interrupt 信息后，人工点了“通过” resume_value = {"action": "approved"} result = graph.invoke(Command(resume=resume_value), config)

这里第一次调用返回时，interrupt()会把打断信息返回给你，但图的状态已经被 checkpointer 持久化了。第二次用同一个thread_id调用Command(resume=...)，不是重新开始，而是从上次停下的节点继续往下走。这是生产级人机交互 Agents 最核心的体验：人可以在任何时候介入，但机器状态不会丢。

3.5 条件边决定要不要人工介入

生成节点完成后，系统要判断是否需要审批。我写了一个判断函数，挂在条件边上：

def should_need_approval(state: AgentState) -> str: # 预算金额超过5万，或者检索置信度低，都触发审批 if state.get("business_data") and state["business_data"][0]["total_amount"] > 50000: return "need_approval" if not state.get("retrieved_docs"): return "need_approval" return "direct_reply" builder = StateGraph(AgentState) builder.add_node("retrieve", retrieve_node) builder.add_node("generate", generate_node) builder.add_node("approval", approval_node) builder.add_node("execute", execute_node) builder.set_entry_point("retrieve") builder.add_edge("retrieve", "generate") builder.add_conditional_edges( "generate", should_need_approval, { "need_approval": "approval", "direct_reply": "execute" } ) builder.add_edge("approval", "execute") graph = builder.compile(checkpointer=checkpointer)

用条件边而不是把“是否审批”写死在某个节点里，好处是以后业务规则变更只需要改判断函数。比如想把阈值从五万改成两万，或者加上“关键词包含紧急”时必须审批，都是函数内部一行改动，图结构完全不用动。

4. 记忆与可观测性：让Agents记住对话、也让团队看得见执行

4.1 用ES存储会话记忆

Agent 如果每次会话都从零开始，用户体验会很割裂。我用一个agent_memory索引存“用户会话摘要”和“关键记忆点”，例如用户偏好、上次未完成的请求、需要重复强调的事项。

{ "mappings": { "properties": { "user_id": {"type": "keyword"}, "session_id": {"type": "keyword"}, "memory_type": {"type": "keyword"}, "content": {"type": "text"}, "summary_embedding": {"type": "dense_vector", "dims": 1024, "index": true, "similarity": "cosine"}, "created_at": {"type": "date"} } } }

每次对话结束，我会把当前会话做一个摘要写入记忆索引。下次 Agent 启动时，用同一个 embedding 模型把用户问题向量化，在记忆索引里做一次近邻搜索，把相关记忆片段拼进 prompt 上下文。这一步让 Agent 显得“有记忆”——它知道用户上次问的是设备采购，这次再说“那批设备怎么样了”时能接住上下文。

4.2 执行轨迹与性能日志

生产环境里，Agent 出了错如果没有日志，排查等于大海捞针。我用一个agent_trace索引记录每一步：

写入时用 ES 的 bulk 批量接口，不要一条条 index，否则吞吐会很难看。我在节点外面包了一个简单的装饰器，记录执行时间。

def trace_node(node_name): def decorator(func): def wrapper(state): start = time.time() try: result = func(state) status = "ok" except Exception as e: status = "error" raise finally: duration = (time.time() - start) * 1000 es_client.index( index="agent_trace", document={ "trace_id": state.get("trace_id"), "node_name": node_name, "duration_ms": duration, "status": status } ) return result return wrapper return decorator

日志进了 ES，后面要做“哪个节点平均耗时最长”“哪个会话中断次数最多”这类分析就非常方便，直接用聚合查询就行，不需要额外搭监控系统。

4.3 Checkpointer：LangGraph持久化机制的落地选择

LangGraph 的 checkpointer 决定了图的状态存在哪。开发阶段我直接用MemorySaver，重启就丢。生产环境我把 checkpointer 换成了外部持久化存储，保证进程重启后会话还能恢复。LangGraph 本身提供了几个官方实现，对接关系型数据库或 Redis 都不复杂。关键是要理解：checkpointer 必须和thread_id绑定，同一会话的多次invoke才能串成一条完整的流。

我踩过一个坑：最开始用MemorySaver，测试人工审批时一切正常，一旦服务重启，再调Command(resume=...)就报错找不到状态。原因就是 checkpointer 没有持久化。所以我的建议是：只要你的 Agent 要上生产，checkpointer 必须换成持久化方案，否则人机交互的“中断恢复”就名存实亡。

5. 常见问题与排查技巧：从死循环到人机中断恢复

5.1 图执行死循环或超过递归限制

LangGraph 默认有递归次数限制，大概是 25 次左右。如果你的条件边写得不够严密，Agent 会在几个节点之间反复横跳，最后报RecursionLimit相关异常。

排查思路：先把图结构打出来，检查条件边分支是否覆盖了所有可能情况。我最开始写should_need_approval时，只考虑了“金额超限”和“文档为空”，漏掉了“检索文档不为空但生成节点本来就要直接回复”的情况，结果在特定输入下 generate 之后又回到 retrieve，形成环。后来把判断函数里的分支穷举了一遍，问题消失。

5.2 混合检索排序效果不好

BM25 和向量分数的量纲完全不一样，直接相加等于让某一边主导。我后来做了一个简单的归一化融合：把两类分数各自做 min-max 归一化，再按权重相加。

def normalize_score(scores): min_s = min(scores) max_s = max(scores) if max_s == min_s: return [1.0 for _ in scores] return [(s - min_s) / (max_s - min_s) for s in scores] def fuse_scores(bm25_hits, knn_hits, alpha=0.4): # alpha 表示 BM25 的权重，1-alpha 表示向量权重 ...

调整 alpha 时我的习惯是：拿 50 条真实问题做评测，人工看 top 5 的相关性，选效果最稳的那个值。不要凭感觉定，要有测试集。

5.3 Elasticsearch版本与查询语法兼容

ES 8.0 到 8.12 之间，kNN 查询写法变化比较大。早期版本把knn放在 search body 里，新版本推荐query和knn平级传入。如果你用的是旧客户端，直接传knn参数会报参数错误。我建议方案是：团队统一锁 ES 版本，客户端版本跟服务器版本保持一致，并且在集成测试里覆盖“文本命中”“向量命中”“混合命中”三种情况。

另外dense_vector维度必须与 embedding 模型输出一致，否则索引创建直接失败。改模型这件事在 ES 里成本不低，最好在项目初期就确定 embedding 模型，后续不要随便换。

5.4 人机中断后状态恢复报错

这个我在 4.3 提过，这里再补充一个细节：使用Command(resume=...)恢复时，thread_id必须和第一次调用完全一致，而且要用同一套编译好的 graph 对象。如果服务是多实例部署，checkpointer 必须共享存储，否则实例 A 中断的会话，实例 B 接不住。

实操验证时，建议用下面的检查清单：

5.5 一些独家避坑心得

• 别把太多逻辑塞进条件边判断函数里。条件边最好只做“看状态、返回下一个节点名”这件事，涉及复杂计算的放前置节点完成。
• ES 索引要做别名管理。我建了kb_docs_v3就是为了后面重灌数据时直接切换别名，不会影响线上 Agent。
• 生成回复草稿后，最好把引用的文档 ID 一起存下来。这样人工审批时能看到“这段回复来自哪篇文档”，提高人对机器结果的信任感。

做完模拟项目X，我个人最大的感受是：人机交互 Agents 的难点不在模型，而在工程。LangGraph 把流程控制从“魔法”变成了“图纸”，ES 把知识和记忆从“临时变量”变成了“持久资产”。只要把这两层理清楚，后面的稳定性和可维护性就是水到渠成的事。

最后再分享一个我在实际调试中养成的小习惯：不管跑什么用例，都顺手把 state 的完整快照打进日志里。很多诡异问题——比如检索结果为空但生成节点还在硬答、审批状态没写进去导致下游取不到——都是靠这份快照一眼定位的。这比事后对着报错猜半天高效得多。