企业官网建设流程全解析

1. 半年卡壳的真相：AI Agent 到底难在哪

去年秋天我接手了一个 AI Agent 项目，目标是让智能体自动完成一套跨系统的业务流程——读取工单、查询知识库、调用内部接口、生成处理结论、回写结果。听起来不复杂对吧？我当时也是这么想的，结果从立项到真正跑通，整整卡了半年。

这半年里我踩的坑，几乎覆盖了 AI Agent 开发的所有典型难点。最开始我以为难点在模型选型，试了几个主流大模型之后发现，模型能力其实够用，真正要命的是工程化落地这一层。Agent 不是聊天机器人，它要自主决策、要调用工具、要维护状态、要在多轮交互中保持上下文一致，还要扛住并发。这些东西单靠一个 Prompt 是撑不起来的。

我后来复盘，卡壳的核心原因有三个。第一是架构选型摇摆不定，一开始想用纯代码编排，后来想上 LangChain，再后来听说 LangGraph 更适合有状态的 Agent，来回折腾浪费了大量时间。第二是并发问题被严重低估，单机跑 demo 的时候一切正常，一上压测就各种超时、状态错乱、工具调用重复执行。第三是可观测性缺失，Agent 内部到底怎么想的、为什么调用了这个工具而不是那个、哪一步耗时最长，完全没有手段去看，出了问题只能靠猜。

这三个问题不解决，Agent 就永远停留在 demo 阶段。我查了大量资料，也看了不少开源项目，发现大家遇到的坑高度相似，但系统性地讲清楚“怎么从零搭一个能扛并发的生产级 Agent”的内容并不多。这也是我今年特别关注 iRTE2026 的原因——从目前放出的议题方向看，Agent 架构、并发处理、工程化部署这些正是重点讨论的内容。

这篇文章我打算把自己这半年的踩坑经验、架构选型的思考过程、并发问题的解决方案、以及完整的实操步骤全部整理出来。不管你是刚接触 AI Agent 的新手，还是已经卡在某个环节的老手，应该都能从中找到对自己有用的东西。我会尽量说人话，把每个技术决策背后的“为什么”讲清楚，让你不光能抄作业，还能理解为什么这么抄。

2. 架构选型：为什么我最终选了 Rust + LangGraph 的混合方案

2.1 主流 AI Agent 架构的横向对比

在动手之前，我花了大概两周时间调研市面上主流的 Agent 架构方案。这个过程非常痛苦，因为每种方案都有自己的拥趸，网上的对比文章要么太浅要么太偏。我最后自己整理了一张对比表，从几个关键维度来评估。

纯 Prompt 编排就是所有逻辑都写在系统提示词里，让模型自己决定调用什么工具。这种方式做 demo 最快，但一旦流程复杂起来，提示词会膨胀到不可维护，而且模型经常会“忘记”之前的指令。我第一个版本就是这么做的，跑到第三周就彻底放弃了。

LangChain 的优势是生态成熟，工具调用、记忆管理、输出解析都有现成组件。但它的抽象层太厚，出了问题很难定位，而且默认的状态管理是基于内存的，多实例部署时会有问题。LangGraph 在 LangChain 的基础上引入了图结构来管理 Agent 的状态流转，每个节点是一个处理步骤，边定义了流转条件，这种显式的状态机模型让复杂流程变得可控。

Spring AI 是 Java 生态的选择，如果你团队本身是 Java 技术栈，它的并发处理能力确实强，但 AI 生态的丰富度不如 Python 系。Rust 自研则是性能和并发的最优解，但开发成本极高，工具调用、模型对接这些都要自己写。

2.2 我为什么最终选择混合方案

经过反复权衡，我最终采用的方案是：核心编排层用 LangGraph 做状态管理，高并发的工具调用层用 Rust 重写，两者通过 gRPC 通信。这个方案听起来复杂，但逻辑其实很清晰。

LangGraph 负责的是 Agent 的“大脑”部分——理解用户意图、规划任务步骤、决定下一步调用哪个工具、维护对话状态。这部分逻辑复杂但并发量不大，用 Python 写开发效率最高。Rust 负责的是“手脚”部分——实际执行工具调用、访问数据库、调用外部 API、处理大量数据。这部分是并发瓶颈所在，用 Rust 写能榨干硬件性能。

两者之间通过 gRPC 通信，LangGraph 把工具调用请求发给 Rust 服务，Rust 执行完把结果返回。这样做的另一个好处是工具调用层可以独立扩容，Agent 编排层不需要跟着扩。

提示：如果你的团队没有 Rust 经验，不要硬上。可以先用 Python 的 asyncio + httpx 做工具调用层，性能虽然不如 Rust，但比同步调用强很多。等业务量上来了再考虑替换。

2.3 状态管理的核心设计

Agent 的状态管理是最容易被低估的部分。我一开始觉得状态就是对话历史，后来发现远不止于此。一个生产级 Agent 的状态至少包含以下几层：

• 会话状态：当前对话的上下文，包括用户输入、Agent 回复、工具调用记录
• 任务状态：当前任务执行到哪一步了，哪些步骤已完成，哪些待执行
• 工具状态：每个工具调用的输入输出、执行耗时、是否成功
• 全局状态：跨会话的长期记忆，比如用户偏好、历史任务摘要

LangGraph 用 TypedDict 来定义状态结构，每个节点可以读取和修改状态。这里有个关键设计决策：状态是不可变的，每个节点返回的是状态的一个新版本，而不是直接修改原状态。这样做的好处是天然支持回滚和重放，出问题的时候可以精确复现。

from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END import operator class AgentState(TypedDict): messages: Annotated[list, operator.add] current_task: str tool_results: dict retry_count: int next_action: str def plan_node(state: AgentState): # 根据当前状态决定下一步 return {"next_action": "call_tool", "current_task": "query_database"} def tool_node(state: AgentState): # 执行工具调用 result = call_rust_service(state["current_task"]) return {"tool_results": {**state["tool_results"], "query": result}} graph = StateGraph(AgentState) graph.add_node("plan", plan_node) graph.add_node("tool", tool_node) graph.add_edge("plan", "tool") graph.add_conditional_edges("tool", should_continue, {"continue": "plan", "end": END})

这段代码看起来简单，但有几个细节值得注意。Annotated[list, operator.add]表示 messages 这个字段在更新时是追加而不是覆盖，这是 LangGraph 的一个巧妙设计。retry_count用来控制重试次数，防止 Agent 陷入死循环。should_continue是一个条件函数，根据当前状态决定是继续循环还是结束。

3. 并发处理：AI Agent 怎么扛住真实流量

3.1 并发问题的根源分析

Agent 的并发问题和普通 Web 服务不太一样。普通 Web 服务主要是 IO 密集，加机器、加连接池基本能解决。Agent 的并发问题更复杂，因为它涉及模型推理和工具调用两个不同性质的环节。

模型推理是计算密集型的，而且有速率限制。你不可能无限制地并发调用模型 API，大多数服务商都有 QPS 限制。工具调用则可能是 IO 密集（查数据库、调 API）也可能是计算密集（数据处理、文件转换）。这两者的并发策略完全不同。

我遇到的具体问题包括：多个请求同时到达时，模型调用排队导致响应时间飙升；工具调用重复执行，同一个查询被跑了三次；状态更新冲突，两个请求同时修改同一个会话的状态导致数据错乱。

3.2 分层并发策略

解决这些问题需要分层处理。我把整个 Agent 的执行链路拆成三层，每层用不同的并发策略。

第一层是请求接入层，负责接收用户请求、做限流和排队。这一层用令牌桶算法做限流，每个用户有独立的令牌桶，防止单个用户占满资源。排队用优先级队列，VIP 用户的请求优先处理。

第二层是 Agent 编排层，负责状态管理和任务规划。这一层的关键是会话隔离，每个会话有独立的状态实例，不同会话之间不共享状态。LangGraph 的 checkpointer 机制可以做到这一点，每个会话 ID 对应一个独立的状态快照。

第三层是工具执行层，负责实际的工具调用。这一层用 Rust 实现，核心是异步任务池 + 结果缓存。每个工具调用被封装成一个异步任务，提交到任务池执行。如果同一个工具用相同参数被调用多次，直接返回缓存结果。

use tokio::sync::Semaphore; use std::sync::Arc; use dashmap::DashMap; pub struct ToolExecutor { semaphore: Arc<Semaphore>, cache: Arc<DashMap<String, String>>, } impl ToolExecutor { pub async fn execute(&self, tool_name: &str, params: &str) -> Result<String, Error> { let cache_key = format!("{}:{}", tool_name, params); if let Some(cached) = self.cache.get(&cache_key) { return Ok(cached.clone()); } let _permit = self.semaphore.acquire().await?; let result = self.call_tool(tool_name, params).await?; self.cache.insert(cache_key, result.clone()); Ok(result) } }

这段 Rust 代码的核心是Semaphore控制并发数，DashMap做结果缓存。Semaphore的 permit 数量根据工具类型动态调整，IO 密集型的工具可以给更多 permit，计算密集型的少给一些。

3.3 模型调用的并发优化

模型调用是另一个瓶颈。我的优化策略是批量合并 + 异步流式。批量合并是指把多个小请求合并成一个大请求发给模型，减少 API 调用次数。异步流式是指用流式接口获取模型输出，边生成边处理，降低首字延迟。

具体实现上，我用了一个请求聚合器，在 50ms 的时间窗口内收集所有待处理的模型调用请求，合并成一个 batch 发给模型。模型返回后再拆分结果分发给各个请求方。这个策略把模型调用的 QPS 降低了大约 60%。

注意：批量合并会引入额外的延迟，对于延迟敏感的场景要谨慎使用。我的做法是对延迟敏感的任务走单独通道，不参与批量合并。

3.4 压测数据与调优过程

调优不能靠感觉，必须有数据支撑。我用 Locust 做压测，模拟了 100、500、1000 三个并发级别。初始版本的表现在 500 并发时就开始恶化，P99 延迟超过 10 秒，错误率飙升到 15%。

通过分析火焰图，我发现瓶颈主要在模型调用的同步等待上。优化后，1000 并发下 P99 延迟控制在 3 秒以内，错误率低于 0.5%。关键优化点包括：模型调用改为异步、工具调用加缓存、状态更新加锁粒度细化、连接池参数调优。

4. 从零搭建：完整实操流程与关键配置

4.1 环境准备与依赖安装

先说环境。我的开发环境是 Ubuntu 22.04，Python 3.11，Rust 1.75。Python 版本建议不要低于 3.10，因为 LangGraph 用到了不少新语法特性。Rust 版本建议用 stable 就好，不需要 nightly。

Python 侧的依赖主要是这几个：

pip install langgraph langchain-core fastapi uvicorn httpx pydantic pip install grpcio grpcio-tools # gRPC 通信 pip install redis # 状态持久化

Rust 侧的依赖在 Cargo.toml 里配置：

[dependencies] tokio = { version = "1.35", features = ["full"] } tonic = "0.10" dashmap = "5.5" serde = { version = "1.0", features = ["derive"] } serde_json = "1.0"

这里有个坑要注意：tonic的版本要和prost匹配，不然编译会报错。我一开始用了最新版的 tonic，结果和项目里已有的 prost 版本冲突，折腾了半天。

4.2 LangGraph 编排层的搭建

编排层的核心是定义状态和节点。我前面已经展示了状态定义，这里补充节点的具体实现。规划节点负责调用模型做任务分解，工具节点负责调用 Rust 服务，判断节点负责决定是否继续循环。

from langchain_core.messages import HumanMessage, AIMessage from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4", temperature=0) def plan_node(state: AgentState): messages = state["messages"] prompt = build_planning_prompt(messages, state["tool_results"]) response = llm.invoke([HumanMessage(content=prompt)]) # 解析模型输出，提取下一步动作 action = parse_action(response.content) return { "messages": [response], "next_action": action["type"], "current_task": action.get("task", "") } def should_continue(state: AgentState): if state["retry_count"] > 3: return "end" if state["next_action"] == "finish": return "end" return "continue"

规划节点的提示词设计很关键。我试过很多版本，最终稳定下来的结构是：先给 Agent 设定角色和可用工具列表，然后给出当前任务状态和已完成的步骤，最后要求模型输出 JSON 格式的下一步动作。JSON 格式一定要在提示词里明确，不然模型会输出各种奇怪的格式。

4.3 Rust 工具执行层的实现

Rust 侧的核心是一个 gRPC 服务，接收工具调用请求并返回结果。我用 tonic 来定义服务接口：

syntax = "proto3"; service ToolService { rpc ExecuteTool (ToolRequest) returns (ToolResponse); } message ToolRequest { string tool_name = 1; string params = 2; string session_id = 3; } message ToolResponse { bool success = 1; string result = 2; string error = 3; int64 elapsed_ms = 4; }

服务实现里，每个工具是一个独立的 handler，注册到一个路由表里。调用时根据 tool_name 查找对应的 handler 执行。这里的关键是超时控制，每个工具调用都要设置超时，防止某个工具卡死拖垮整个服务。

async fn execute_tool(&self, request: Request<ToolRequest>) -> Result<Response<ToolResponse>, Status> { let req = request.into_inner(); let start = Instant::now(); let result = tokio::time::timeout( Duration::from_secs(30), self.dispatch(&req.tool_name, &req.params) ).await; let elapsed = start.elapsed().as_millis() as i64; match result { Ok(Ok(output)) => Ok(Response::new(ToolResponse { success: true, result: output, error: String::new(), elapsed_ms: elapsed, })), Ok(Err(e)) => Ok(Response::new(ToolResponse { success: false, result: String::new(), error: e.to_string(), elapsed_ms: elapsed, })), Err(_) => Ok(Response::new(ToolResponse { success: false, result: String::new(), error: "timeout".to_string(), elapsed_ms: elapsed, })), } }

4.4 状态持久化与恢复

Agent 的状态必须持久化，不然服务重启后所有会话都丢了。我用 Redis 做状态存储，LangGraph 的 checkpointer 机制可以无缝对接。

from langgraph.checkpoint.redis import RedisSaver checkpointer = RedisSaver.from_conn_string("redis://localhost:6379") graph = workflow.compile(checkpointer=checkpointer) # 执行时传入 thread_id 作为会话标识 config = {"configurable": {"thread_id": session_id}} result = graph.invoke(input_state, config)

Redis 的 key 设计要注意，我用的是agent:state:{session_id}作为前缀，方便批量管理和清理。TTL 设置为 24 小时，过期的会话自动清理。

提示：Redis 一定要开持久化，不然重启后状态还是会丢。我用的是 AOF 持久化，每秒同步一次，兼顾性能和安全。

5. 踩坑实录：那些让我熬夜的典型问题

5.1 工具调用重复执行

这个问题困扰了我很久。现象是同一个工具调用被执行了多次，导致数据库里出现重复数据。排查后发现原因是模型在流式输出时被中断，重试机制触发了重复调用。

解决方案是在工具执行层加幂等键。每次工具调用生成一个唯一的 idempotency key，执行前先检查这个 key 是否已经执行过，如果执行过直接返回缓存结果。幂等键的生成规则是session_id + tool_name + params_hash，这样同一个会话里相同的工具调用只会执行一次。

5.2 状态更新冲突

多轮对话中，如果用户快速发送多条消息，可能会出现状态更新冲突。比如第一条消息触发的工具调用还没返回，第二条消息已经开始处理了，两个流程同时修改状态，导致数据错乱。

解决方案是会话级锁。每个会话在处理时先获取锁，处理完释放。锁用 Redis 的 SETNX 实现，设置合理的超时时间防止死锁。如果获取锁失败，说明该会话正在处理中，新消息进入队列等待。

5.3 模型输出格式不稳定

模型有时候不按要求的 JSON 格式输出，导致解析失败。我试过几种方案：一是用 function calling 强制格式，二是用输出解析器做容错，三是在提示词里加 few-shot 示例。

最终稳定下来的方案是三者结合。优先用 function calling，如果模型不支持就降级到输出解析器，同时在提示词里放两个示例。解析失败时不要直接报错，而是把原始输出返回给模型让它重新格式化，最多重试两次。

5.4 常见问题速查表

5.5 几个让我印象深刻的坑

第一个是时区问题。Agent 处理时间相关的任务时，模型返回的时间没有时区信息，导致计算错误。后来我在提示词里明确要求所有时间都带时区，并且在工具层做统一转换。

第二个是长文本截断。模型有上下文长度限制，长对话会被截断。我的解决方案是做对话摘要，当对话超过一定轮数时，用模型把前面的对话压缩成摘要，只保留关键信息。

第三个是并发下的日志混乱。多个请求的日志混在一起，排查问题非常困难。后来我在日志里加了 session_id 和 request_id，用结构化日志输出，排查效率提升了很多。

6. 可观测性建设：让 Agent 的每一步都可见

6.1 为什么可观测性如此重要

Agent 和普通服务最大的区别是它的决策过程是不透明的。普通服务出问题，看日志基本能定位。Agent 出问题，你看到的是“它调用了错误的工具”或者“它给出了奇怪的回答”，但为什么这么决策，完全不知道。

我吃过这个亏。有一次 Agent 在生产环境频繁调用一个不该调用的工具，查了半天日志也没找到原因。后来把模型的完整输入输出都打出来，才发现是提示词里有个歧义表述，模型理解偏了。

6.2 三层可观测性体系

我建立的可观测性体系分三层。第一层是链路追踪，用 OpenTelemetry 记录每个请求的完整链路，包括模型调用、工具调用、状态变更。每个环节的耗时、输入输出都记录下来。

第二层是决策日志，专门记录 Agent 的决策过程。每次规划节点执行时，把模型的输入提示词、输出结果、解析后的动作都记录下来。这些日志单独存储，方便后续分析 Agent 的决策模式。

第三层是业务指标，包括任务完成率、平均执行步数、工具调用成功率、用户满意度等。这些指标用 Prometheus 采集，Grafana 展示。

6.3 关键监控指标

这些指标不是拍脑袋定的，是根据实际运行数据统计出来的基线。比如任务执行步数，正常任务平均 4-5 步完成，超过 8 步基本可以判定是 Agent 在绕圈子，需要人工介入。

7. 关于 iRTE2026 的期待与准备

7.1 我为什么一定要去

回到标题。做 AI Agent 卡了半年，我太清楚这个领域的坑有多深了。很多问题不是靠看文档能解决的，需要和有实战经验的人交流。iRTE2026 从目前公布的议题看，Agent 架构设计、并发处理、工程化落地这些正是我最关心的方向。

我特别关注的是生产环境下的 Agent 稳定性这个议题。Demo 跑通和线上稳定运行之间隔着巨大的鸿沟，我踩过的坑希望别人不用再踩。另外多 Agent 协作也是我下一步要探索的方向，单个 Agent 的能力边界很明显，多个 Agent 分工协作可能是突破方向。

7.2 我准备带的几个问题

去之前我整理了三个具体问题，准备在现场找答案。第一个是Agent 状态管理的最佳实践，特别是跨会话的长期记忆怎么设计。第二个是模型调用的成本优化，大规模部署时怎么控制成本。第三个是Agent 的评测体系，怎么量化评估一个 Agent 的好坏。

7.3 给同样卡壳的朋友的建议

如果你也在做 AI Agent 并且卡住了，我的建议是先把架构定下来，不要频繁换方案。我前三个月最大的浪费就是来回换架构。定下来之后，优先解决并发和状态管理，这两个问题不解决，后面都是白搭。最后，可观测性一定要早做，不要等到出问题了才想起来加日志。

Agent 这个方向变化很快，新技术新框架层出不穷。但底层的东西是不变的：状态管理、并发控制、错误处理、可观测性。把这些基础打牢，上层用什么框架都能快速上手。iRTE2026 我肯定会去，到时候如果有新的收获，再整理出来分享。