检索是 RAG 系统的搜索引擎，分块则是这个搜索引擎的基础。分块太长、太短、有噪声、切错了位置——随便犯哪个错LLM 都会有问题。行业里有句话流传很广："分块决定了 RAG 质量的 70%。"

这个说法不夸张：好的分块让检索器拿到完整、有上下文、真正相关的信息；差的分块把文档打成碎片，上下文断裂，LLM 只能靠"编"来填补空白。

什么是分块？

RAG 的起点是文档收集与摄取：把所有原始材料（文档、文章、知识库条目）汇聚到一起。在进入检索环节之前，这些文档要经过文本分块处理也就是切分成更小的、有意义的片段。

每个分块应当是连贯且自包含的，这样检索器才能在面对查询时快速定位、排序，并返回最相关的信息。

分块就是在生成 Embedding 之前，把大段文本拆成更小语义单元的过程。检索器真正搜索的对象而不是整篇文档就是这些分块。

分块做得好，文档中的内容就能被干净地捕获，上下文得以保留LLM 能做出有意义的推理。分块做得差，语义被割裂检索充满噪声。向量存储、Embedding 模型、Reranker——这些统统排在分块之后，分块才是真正的起点。

固定大小分块

这是最简单的方式。按预设的字符数或 Token 数直接切分，比如每 500 个 Token 一块完全不管句子和段落的边界在哪。

速度快，行为可预测，处理大规模、结构混乱的数据集时很实用。但缺点也很明显——语义经常被拦腰切断。一个句子在这个分块里开了头，到下一个分块才结束，Embedding 的语义表达力就会打折扣。

实践中一般会在相邻分块之间设置一定的重叠来缓解这个问题：

 from langchain.text_splitter import RecursiveCharacterTextSplitter  

splitter = RecursiveCharacterTextSplitter(  
    chunk_size=500,  
    chunk_overlap=50  
)  

 chunks = splitter.split_text(long_text)

切分文本时，连续的分块之间通常会加入一小段重叠区域来维持上下文的连贯。所谓重叠，就是前一个分块的尾部几句话，在下一个分块的开头再出现一次。

这么做是为了防止跨越分块边界的关键信息丢失。没有重叠的话，检索器可能只拿到部分内容LLM 因此漏掉了关键上下文，给出残缺甚至误导性的回答。重叠量一般控制在分块长度的 10% 到 20%，在冗余和效率之间找一个平衡点。

固定大小分块适合的场景包括日志文件、邮件、代码仓库，以及结构参差不齐的大型语料库。

基于句子的分块

这种方式按完整句子来划分文本，而不是按任意长度一刀切。每个分块至少包含一个或多个完整的句子，语法完整，语义连贯。

好处是每个分块都是一个有意义的思想单元。检索器向 LLM 返回的信息更精确、更易理解，碎片化回答的风险降低不少。实际使用中通常也会搭配小幅重叠，进一步保证分块之间的衔接。

基于段落的分块

以完整段落为单位切分，不再拘泥于单个句子或固定 Token 数。这种方式天然保留了文档的结构和行文节奏，检索器更容易抓到完整的想法。

每个分块往往对应一个独立的主题或子主题，LLM 处理起来更从容，也更容易给出准确的回答。对长篇文档、研究论文、综述类文章来说，段落级分块效果不错。和句子级分块一样，也可以加重叠来保持连贯。

语义分块

语义分块的切入点不是长度，而是语义本身。它利用 Embedding 或相似度分数来识别文本中天然的断裂点——主题切换、上下文转折、章节边界。

产出的分块语义清晰度更高，边界和语义对齐，检索质量有明显提升，尤其在知识库、技术文档、结构化文章这类内容上效果突出。代价是计算开销更大而且分块长度不一致，后续处理需要额外考虑。

 from langchain_experimental.text_splitter import SemanticChunker  
 from sentence_transformers import SentenceTransformer  
   
 model = SentenceTransformer("all-MiniLM-L6-v2")  
 chunker = SemanticChunker(model, breakpoint_threshold=0.4)  
   
 chunks = chunker.split_text(long_text)

如果文档质量高、主题流转有明确脉络，语义分块往往是精度最高的选择。

递归分割

递归分割是固定大小和语义分块之间的一个折中方案。核心思路是优先尊重文档结构，只有在必要时才进一步拆分。

具体做法是先尝试按标题切分。如果某个章节还是太长，就按段落切。段落还不够就按句子。句子仍然超限，最后才按字符兜底。这样得到的分块既保有语义完整性，尺寸也在可控范围内。

 recursive_splitter = RecursiveCharacterTextSplitter(  
     separators=["\n## ", "\n### ", "\n", ". ", ""],  
     chunk_size=600,  
     chunk_overlap=80  
 )  
   
 chunks = recursive_splitter.split_text(long_doc)

开发者文档、技术手册、学术论文、研究报告——凡是层级结构明确的内容，递归分割都很适合。

滑动窗口分块

有些文本的语义天然是跨句分布的。法律合同、科学论文、长段论证，一个完整的意思可能横跨好几个句子。滑动窗口就是为这种场景设计的。

它不生成彼此独立的分块，而是创建相互重叠的窗口。比如窗口大小 400 Token，每次滑动 200 Token，这样相邻的分块之间有一半的内容是共享的，语义在边界处不会断裂。

上下文保持得很好，但分块数量会膨胀，存储和检索的成本都会上升。

法律 RAG、金融分析、医学文献检索、合规审查——这些领域用滑动窗口的比较多。

层次化分块

层次化分块是一个多层级的架构：小分块负责细粒度精确检索，中等分块支撑平衡的推理，大分块维持全局上下文。

检索时，系统先用小分块锁定精确位置，再把关联的大分块拉进来补充完整上下文。这种组合能有效压制幻觉，提升推理的深度。

企业级 RAG 系统和 LlamaIndex 这类多粒度检索框架，背后都有层次化分块的影子。

实践中常见的分块失误

多数 RAG 项目翻车根源都是分块层面的问题。分块过大模型被不相关的细节淹没。分块过小语义丧失殆尽。句子被拦腰切断、不相关的段落被混到一个分块里，Embedding 质量直接垮掉。没有重叠，上下文断裂。没有元数据，检索器找不到方向。还有一个常见错误——所有类型的文档套用同一种分块策略。

分块没有万能方案。政策文件和教科书不一样，通话记录和研究论文不一样。策略必须跟着文档类型和检索任务走。

总结

分块不是一个可有可无的预处理环节，它是 RAG 管道的脊梁。好的分块是一个有意义的、自包含的知识单元；差的分块是一个孤零零的碎片，把 LLM 带向歧途。

检索是引擎分块是燃料。燃料的质量决定了整个系统是输出干净、可靠的结果，还是不断产出噪声和幻觉。LLM 本身再好，也救不了烂分块。

https://avoid.overfit.cn/post/e6520bd283254415ae61cfa28fb2ef32

作者：Abinaya Subramaniam

大型语言模型（LLM）在理解和生成上下文连贯的对话方面取得了巨大成功。然而，它们固有的“记忆缺陷”——即有限的上下文窗口——严重制约了其在跨会话、跨应用的长时间交互中保持一致性的能力。一旦对话超出上下文长度，LLM 就会像一个“失忆”的伙伴，忘记用户的偏好、重复提问，甚至与之前确立的事实相矛盾。想象一下这个场景：你告诉一个AI助手你是素食主义者且不吃乳制品。几天后，当你向它寻求晚餐建议时，它却推荐了烤鸡。这种体验无疑会削弱用户对AI的信任和依赖。
为此，PolarDB PostgreSQL版（以下简称PolarDB-PG）全新推出一站式记忆管理AI应用，使智能体能够跨会话、跨应用持续保留用户偏好、事实背景与历史交互信息，解决大模型有限上下文窗口和跨会话记忆丢失的核心痛点。

1.构建智能体记忆面临的挑战

开发、运维效率低：记忆系统构建需要选型或开发记忆引擎，对接各类数据库系统以及模型服务，开发、运维成本高；当前主流记忆框架均为检索式记忆系统，后端需要对接关系库、向量库甚至图库等多种记忆库资源，数据一致性难以保障；对于AI快速驱动业务演进而言，企业客户很难对数据库、记忆引擎、模型服务等底层设施做到完全兜底。
记忆生成、检索效果不佳：不少企业客户希望自建记忆系统，但遇到记忆事实、偏好等提取不全导致关键信息遗漏；因记忆系统整体链路长导致记忆检索延迟高，导致交互问答不流畅；对需要用户画像和记忆推理需求场景，因只能提供向量化记忆导致检索结果相关性欠强；因模型算法效果在记忆整体应用中起到十分关键的作用，模型算法与提示词配置的灵活度，也直接决定了方案迭代的速度。
系统成本压力大：随用户规模增长，系统在并发度、存储规模等方面缺乏弹性扩缩容能力；硬件、多种数据库系统、记忆引擎等多license系统，系统费用成本叠加；对于持续爆发增长的记忆库，缺乏支撑记忆生命周期管理的有效机制等。

2.PolarDB一站式记忆管理系统

针对上述挑战，PolarDB-PG推出全新AI应用——一站式长记忆管理系统正式发布上线。PolarDB-PG记忆管理真正融合了图+向量一站式记忆库 + 开放记忆引擎 + 模型算子能力，提供了全面白屏化的参数配置，提示词策略管理以及模型算法混池加速能力，支撑“记忆读写 → 上下文注入 → 模型推理 → 结果反馈”的完整闭环。一期已接入Mem0（发音为 "mem-zero"）记忆引擎，兼容开源 Mem0社区生态，使智能体能够跨会话、跨应用持续保留用户偏好、事实背景与历史交互信息，从而实现真正的个性化和持续学习体验。

PolarDB-PG一站式记忆管理系统架构

1、记忆引擎

目前，PolarDB-PG已支持Mem0框架，全面兼容开源项目Mem0社区生态；支持Mem0（向量基础版）和Mem0g（图增强版）；对开源Mem0系统实现了系列增强，包括：中英文模型接入能力；支持根据userid多图管理功能；支持根据userid向量分区管理功能；同步、异步记忆写出能力；增加sslmode连接参数，支持ssl连接；支持提示词模版的定制优化以及Mem0企业版的部分功能对齐等。后续PolarDB-PG还将和MemOS合作，为AI构建专属的“记忆操作系统”，实现记忆全生命周期的精细化管理与动态调度。

2、一站式记忆库

PolarDB-PG向量数据库引擎 + 图数据库引擎一站式组合。其中，向量数据库引擎采用经优化的PGVector插件，PGVector在PG社区已经被广泛应用，具备十分良好的AI生态支持。图数据库引擎兼容开源AGE（A Graph Extension，为Apache软件基金会的顶级项目），且经过PolarDB-PG与云原生能力的增强融合以及在大量图客户上的多年应用改进和性能优化，不仅表现成熟稳定，且具备在百亿级规模图场景下仍然保持万级以上QPS和百毫秒以下的查询延迟的极佳表现。记忆库支持云原生集中式版本或分布式版本，无需担心扩展性风险。

3、PolarDB模型算子

统一采用PolarDB模型算子提供模型部署、推理、调度体系化能力。模型在记忆管理中扮演了核心的角色，其中：大语言模型LLM负责从用户与智能体的对话中自动提取出具有长期价值的关键事实与偏好，同时用于新记忆与已有记忆的融合（增删改）以及基于图的实体三元组信息抽取；嵌入模型EMB负责将关键信息转化为高维向量，实现高效的语义检索；Rerank模型则用于记忆召回后的精排序。模型调用和推理的效率占据了用户体验的关键一环，本方案支持多种形式的模型对接途径，包括：a. 数据库自有模型算子形式；b. 百炼模型服务形式；通过高度优化的链路，大幅提升记忆相关推理效率。

4、图形化控制台

PolarDB-PG记忆管理在PolarDB系统中属于AI应用的一种形式，提供了全面图形化的管理界面：


模型算法与数据库配置

记忆提取策略配置

记忆图谱可视化

5、AI应用构建平台

支持沿用Mem0已对接周边生态，包括：Langchain、LangGraph、AgentOps、LlamaIndex等框架/平台；支持将PolarDB记忆引擎作为插件加入到Dify框架实现任务流定制；支持与阿里云AgentRun企业级 AI Agent 一站式基础设施平台‌和AgentScope开源智能体开发框架的一体化整合应用。

PolarDB记忆管理支持Dify的插件化应用

3.系统核心优势

1、端到端一站式记忆管理

开箱即用，融合记忆引擎、记忆库、模型算子服务以及KVCache加速能力，免去多系统联调、维护成本。

2、图形化配置，简单易用

控制台可视化管理多项目记忆，支持记忆引擎、模型算法、提示词策略等灵活配置；支持多项目的记忆管理能力，记忆项目配置支持完全采用图形化的界面形式；支持对记忆引擎、记忆模型算子、记忆提取策略（提示词）等选项的配置；提供极简REST API或客户端SDK，自动完成记忆事实提取、记忆增删改融合以及记忆搜索。

3、图式记忆和向量式记忆融合，记忆更准，成本更低

支持基于向量的简单记忆库模式，同时支持图（Graph）+ 向量融合的高级记忆库模式；支持图结构的关系推理（时序推理、因果推理等），记忆召回率提升40%；一站式解决图库、向量库和关系库，大幅降低TCO成本。

4、内置集成大模型推理服务，保障稳定服务

支持配置LLM、Embedding、Rerank等模型算子用于记忆生成与管理；特别采用模型算子混池架构，常规请求路由至百炼，请求规模超过百炼限定时，自动切换自有资源兜底；自有模型算子VPC内网部署，模型推理延迟相比百炼可进一步提升30%+。

5、多租户、多图粒度管理，资源可扩展

支持按项目、业务线等维度划分独立的记忆空间，保障资源隔离、数据安全与规模可扩展；支持按UserID自动切子图管理，记忆规模不受限，同等记忆规模下召回效率提升50%+。

6、百亿级记忆规模，毫秒级响应

经历百亿级规模向量、图谱数据客户最佳实践，满足万级高QPS、<50ms低延迟在线服务高标准；跨会话长记忆+会话内基于KVCache Token加速，请求延迟下降88.3%（上下文长度200k，30并发）。

4.记忆库应用场景适配

PolarDB记忆管理支持两类长记忆方案，基于纯向量记忆库方案，和向量记忆库+图记忆库的组合方案，分别适用于以下场景：

1、纯向量记忆库方案‌

• 应用场景‌：

  • 需要快速语义检索的对话场景，例如在线客服、实时聊天机器人等。
  • 成本敏感型应用，假设需要分别采购向量数据库和图数据库两种产品或服务，采用纯向量方案能减少至少一半的产品费用支出。技术特点‌：通过LLM提取对话关键事实并向量化存储。采用动态阈值控制检索范围，平衡召回率与精准度。

2、向量记忆库+图记忆库组合方案‌

• 应用场景‌：

  • 复杂关系推理场景：如医疗诊断（跟踪患者病史和药物相互作用）、旅行规划（整合航班、酒店、景点等关系）等。
  • 长期知识管理‌：通过三元组（实体1-关系-实体2）结构化存储知识，适合构建企业级知识库或跨会话连贯性要求高的智能助手，如需跟踪用户偏好演变关系的智能座仓AI助手、AI伴侣等，做到长期个性化服务。
  • 动态演进型系统‌：知识图谱支持增量更新和子图检索，适合业务规则频繁变化的场景（如金融风控中的动态规则库）。

• 技术特点‌：

  • 向量库处理语义搜索，图库存储实体间关联关系。
  • 支持时间感知或因果推理的动态知识图谱更新。
  • 基于Mem0g方案，通过两阶段流水线实现结构化记忆。

两种方案的互补性体现在：向量+图虽能处理复杂关系，但检索效率上带来更大挑战；而纯向量方案在简单场景中更高效，但缺乏对深层关系的建模能力。实际部署时，可结合业务复杂度与实时性需求进行混合架构设计。

5.应用展望

目前，PolarDB记忆管理已落地新能源车企开发助手、教育伴学等场景，在文本记忆、多模态记忆等多种场景进行了全面适配，大幅提升个性化交互沉浸感。除以上场景外，PolarDB记忆管理还在企业知识库、旅游规划、电商导购、医疗陪护等多个关键领域展现出客户价值，成为推动AI应用从“对话机器人”迈向“智能伙伴”的关键基础设施。PolarDB 与 Mem0/MemOS 的深度整合，让每一位开发者都能轻松构建真正“记得住、懂你心、扛得住、响应快”的记忆系统。

6.了解更多

欢迎搜索钉钉群号：169605009089入群与技术专家交流！

这两年，大模型几乎成了开发者的“标配工具”：
写代码、查资料、做总结、当智能助手。

但你有没有认真想过一个问题：

我们真的必须把所有请求都发到云端 API 吗？

随着模型体积持续下降、硬件性能快速提升，以及 Ollama 这类工具逐渐成熟，
本地运行大模型，已经从早期的“极客尝鲜”，演进为一种可以在真实项目中落地的工程方案。

这篇文章，我们就来完整走一遍：

如何使用 LangChain，基于最新 Runnable API，调用本地启动的 Ollama 模型，构建一个真正可用的本地大模型应用。

一、为什么选择 LangChain + Ollama？

先给结论：

Ollama 解决“模型怎么跑”，LangChain 解决“能力怎么用”。

这是目前本地大模型场景中，最自然、最稳定的一种组合方式。

1️⃣ Ollama：本地大模型的“Docker”

你可以把 Ollama 理解为：
专门为大模型设计的一层运行时基础设施。

它解决的问题非常聚焦：

• 统一模型的下载、管理与启动
• 对外提供标准化 HTTP API（默认端口 11434）
• 支持 LLaMA、Qwen、Mistral、DeepSeek 等主流模型
• Mac / Linux / Windows 全平台可用
• 天然适合 Docker / 私有化部署

一句话总结：

Ollama 把“跑模型”这件事，做成了基础设施能力。

2️⃣ LangChain：AI 应用的“控制中心”

如果你只是想“问一句、回一句”，直接调 Ollama API 当然也没问题。
但一旦进入真实工程场景，需求会迅速复杂化：

• Prompt 如何复用、版本化？
• 对话上下文如何管理？
• 如何组合多步推理？
• 后续怎么接 RAG、Agent、工具调用？

这些正是 LangChain 擅长的事情：

• Prompt 模板与结构化输入
• Runnable / LCEL 编排能力
• 对话历史（Memory）管理
• Tool、RAG、Agent 的统一抽象
• 可自然演进到 LangGraph

所以一个非常自然的分工是：

LangChain 负责“编排与逻辑”，Ollama 负责“模型与算力”。

二、准备工作：本地启动 Ollama 模型

1️⃣ 使用 Docker 部署 Ollama（推荐）

docker run \
-d \
--restart=always \
--name ollama \
--gpus=all \
-p 11434:11434 \
-v /home/data/ollama:/root/.ollama \
ollama/ollama

如果你对部署细节感兴趣，可以参考我之前的文章：

• 《如何使用 Ollama 打造你的本地 AI 助手》
• 《为本地部署的大模型添加 API Key 认证：Nginx 实现方案》

2️⃣ 拉取并运行模型

以 qwen3:8b 为例：

ollama pull qwen3:8b

简单测试：

ollama run qwen3:8b

如果可以正常对话，说明模型已经在本地成功运行。

三、LangChain 接入本地 Ollama（OpenAI 协议）

接下来进入核心部分：
如何用 LangChain 调用本地 Ollama？

1️⃣ 安装依赖

pip install langchain langchain-openai

这里我们使用 OpenAI 兼容协议，这是目前最稳定、生态最完整的一种方式。

2️⃣ 创建 Ollama LLM（ChatOpenAI）

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    name="ollama-ai",
    model="qwen3:8b",
    base_url="http://localhost:11434/v1",
    api_key="your api key",
    temperature=0.7,
    timeout=300,
)

几个关键点说明：

• model 必须与 Ollama 中的模型名称一致
• base_url 指向 Ollama，并注意使用 /v1 后缀
• 这里使用的是 OpenAI 标准协议，不是 Ollama 私有 API

3️⃣ 最简单的一次调用

response = llm.invoke("用一句话解释什么是 LangChain")
print(response)

到这里，你已经完成了：

LangChain → 本地 Ollama → 本地大模型

这条完整调用链。

四、进阶用法：Prompt + Runnable（LCEL）

在真实项目中，几乎不会直接“裸调”模型。

1️⃣ PromptTemplate

from langchain_core.prompts import PromptTemplate

prompt = PromptTemplate(
    input_variables=["question"],
    template="你是一个资深后端工程师，请用简洁、专业的语言回答：{question}",
)

2️⃣ 输出解析（StrOutputParser）

from langchain_core.output_parsers import StrOutputParser

parser = StrOutputParser()

显式的输出解析，是 LangChain 新 API 的重要特征：

• 输出类型清晰
• 便于后续切换为 JSON / Pydantic
• 更适合工程化

3️⃣ Runnable 组合（推荐写法）

chain = prompt | llm | parser

response = chain.invoke({
    "question": "为什么本地部署大模型越来越流行？"
})
print(response)

这就是 LangChain 当前主推的 LCEL（表达式）写法，
比早期的 LLMChain 更透明、也更可组合。

五、加入 Memory：真正的本地对话能力

⚠️ 一个非常重要的变化：

在新的 Runnable 体系中，
Memory 不再是 Chain 的“隐藏参数”，而是显式的状态管理。

1️⃣ 定义对话历史存储

from langchain_core.chat_history import InMemoryChatMessageHistory

store = {}

def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

2️⃣ Prompt 显式消费 history（关键）

from langchain_core.prompts import PromptTemplate

prompt = PromptTemplate(
    input_variables=["history", "question"],
    template="""
         你是一个资深后端工程师。

         以下是之前的对话历史：
         {history}

         当前用户问题：
         {question}

         请基于上下文给出连贯、准确的回答。
    """.strip()
)

这是很多人第一次使用 RunnableWithMessageHistory 时最容易忽略的一点：
历史是否生效，取决于 Prompt 是否显式使用 {history}。

3️⃣ 构建带记忆的 Runnable

from langchain_core.runnables.history import RunnableWithMessageHistory

chain = prompt | llm | parser

chat_chain = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="question",
    history_messages_key="history",
)

4️⃣ 调用（带 session_id）

config = {"configurable": {"session_id": "local-chat"}}

print(chat_chain.invoke(
    {"question": "什么是 Ollama？"},
    config=config
))

print(chat_chain.invoke(
    {"question": "它和 LangChain 有什么关系？"},
    config=config
))

到这里，你已经拥有了一个：

• 支持上下文
• 完全本地
• 状态可控

的对话系统。

而且 所有数据都只存在你的本地机器上。

六、这套方案适合谁？

非常适合：

• ✅ 本地工具 / 桌面应用
• ✅ 内部知识库 / 私有 RAG
• ✅ 研发辅助工具（代码、文档、SQL）
• ✅ 对数据安全敏感的企业场景
• ✅ 学习大模型工程化的开发者

不太适合：

• ❌ 超大并发场景
• ❌ 极限性能 / 超大模型
• ❌ 面向公网的 C 端产品

七、一些来自实践的工程建议

最后分享几点真实踩坑后的经验：

1. 模型别贪大

   • 7B / 8B 是当前本地部署的性价比甜点位

2. Prompt 比模型更重要

   • 本地模型对 Prompt 非常敏感

3. LangChain 要“模块化使用”

   • Prompt / LLM / Parser / Memory 明确分层

4. Memory 要可演进

   • InMemory → Redis → 数据库 → Checkpointer

5. Ollama 非常适合私有化场景

   • Docker + 内网 + 权限控制，工程成本极低

结语

过去一年，我们讨论最多的问题是：

“该用哪个云端大模型？”

而现在，越来越多开发者开始认真思考：

“哪些能力，其实可以放回本地？”

LangChain + Ollama 并不是为了“替代云”，
而是为我们提供了一个：

真正可控、可组合、可落地的本地大模型方案。

如果你正在做：

• 本地 AI 工具
• 私有化大模型
• Agent / RAG 工程实践

那么这套组合，非常值得一试。

如果你觉得这篇文章对你有帮助，欢迎 点赞 / 转发 / 收藏。
下一篇，我会继续分享 LangGraph 在本地大模型场景下的实战用法。

摘要

随着 2026 AI 元年智能体规模化落地趋势的凸显，从 0 到 1 搭建适配场景的智能体应用成为企业与开发者的核心需求。本文聚焦智能体应用搭建的全流程，明确以“感知-决策-执行”闭环为核心的搭建逻辑，系统拆解需求定位、技术选型、模块构建、测试迭代四大核心步骤，梳理数据安全、成本控制、人机协同等关键注意事项，为不同场景下的智能体搭建提供可落地的实操指南，助力实现技术能力与场景需求的精准匹配。

目录

一、搭建核心逻辑：以场景为锚点，构建闭环能力

二、全流程搭建步骤：从需求到落地的四阶段拆解

2.1 需求拆解与场景定位

2.2 技术选型与框架选择

2.3 核心功能模块搭建

2.4 测试验证与迭代优化

三、关键注意事项：保障搭建质量与落地效果

3.1 数据安全与合规管控

3.2 成本适配与轻量化部署

3.3 人机协同边界的明确

四、智能体应用搭建 QA 问答

五、结语

六、参考文献

一、搭建核心逻辑：以场景为锚点，构建闭环能力

从 0 到 1 搭建智能体应用，核心逻辑是围绕具体场景需求，构建“感知-决策-执行-优化”的完整能力闭环，而非单纯堆砌技术模块。这一逻辑的核心是“场景适配优先”——智能体的价值最终体现在对具体场景的赋能效果上，因此搭建全流程需以场景痛点为锚点，确保每一步构建都服务于问题解决。

从本质来看，智能体应用搭建的核心价值在于打破传统 AI 工具的功能局限：通过整合感知、决策、执行能力，让智能体从“被动响应工具”升级为“主动解决问题的数字助手”，既降低人工干预的频次与成本，又能实现技术能力在同类场景中的规模化复用，为企业智能化转型提供轻量化、可落地的解决方案。

二、全流程搭建步骤：从需求到落地的四阶段拆解

智能体应用搭建需遵循“需求-技术-构建-验证”的线性逻辑，拆解为四大核心阶段，确保每一步衔接顺畅、目标明确，避免因流程缺失导致落地失败。

2.1 需求拆解与场景定位

需求拆解与场景定位是搭建的基础，直接决定后续技术选型与模块设计的方向，核心要完成“目标明确-流程梳理-边界界定”三件事。首先，明确核心应用目标：需精准定位智能体的核心功能，例如客户服务场景的“7×24 小时问答与问题闭环”、工业场景的“设备故障预测与维护提醒”，避免功能泛化导致资源浪费。其次，梳理场景全流程：拆解目标场景中的核心环节与关键节点，例如“用户咨询-需求识别-数据检索-答案生成-反馈收集”，明确智能体的输入（如用户指令、设备数据）、输出（如回答内容、维护指令）及各环节的触发条件。最后，界定能力边界：明确智能体可自主完成的任务与需人工介入的场景，例如复杂问题的转接、高风险决策的审核，避免因能力边界模糊导致用户体验下降。

2.2 技术选型与框架选择

技术选型需遵循“适配性优先、低成本起步”原则，核心围绕“框架-数据-算力”三大核心要素展开。在框架选择上，优先选用支持多工具联动、可扩展性强的开源框架，例如 LangChain、AutoGPT 适合文本类智能体搭建，AgentGPT 适合轻量化场景快速落地，工业智能体可选择适配工业协议的专用框架，降低开发门槛与后续迭代成本。在数据准备上，需搭建“采集-存储-预处理”全链路模块：采集场景相关的结构化（如用户画像、设备参数）与非结构化数据（如文本咨询、设备图像），选择安全合规的存储方案（如企业私有云、加密数据库），通过数据清洗、标注、脱敏等预处理，提升数据质量。在算力配置上，根据场景需求灵活选择部署方式：轻量化场景（如小型客服智能体）可采用云服务器按需付费；复杂场景（如多模态工业智能体）可结合本地算力与边缘计算，平衡性能与成本。

2.3 核心功能模块搭建

核心功能模块搭建需围绕“感知-决策-执行”闭环展开，三大模块相互联动，构成智能体的核心能力。感知模块负责“信息输入与解析”，需支持多类型信息接收，例如文本、语音、图像、传感器数据等，通过 OCR 识别、语音转文字、多模态语义理解等技术，实现信息的精准提取与意图识别，为后续决策提供基础。决策模块是智能体的“核心大脑”，需结合规则引擎与大模型能力：规则引擎用于处理明确的标准化场景（如固定流程的业务办理），大模型用于处理复杂的非标准化场景（如模糊需求解读、多路径选择），通过两者协同实现自主决策，例如根据用户咨询意图匹配对应的服务流程。执行模块负责“动作落地”，需对接场景相关的工具与系统，例如客服智能体对接 CRM 系统实现客户信息调取，工业智能体对接 MES 系统下发维护指令，通过标准化接口确保指令精准执行，同时反馈执行结果。

2.4 测试验证与迭代优化

测试验证与迭代优化是确保智能体落地效果的关键，需分阶段开展“功能-性能-场景”全维度测试。功能测试阶段，模拟真实场景下的各类输入，验证智能体的响应准确性、流程完整性，例如客服智能体测试不同咨询问题的解答准确率，工业智能体测试设备数据异常的识别精度，排查功能漏洞。性能测试阶段，重点验证响应速度、并发处理能力与稳定性，例如测试 100 人同时咨询时的响应延迟，连续运行 72 小时的稳定性，确保满足场景的实际使用需求。场景验证阶段，在真实环境中进行小范围试点，收集用户反馈与实际运行数据，针对性优化决策逻辑、指令匹配度等，例如根据用户反馈调整回答话术，根据设备运行数据优化故障预测模型，实现“测试-反馈-优化”的闭环迭代。

三、关键注意事项：保障搭建质量与落地效果

在智能体搭建全流程中，需重点关注数据安全、成本控制、人机协同三大核心问题，避免因细节疏漏导致搭建失败或落地效果不佳。

3.1 数据安全与合规管控

数据是智能体运行的基础，需全程保障数据安全与合规。一方面，严格遵循数据安全相关法规，例如个人信息保护法、数据安全法，确保数据采集、存储、传输、使用全流程合规，避免敏感信息泄露。另一方面，搭建数据安全防护体系，采用数据加密、访问权限管控、脱敏处理等技术，例如对用户隐私信息进行加密存储，对设备商业数据设置分级访问权限，定期开展数据安全审计，防范数据安全风险。

3.2 成本适配与轻量化部署

成本控制是智能体规模化落地的关键，尤其是中小企业需避免盲目投入。建议采用“轻量化起步、逐步迭代”的部署策略：初期优先搭建核心功能模块，选用低成本的云服务与开源框架，降低初期投入；根据业务发展需求逐步扩展功能，升级算力资源。同时，做好成本评估与优化，例如通过数据压缩减少存储成本，通过算力调度提升资源利用率，避免资源闲置。

3.3 人机协同边界的明确

智能体的核心价值是辅助人工而非替代人工，需明确人机协同的边界。在搭建过程中，需预设人工介入机制：对于超出智能体能力范围的复杂问题（如特殊业务咨询）、高风险决策（如重大设备停机指令），自动转接人工处理；同时，搭建人机协同平台，实现人工对智能体运行状态的监控、决策结果的审核与异常情况的干预，确保智能体的运行安全与效果可控。

四、智能体应用搭建 QA 问答

4.1 基础认知类

Q：什么是智能体应用搭建？核心逻辑是什么？

A：智能体应用搭建是指从 0 到 1 构建具备“感知-决策-执行”闭环能力的智能系统，能自主响应特定场景需求并完成任务。核心逻辑是以场景为锚点，围绕具体需求（如客服、工业控制）搭建“感知-决策-执行-优化”的完整能力闭环，而非单纯整合技术工具，最终实现从“被动响应”到“主动解决问题”的转变。

Q：搭建智能体应用需要哪些核心技术？

A：核心技术围绕“感知-决策-执行”三大模块展开：感知层需支持多类型信息接收（文本、语音、图像等），依赖 OCR、语音识别、多模态语义理解技术；决策层需结合规则引擎与大模型，处理标准化与非标准化场景；执行层需对接具体场景工具（如 CRM、工业系统），通过接口实现指令落地。

Q：搭建智能体的流程是怎样的？从 0 到 1 要分几步？

A：从 0 到 1 搭建智能体共分四步：第一步是需求拆解与场景定位，明确核心功能与流程；第二步是技术选型与框架选择，确定工具与部署方案；第三步是功能模块搭建，构建“感知-决策-执行”闭环；第四步是测试验证与迭代优化，通过测试提升稳定性与适配性。

4.2 技术选型类

Q：新手搭建智能体，优先选择哪些开源框架？

A：新手优先选择轻量化、易上手的开源框架：文本类场景（如客服智能体）选 LangChain，支持多工具联动与流程编排；轻量化自主决策场景选 AutoGPT，降低开发门槛；工业场景（如设备控制）可选用适配工业协议的专用框架（如基于 Python 的工业智能框架）。

Q：搭建智能体时，数据安全需要注意什么？

A：需重点关注三点：一是合规性，遵循《个人信息保护法》等法规，确保数据采集、存储、传输合规；二是防护措施，对敏感数据（如用户隐私）进行加密存储与脱敏处理，设置分级访问权限；三是定期审计，定期开展数据安全检查，避免数据泄露风险。

Q：如何控制智能体搭建的成本？

A：建议采用“轻量化起步、逐步迭代”策略：初期优先搭建核心功能模块，选用云服务器按需付费（如阿里云、腾讯云的轻量服务器），降低初期投入；后期根据业务需求逐步扩展功能，避免盲目升级算力；同时通过数据压缩、算力调度提升资源利用率，减少闲置成本。

4.3 实操落地类

Q：搭建智能体时，如何明确人机协同的边界？

A：需预设“智能体自主处理 + 人工介入”的双重机制：智能体负责标准化、低风险任务（如常规咨询、简单指令执行）；对于复杂问题（如特殊业务办理）、高风险决策（如设备停机），自动转接人工处理；同时搭建监控平台，人工可干预智能体的异常运行，确保安全可控。

Q：智能体搭建完成后，如何进行测试与优化？

A：分三阶段测试：一是功能测试，模拟真实场景输入，验证响应准确性与流程完整性；二是性能测试，测试响应速度、并发处理能力（如 100 人同时咨询）与稳定性；三是场景验证，在真实环境小范围试点，收集用户反馈，针对性优化决策逻辑与指令匹配度。

Q：不同场景（如客服、工业）搭建智能体，核心差异是什么？

A：核心差异在于场景需求与技术适配：客服场景需侧重“多模态交互 + 快速响应”，优先支持语音、文本等多类型输入；工业场景需侧重“设备数据采集 + 精准执行”，需对接工业系统（如 MES、PLC），确保指令与设备操作的精准匹配。

4.4 进阶优化类

Q：如何让智能体具备持续迭代能力？

A：需搭建“测试-反馈-优化”的闭环机制：在智能体中嵌入反馈收集模块，记录用户使用体验与执行结果；定期分析数据，调整决策逻辑与执行策略；同时预留扩展接口，支持后续功能升级与场景拓展，让智能体随业务需求持续优化。

Q：搭建智能体时，如何避免功能泛化？

A：核心是聚焦场景痛点：搭建前明确智能体的核心目标（如“7×24 小时客户咨询”），避免添加无关功能；在功能模块设计时，只保留与核心目标相关的能力，例如客服智能体无需添加复杂的数据分析功能，工业智能体无需支持多语言交互，确保能力与需求精准匹配。

Q：中小企业搭建智能体，有哪些低成本的实操建议？

A：一是选用轻量化工具，优先选择开源框架与云服务，降低开发与部署成本；二是小范围试点，先在单一场景（如客服咨询）落地，验证效果后再逐步扩展；三是借力第三方服务，部分平台提供智能体搭建的轻量化工具（如无需代码的可视化平台），降低技术门槛。

五、结语

从 0 到 1 搭建智能体应用，核心是把握“场景适配”与“闭环能力”两大核心要点，通过科学的流程拆解与严谨的细节管控，实现技术能力与业务需求的精准匹配。在 2026 AI 元年的技术浪潮下，智能体搭建不再是专业技术团队的专属，随着开源框架的普及与轻量化工具的推出，中小企业与个人也能实现低成本搭建。未来，随着技术的持续迭代，智能体搭建将更趋简化，但场景适配性、数据安全性与人机协同效率仍将是核心竞争力。唯有以场景为锚点，兼顾技术可行性与商业价值，才能让智能体真正发挥赋能作用，推动业务的智能化转型。

六、参考文献

[1] 中国信息通信研究院. 2026 人工智能产业发展白皮书[R]. 北京：中国信通院，2026.

[2] 工业和信息化部. 新一代人工智能发展规划（2024-2030 年）[Z]. 北京：工信部，2024.

[3] 佚名. 手把手用 LangChain 实现简易 AutoGPT[EB/OL]. CSDN 博客，2026-01-08.
https://blog.csdn.net/weixin\\_35756624/article/details/155976857.

[4] 佚名. 【Agent 智能体】开发流程与开源框架对比[EB/OL]. CSDN 博客，2026-01-28.
https://blog.csdn.net/weixin\\_44262492/article/details/155914728.

[5] 佚名. 03 | 原型系统：开源工具自建 AI 大模型底座[P]. 2024.

[6] 佚名. AutoGPT 进化实战:用 LangChain 从零打造你的自主 AI 代理[EB/OL]. CSDN 博客，2025-12-26.
https://blog.csdn.net/liu1983robin/article/details/145749760.

在构建问答Agent时，多轮对话的上下文记忆是核心需求——让Agent能记住历史对话内容，结合「历史问题+历史回答+当前问题」给出连贯回复，而非孤立回答每个问题。

LangChain中的ConversationBufferMemory是轻量、易用的短期记忆组件，核心作用是按顺序缓存对话历史，并将历史内容注入到模型的输入提示中，实现问答Agent的短期记忆能力，适合中小长度的多轮对话场景。

本文将基于LangChain框架，从核心原理、完整可运行代码、关键细节、进阶优化四个维度，教你为问答Agent集成ConversationBufferMemory，支持OpenAI/国产大模型（通义千问/文心一言），代码可直接复用。

一、核心概念铺垫

1.1 ConversationBufferMemory 核心作用

• 以键值对形式按时间顺序存储对话历史（问题+回答）；
• 支持将对话历史格式化为字符串/消息对象，注入到LLM的输入提示中；
• 提供清空记忆、获取记忆、修改记忆的便捷方法；
• 轻量无依赖，无需额外存储，对话历史保存在内存中（会话结束即销毁，符合「短期记忆」定位）。

1.2 核心搭配

ConversationBufferMemory通常与ConversationChain（通用对话链）/RetrievalQA（知识库问答链）搭配使用，本文先实现基础问答Agent（基于ConversationChain），后续补充带知识库的问答Agent优化方案。

1.3 关键参数

二、环境准备

安装LangChain核心依赖+大模型适配依赖（以OpenAI/通义千问为例，二选一即可）：

# 核心依赖：LangChain核心+社区组件
pip install langchain-core langchain-community -i https://pypi.tuna.tsinghua.edu.cn/simple

# 可选1：OpenAI模型依赖（GPT-3.5/GPT-4）
pip install langchain-openai -i https://pypi.tuna.tsinghua.edu.cn/simple

# 可选2：国产大模型依赖（通义千问/文心一言/智谱清言）
pip install langchain-qianfan langchain-dashscope -i https://pypi.tuna.tsinghua.edu.cn/simple

三、完整实现：基础问答Agent+短期记忆

3.1 方案1：基于OpenAI模型（GPT-3.5/GPT-4）

# 1. 导入核心模块
from langchain_openai import ChatOpenAI
from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferMemory
from langchain.prompts import PromptTemplate
import os

# 2. 配置环境（OpenAI API密钥）
# 国内用户需配置代理：os.environ["HTTP_PROXY"] = "http://127.0.0.1:7890"
os.environ["OPENAI_API_KEY"] = "你的OpenAI API密钥"

# 3. 初始化LLM模型
llm = ChatOpenAI(
    model="gpt-3.5-turbo",  # 推荐gpt-3.5-turbo，性价比高
    temperature=0.1,        # 越低回答越稳定，适合问答场景
    max_tokens=2048
)

# 4. 初始化ConversationBufferMemory（核心：短期记忆）
memory = ConversationBufferMemory(
    memory_key="chat_history",  # 记忆变量名，需与提示模板中的{chat_history}一致
    return_messages=False,      # 返回字符串格式的对话历史，适合基础场景
    input_key="input"           # 输入问题的变量名，默认input即可
)

# 5. 自定义带记忆的提示模板（必须包含{chat_history}和{input}）
# 模板说明：chat_history=历史对话，input=当前问题，让模型结合两者回答
prompt = PromptTemplate(
    input_variables=["chat_history", "input"],  # 必须包含记忆变量和输入变量
    template="""你是一个专业的问答助手，善于结合历史对话内容回答当前问题。
    历史对话：{chat_history}
    当前问题：{input}
    请简洁、准确地回答当前问题，无需额外赘述。"""
)

# 6. 构建带记忆的问答链（核心：将LLM、记忆、提示模板绑定）
conversation_chain = ConversationChain(
    llm=llm,
    memory=memory,
    prompt=prompt,
    verbose=True  # 开启详细日志，可查看输入的提示内容（含历史对话）
)

# 7. 测试多轮问答（验证记忆效果）
if __name__ == "__main__":
    # 第一轮问答
    print("===== 第一轮 =====")
    res1 = conversation_chain.invoke({"input": "什么是大语言模型？"})
    print("回答：", res1["output"], "\n")

    # 第二轮问答（结合历史：问大语言模型的核心优势）
    print("===== 第二轮 =====")
    res2 = conversation_chain.invoke({"input": "它的核心优势是什么？"})
    print("回答：", res2["output"], "\n")

    # 第三轮问答（结合历史：问该优势的应用场景）
    print("===== 第三轮 =====")
    res3 = conversation_chain.invoke({"input": "这些优势能用到哪些领域？"})
    print("回答：", res3["output"], "\n")

    # 手动查看记忆中的对话历史
    print("===== 查看短期记忆 =====")
    print(memory.load_memory_variables({}))

    # 清空记忆（可选）
    # memory.clear()
    # print("清空记忆后：", memory.load_memory_variables({}))

3.2 方案2：基于国产模型（通义千问，国内用户推荐）

替换上述步骤2和步骤3即可，其余代码完全不变，适配性拉满：

# 2. 配置环境（通义千问API密钥，从阿里云DashScope获取）
os.environ["DASHSCOPE_API_KEY"] = "你的通义千问API密钥"

# 3. 初始化通义千问模型（替换OpenAI）
from langchain_dashscope import ChatDashScope
llm = ChatDashScope(
    model="qwen-plus",  # 通义千问轻量版，免费额度足够测试
    temperature=0.1,
    max_tokens=2048
)

3.3 运行结果与关键日志

核心输出（记忆生效）

===== 第一轮 =====
回答： 大语言模型是基于大尺度语料训练、具备强大自然语言理解与生成能力的人工智能模型，能完成文本创作、问答、翻译等多种自然语言处理任务。

===== 第二轮 =====
回答： 大语言模型的核心优势包括：1. 强大的上下文理解与语义分析能力；2. 灵活的自然语言生成能力，可输出流畅、贴合语境的文本；3. 泛化能力强，能处理未见过的新问题；4. 多任务适配，无需单独训练即可完成多种NLP任务。

===== 第三轮 =====
回答： 这些优势可应用在智能客服、内容创作、教育辅导、代码开发、数据分析、机器翻译、智能助手等领域，覆盖互联网、教育、金融、制造业等多个行业。

===== 查看短期记忆 =====
{'chat_history': 'Human: 什么是大语言模型？\nAI: 大语言模型是基于大尺度语料训练、具备强大自然语言理解与生成能力的人工智能模型，能完成文本创作、问答、翻译等多种自然语言处理任务。\nHuman: 它的核心优势是什么？\nAI: 大语言模型的核心优势包括：1. 强大的上下文理解与语义分析能力；2. 灵活的自然语言生成能力，可输出流畅、贴合语境的文本；3. 泛化能力强，能处理未见过的新问题；4. 多任务适配，无需单独训练即可完成多种NLP任务。\nHuman: 这些优势能用到哪些领域？\nAI: 这些优势可应用在智能客服、内容创作、教育辅导、代码开发、数据分析、机器翻译、智能助手等领域，覆盖互联网、教育、金融、制造业等多个行业。'}

Verbose日志（关键：验证历史对话注入）

开启verbose=True后，可看到模型的实际输入提示包含了历史对话，这是记忆生效的核心：

> Entering new ConversationChain chain...
Prompt after formatting:
你是一个专业的问答助手，善于结合历史对话内容回答当前问题。
    历史对话：Human: 什么是大语言模型？
AI: 大语言模型是基于大尺度语料训练、具备强大自然语言理解与生成能力的人工智能模型，能完成文本创作、问答、翻译等多种自然语言处理任务。
    当前问题：它的核心优势是什么？
    请简洁、准确地回答当前问题，无需额外赘述。
> Finished chain.

四、关键细节：避免记忆失效的核心要点

ConversationBufferMemory使用简单，但容易因参数不匹配、提示模板错误导致记忆失效，以下是必须遵守的3条铁律：

4.1 提示模板必须包含memory_key指定的变量

比如memory_key="chat_history"，则提示模板中必须有{chat_history}，且输入变量列表要包含该变量：

# 正确：input_variables包含chat_history和input
prompt = PromptTemplate(
    input_variables=["chat_history", "input"],
    template="历史对话：{chat_history}  当前问题：{input}"
)

# 错误：缺少chat_history，记忆无法注入
prompt = PromptTemplate(
    input_variables=["input"],
    template="当前问题：{input}"
)

4.2 invoke入参必须是字典，且键为input_key

默认input_key="input"，因此调用时必须传{"input": "你的问题"}，而非直接传字符串：

# 正确
conversation_chain.invoke({"input": "什么是大语言模型？"})

# 错误：入参不是字典，记忆无法关联当前问题
conversation_chain.invoke("什么是大语言模型？")

4.3 避免手动修改对话历史（除非特殊需求）

ConversationBufferMemory会自动追加每次的input和output到记忆中，无需手动修改：

# 自动追加：无需干预
conversation_chain.invoke({"input": "问题1"})  # 记忆中添加问题1+回答1
conversation_chain.invoke({"input": "问题2"})  # 记忆中追加问题2+回答2

# 手动修改（特殊需求时使用）
memory.save_context(
    inputs={"input": "手动添加的问题"},
    outputs={"output": "手动添加的回答"}
)

五、进阶优化：适配更复杂的问答场景

5.1 优化1：带知识库的问答Agent+短期记忆

实际场景中，问答Agent通常需要结合私有知识库（如PDF/文档），此时将ConversationChain替换为RetrievalQA，并搭配ConversationBufferMemory即可实现「知识库+多轮记忆」的问答能力：

# 新增：导入知识库相关模块
from langchain_community.document_loaders import TextLoader
from langchain_community.vectorstores import FAISS
from langchain_text_splitters import CharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain.chains import RetrievalQA

# 1. 加载知识库（以文本文件为例，可替换为PDF/Word加载器）
loader = TextLoader("your_knowledge_base.txt")  # 你的知识库文件
documents = loader.load()
# 分割文本为小片段
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=100)
texts = text_splitter.split_documents(documents)
# 构建向量库
embeddings = OpenAIEmbeddings()
db = FAISS.from_documents(texts, embeddings)
retriever = db.as_retriever(search_kwargs={"k": 3})  # 每次检索3个相关片段

# 2. 初始化记忆（与基础版一致）
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=False,
    input_key="question"  # 注意：RetrievalQA的默认输入键是question，需修改
)

# 3. 构建带记忆的知识库问答链
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",  # 适合小片段文本，简单高效
    retriever=retriever,
    memory=memory,
    chain_type_kwargs={
        "prompt": PromptTemplate(
            input_variables=["chat_history", "context", "question"],
            template="""结合历史对话和知识库内容回答当前问题，若知识库无相关内容，仅结合历史对话回答。
            历史对话：{chat_history}
            知识库内容：{context}
            当前问题：{question}
            回答要求：简洁、准确，基于知识库内容，不要编造。"""
        )
    },
    verbose=True
)

# 4. 测试：结合知识库+历史对话的多轮问答
qa_chain.invoke({"question": "知识库中提到的大语言模型有哪些应用？"})
qa_chain.invoke({"question": "这些应用中，哪个在教育领域的落地效果最好？"})  # 结合历史

5.2 优化2：限制记忆长度（避免历史对话过长）

ConversationBufferMemory会无限制追加对话历史，当对话轮数过多时，会导致提示词过长、推理成本增加、模型注意力分散。

解决方案：使用ConversationBufferWindowMemory（窗口记忆），仅保留最近N轮对话，本质是ConversationBufferMemory的进阶版，参数完全兼容：

from langchain.memory import ConversationBufferWindowMemory

# 仅保留最近2轮对话，超出的自动丢弃
memory = ConversationBufferWindowMemory(
    memory_key="chat_history",
    k=2,  # 核心参数：保留最近k轮对话
    return_messages=False
)

# 用法与ConversationBufferMemory完全一致，无需修改其他代码
conversation_chain = ConversationChain(llm=llm, memory=memory, prompt=prompt)

5.3 优化3：记忆格式为消息对象（适合复杂提示）

当提示模板需要更精细的对话格式时，将return_messages=True，记忆会返回HumanMessage/AIMessage对象，而非拼接字符串，便于灵活格式化：

# 初始化记忆：返回消息对象
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True,  # 核心：返回消息对象
    input_key="input"
)

# 自定义提示模板：遍历消息对象格式化
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder

# 使用MessagesPlaceholder接收消息对象，无需手动拼接
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是专业的问答助手，结合历史对话回答问题。"),
    MessagesPlaceholder(variable_name="chat_history"),  # 匹配memory_key
    ("human", "{input}")  # 匹配input_key
])

# 构建链（用法不变）
conversation_chain = ConversationChain(llm=llm, memory=memory, prompt=prompt)

六、常用操作：记忆的增删改查

ConversationBufferMemory提供了便捷的方法操作记忆，满足个性化需求：

# 1. 查看记忆内容
memory.load_memory_variables({})  # 返回字典，键为memory_key

# 2. 清空记忆（会话结束/切换用户时使用）
memory.clear()

# 3. 手动添加记忆
memory.save_context(
    inputs={"input": "手动添加的问题"},
    outputs={"output": "手动添加的回答"}
)

# 4. 手动删除记忆（需先获取记忆，再修改，最后重新保存）
# 步骤1：获取记忆内容
chat_history = memory.load_memory_variables({})["chat_history"]
# 步骤2：修改/删除内容（如删除最后一行）
chat_history = "\n".join(chat_history.split("\n")[:-2])
# 步骤3：重新保存
memory.chat_memory.add_user_message("")  # 清空原有记忆
memory.chat_memory.add_ai_message("")
memory.save_context(inputs={"input": ""}, outputs={"output": chat_history})

七、总结

7.1 核心流程回顾

为问答Agent添加ConversationBufferMemory的核心步骤仅5步：

1. 安装LangChain核心依赖+大模型依赖；
2. 初始化LLM模型（OpenAI/国产模型）；
3. 初始化ConversationBufferMemory，指定memory_key；
4. 构建包含记忆变量的提示模板；
5. 将LLM、记忆、提示模板绑定到对话链，调用invoke实现多轮问答。

7.2 记忆组件选型建议

7.3 性能优化建议

1. 优先使用ConversationBufferWindowMemory，并设置合理的k值（如3-5轮）；
2. 降低LLM的max_tokens，避免无意义的长回答；
3. 开启verbose=False（生产环境），减少日志开销；
4. 生产环境中，可将记忆与会话ID绑定，实现多用户隔离。

八、常见问题排查

问题1：记忆失效，模型不结合历史对话回答

• 原因：提示模板缺少memory_key变量，或input_variables未包含该变量；
• 解决：检查提示模板，确保包含{chat_history}（或自定义的memory_key），且input_variables列表包含该变量。

问题2：调用时提示「key error: input」

• 原因：invoke入参不是字典，或键与input_key不匹配；
• 解决：调用时传{"input": "你的问题"}（默认input_key="input"），若修改了input_key，则传对应键。

问题3：知识库问答Agent记忆失效

• 原因：RetrievalQA的默认输入键是question，而非input，记忆的input_key不匹配；
• 解决：初始化记忆时设置input_key="question"。

问题4：对话历史过长，模型推理变慢

• 原因：ConversationBufferMemory无限制追加历史；
• 解决：替换为ConversationBufferWindowMemory，设置k值限制轮数。

大家好，我是R哥。

最近 Agent Skills 大火了，大家都在用 Skills 提升效率，很多 MCP 的功能已经被 Skills 代替了，MCP 已经失去了当初的光芒。

在《Claude Skills 彻底爆了，从实现原理到 Claude Code、CodeX、OpenCode 实战，一网打尽！》这篇文章中，我介绍了 Skills 的原理、创建及使用，今天我再来详细分享下市面上的 Skills。

现在市面上充斥着各种 Skills 导航站，GitHub 上各种 Skills 仓库也是层出不穷，功能也是五花八门。

最近我就在 GitHub 上发现了一个非常不错的 Skills 仓库：Awesome Claude Skills，地址如下：

https://github.com/ComposioHQ/awesome-claude-skills

目前该仓库已经狂收 26000+ 个 Star，收集了目前市面上主流的 Claude Skills，分类非常详细，涵盖了文档处理、开发与代码工具、数据与分析、商业与营销、沟通与写作、创意与媒体、效率与组织、协作与项目管理、安全与系统等多个方面。

我整理并翻译了该仓库的完整内容，大家可以收藏一下。

Awesome Claude Skills

文档处理

开发与代码工具

数据与分析

商业与营销

沟通与写作

创意与媒体

效率与组织

协作与项目管理

安全与系统

如何使用这些 Skills

在《Claude Skills 彻底爆了，从实现原理到 Claude Code、CodeX、OpenCode 实战，一网打尽！》这篇文章中，我介绍了 Skills 的原理、创建及使用。

下面我就拿其中一个 Skill 举例，看如何来使用这些 Skills。

1、首先将该对应的 Skill 下载到本地，比如我选择一个 git-pushing Skill。

2、把它复制到 ~/Users/John~/.claude/skills 目录下面。

3、这是一个提交 Git 的 Skill，所以，我们可以使用 commit and push 或者 "提交并推送" 等自然语言来自动调用这个 Skill。

如下图所示：

一句话就提交代码了，都不用自己想提交说明了，提交说明是和当前的变更文件是一致的，但是提交说明是英文的，有需要可以自己修改下 Skill 代码来实现中文提交说明。

总结

Skills 这波爆火，我的理解就一句话，把你每天重复干的事，做成可复用的动作，你不用每次从 0 写提示词，也不用记一堆参数，挑对 Skill 直接开干，省心还稳定。

这篇我把 Awesome Claude Skills 这个仓库过了一遍，真要落地，别一上来就装一堆，可能大部分你都用不到，不然也会影响加载效率。。

建议先从最刚需的 1 - 2 个场景开始，先跑通一次 “输入 - 处理 - 输出” 的闭环，再把你常用的操作固化成自己的 Skill 和话术。

这样用下来，效率提升会很实在，而且越用越顺。

未完待续，R哥持续分享更多 AI 编程经验，包括更加复杂的 Skills 使用，关注我一起学 AI。

⚠️ 版权声明：

本文系公众号 "AI技术宅" 原创，转载、引用本文内容请注明出处，抄袭、洗稿一律投诉侵权，后果自负，并保留追究其法律责任的权利。

在 AI 技术日新月异的今天，AI Agent（智能体）正逐渐从概念走向落地。它不仅能进行对话，更具备了思考、规划和执行任务的能力。然而，构建一个成熟的 Agent 系统，并非简单的 API 调用，而是多种核心技术协同工作的结果。

在深入开发之前，理清这些基础概念，有助于我们更好地理解 AI 系统的底层运行逻辑。

一、 智能的内核：大语言模型与交互边界

1. LLM（大语言模型）：通识大脑

LLM 是 Agent 的核心引擎。它拥有强大的语言理解能力，但它是一个“静态大脑”，其知识停留在训练截止的那一刻，无法感知企业内部的私有数据。

2. Context Window（上下文窗口）：短期记忆

这是模型单次交互能处理的信息上限。

• 局限： 即使窗口再大，也不能盲目塞入所有数据。正如在数学题中加入无关的干扰信息会降低准确率一样，过长的背景会导致模型“注意力不集中”，甚至产生幻觉。

3. Prompt Engineering（提示工程）：沟通的艺术

• Zero-shot（零样本）： 不给示例，直接下指令。这要求指令必须高度具体（如：从“写个政策”优化为“写个 200 字符合 GDPR 标准的隐私政策”）。
• Few-shot（少样本）： 提供几个理想的问答示例，这能有效地规范 AI 输出的语气（Tone）和特定格式。
• Chain of Thought（思维链）： 引导 AI 展示推理步骤，强制模型分配更多计算资源在逻辑推导上，从而处理复杂问题。

二、 知识的扩展：从“翻书”到“记忆”

为了让 AI 访问私有数据，我们需要构建一套“外挂硬盘”。

4. 向量数据库 vs 传统数据库

传统的 SQL 数据库是基于值或关键词的匹配（如 LIKE %vacation%）。而向量数据库（如 ChromaDB, Pinecone）则是基于含义（Meaning）的匹配。即使搜索词不一致，只要语义接近，系统就能精准定位。

5. Embeddings 与数据预处理

• 数据切分（Chunking）： 我们不能将 500GB 的文档直接塞给 AI。必须将其切成小块。
• 重叠（Overlap）： 在切分时，通常会保留一定的文字重叠。这能防止上下文在切分处丢失，从而大幅提升检索的准确性。
• Embeddings： 将切分好的文本块转化为高维数字向量，让计算机能够以数学方式计算语义的相关性。

6. RAG（检索增强生成）：知识的补丁

RAG 是目前解决 AI 幻觉的最优方案。它通过“检索 -> 增强 -> 生成”的流程，让 AI 像是在参加开卷考试：先去数据库里“翻书”找到事实，再根据事实组织答案。

三、 行动的逻辑：框架、编排与协议

7. LangChain：开发的“胶水”层

LangChain 是一个强大的抽象层，旨在简化开发流程。

• 核心价值： 它像管道一样将模型、提示词模板和向量库连接起来。有了它，你从 OpenAI 切换到 Google Gemini 可能只需要更改一行代码，极大地提高了系统的灵活性。

8. LangGraph：有状态的“总导演”

当任务需要循环和决策时，简单的线性管道就不够用了。

• 节点与边： LangGraph 通过节点（步骤）和边（路径）构建工作流。
• 共享状态（State）： 这是它的核心。它维护着一个在各节点间传递的“字典”，记录着当前的文档、评分等信息。基于这个状态，系统可以执行复杂逻辑：例如“如果合规分数低于 75 分，则循环回退到搜索节点重新查阅”。

9. MCP（模型上下文协议）：标准化的“USB 接口”

这是连接外部工具（如 GitHub、数据库）的通用标准。它让 AI 具备了“即插即用”的能力，开发者无需为每个工具编写特定的硬编码集成，只需符合 MCP 协议，Agent 就能自主调用。

四、 总结：各组件是如何协同工作的？

构建一个完整的 AI 系统，本质上是让这些组件各司其职、形成闭环：

1. 准备： 文档经过切分与重叠处理，通过 Embeddings 存入向量数据库。
2. 触发： 用户提问，LangChain 调度 RAG 流程，根据语义意图找回知识。
3. 决策： LangGraph 根据当前状态判断：是直接回答，还是需要循环修正？
4. 执行： 如果需要实时数据，通过 MCP 协议调用外部工具。
5. 产出： LLM 结合所有事实与逻辑推理，输出最终方案。

理清了这些基石，你就已经掌握了从“对话机器人”跨越到“全能 Agent”的底层蓝图。

本文由mdnice多平台发布

使用 django 的 command 启动 langGraph Server, 但要求基于 AsyncPostgresSaver,
没有找到相关的可用的代码, 这里记录下 直接抛出代码

"""
Run the LangGraph Agent Server
"""

import asyncio
import uvicorn
from django.core.management.base import BaseCommand
from django.conf import settings
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from langserve import add_routes
from langchain_core.globals import set_verbose
from psycopg_pool import AsyncConnectionPool
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from apps.ai.graph.app import AsyncGraphApp


class Command(BaseCommand):
    """
    Run the LangGraph Agent Server
    """

    help = "Starts the LangGraph Agent Server"

    def add_arguments(self, parser):
        parser.add_argument("--host", type=str, default="0.0.0.0")
        parser.add_argument("--port", type=int, default=2028)

    def handle(self, *args, **options):
        asyncio.run(self.handle_async(*args, **options))

    async def handle_async(self, *args, **options):
        """启动 Agent Server"""
        host = options["host"]
        port = options["port"]

        self.stdout.write(f"Starting Agent Server at http://{host}:{port}...")

        # Get the LangGraph application
        checkpointer = await self.get_checkpointer()
        graph_app = AsyncGraphApp().compile(checkpointer=checkpointer).app
        print(f"graph_app----------------->: {graph_app}")

        # Initialize FastAPI app
        app = FastAPI(
            title="Baby Consultant Agent",
            version="1.0",
            description="A LangGraph-based agent for baby consultation",
        )

        # Set CORS
        app.add_middleware(
            CORSMiddleware,
            allow_origins=["*"],
            allow_credentials=True,
            allow_methods=["*"],
            allow_headers=["*"],
        )
        set_verbose(True)
        # Add routes using LangServe
        # This exposes the graph at /agent/invoke, /agent/stream, etc.
        add_routes(
            app,
            graph_app,
            path="/agent",
        )

        # Run with Uvicorn
        config = uvicorn.Config(app, host=host, port=port)
        # 基于当前的Async Running Loop 启动unicorn
        server = uvicorn.Server(config)
        await server.serve()

    async def get_checkpointer(self):
        """获取 Checkpointer"""
        # 1. 显式创建连接池 (让它在应用生命周期内一直存活)
        connection_kwargs = {
            "autocommit": True,
            "prepare_threshold": 0,
        }

        # 使用同步的 ConnectionPool
        pool = AsyncConnectionPool(
            conninfo=settings.LANGGRAPH_POSTGRES_CONNECTION_STRING,
            max_size=20,
            kwargs=connection_kwargs,
        )

        # 2. 将连接池传入构造函数
        checkpointer = AsyncPostgresSaver(pool)

        # 3. 初始化数据库表
        await checkpointer.setup()

        return checkpointer

一、为什么智能体正在成为 AI 应用的主流形态

随着大模型能力逐步稳定、推理成本不断下降，AI 应用正在从“一次性对话工具”演进为“可持续运行的系统”。
智能体（AI Agent）正是这种系统化 AI 应用的典型代表。

与传统大模型应用不同，智能体具备以下特征：

• 目标驱动
• 任务规划
• 工具调用
• 反馈修正
• 持续执行

因此，智能体逐渐成为大模型应用落地的主流形态。

但在真实项目中，开发者面临的第一个问题往往不是模型选型，而是：

从 0 到 1，智能体到底该怎么搭？

二、三种智能体实现路径的核心差异

在工程实践中，智能体的实现方式主要分为三类：零代码、低代码、全代码。

1. 零代码方案

• 适合人群：非技术人员、产品、个人用户
• 技术门槛：无编程基础
• 开发周期：小时级
• 适用场景：原型验证、个人提效、轻量自动化

2. 低代码方案

• 适合人群：具备基础编程能力的开发者
• 技术门槛：Python 基础
• 开发周期：1–3 天
• 适用场景：部门级工具、业务辅助系统

3. 全代码方案

• 适合人群：技术团队、中大型企业
• 技术门槛：完整工程能力
• 开发周期：7–15 天
• 适用场景：核心业务系统、多智能体协作

经验结论：路径选错，后期几乎一定重构。

三、零代码方案：最快验证智能体价值（MVP）

零代码方案的核心价值在于速度，适合在需求尚未明确时快速验证价值。

常见应用场景：

• 文档整理
• 竞品监测
• 简单客服
• 自动生成报告

局限性：

• 不适合敏感数据
• 难以对接复杂系统
• 扩展能力有限

四、低代码方案：最推荐的工程化落地路径

低代码方案是大多数成功项目的选择，尤其适合个人开发者和中小企业。

常见技术组合包括：

• Agent 框架（LangChain、LangGraph）
• 大模型 API（如通义千问）
• 向量数据库
• 轻量服务部署

优势：

• 成本可控
• 易迭代
• 可上线生产
• 易维护

五、全代码方案：企业级智能体的标准形态

当智能体进入核心业务流程，必须采用全代码方案。

典型系统通常包含五个模块：

1. 感知模块：统一接入数据
2. 规划模块：拆解任务步骤
3. 执行模块：调用工具与系统
4. 记忆模块：保存历史经验
5. 反馈模块：持续优化结果

该方案适用于高安全、高复杂度场景。

六、推荐的实践顺序（非常重要）

工程实践中，大多数成功项目遵循以下顺序：

1. 零代码验证价值
2. 低代码跑通业务
3. 全代码系统化建设

直接跳过前两步，失败率极高。

七、总结

智能体落地问题，本质是系统工程问题，而不是模型问题。

一个成功的智能体系统，必须能够：

• 跑通闭环
• 可持续运行
• 支持扩展
• 可监控维护
• 融入现有业务系统

从 0 到 1，选对实现路径，比选技术更重要。

摘要

插件作为大模型与智能体突破原生能力边界、实现场景化功能落地的核心载体，是从 “通用 AI 能力” 到 “行业专属解决方案” 的关键桥梁。本文从插件的核心定义与价值出发，系统拆解大模型 / 智能体插件的底层工作逻辑，梳理从 0 到 1 的插件认知、选型、使用、定制全流程，详解不同场景下的插件搭配技巧与避坑指南，同时结合行业实操案例给出落地建议，并补充高频 QA 问答解决入门核心痛点，帮助零基础从业者快速掌握插件使用逻辑，实现大模型与智能体的能力最大化延伸，玩转插件生态的核心玩法。​关键词​：插件；大模型插件；智能体插件；AI 工具使用；从 0 到 1 学插件；插件定制；AI 能力延伸；智能体生态

一、插件的核心认知：大模型与智能体的 “能力扩展卡”

1.1 插件的定义与核心价值

插件是为大模型、智能体量身打造的​模块化功能扩展组件​，通过标准化接口与大模型 / 智能体核心系统对接，无需改变底层模型架构，即可快速为其新增专属功能、接入外部数据、实现跨平台联动。简单来说，大模型 / 智能体的原生能力是 “通用基础款”，而插件就是 “个性化拓展包”，让 AI 从 “能说会想” 升级为 “能做会干”。

其核心价值体现在三大维度：

• ​突破能力边界​：弥补大模型 “知识滞后、计算薄弱、无实操能力” 的短板，比如通过计算器插件解决数学运算、通过翻译插件实现多语种精准转换、通过数据分析插件完成数据可视化；
• ​适配场景落地​：针对办公、学习、研发、电商等不同场景，提供定制化功能，让通用 AI 适配专属需求，比如自媒体从业者用排版插件、程序员用代码调试插件、运营者用数据统计插件；
• ​降低使用门槛​：无需掌握 AI 开发技术，普通用户通过一键安装插件，即可让大模型 / 智能体具备专业能力，实现 “零代码玩转高阶 AI”。

1.2 插件与大模型、智能体的底层工作逻辑

插件与大模型 / 智能体的协作遵循 **“调用 - 执行 - 反馈”** 的闭环逻辑，核心分为三步，零基础也能轻松理解：

1. ​需求识别​：用户向大模型 / 智能体发出指令后，其核心系统先判断原生能力是否能满足，若无法满足则自动匹配已安装的对应插件；
2. ​插件调用​：核心系统通过标准化接口向插件发送执行指令，插件承接需求后完成专属处理（如数据计算、外部查询、功能执行）；
3. ​结果反馈​：插件将处理结果回传给核心系统，由大模型 / 智能体整理成自然语言或可视化结果，反馈给用户。

整个过程毫秒级完成，用户感知不到底层调用逻辑，仅需发出自然语言指令，即可实现插件功能的无缝使用，这也是插件能快速普及的核心原因。

1.3 插件的核心分类：按功能与使用场景划分

目前主流的大模型 / 智能体插件生态，按功能属性可分为 6 大类，覆盖绝大多数日常与工作场景，零基础入门可先从高频通用类开始掌握：

二、从 0 到 1：插件使用全流程，新手也能一步到位

2.1 第一步：选对平台 —— 插件生态的核心载体

插件的使用依赖于支持插件功能的大模型 / 智能体平台，零基础入门优先选择插件生态完善、操作门槛低、免费插件多的主流平台，避免因平台小众导致插件资源少、使用难度高，以下是目前最适合新手的三大主流平台，各有优势：

1. ​通用大模型平台​：ChatGPT（4o 及以上版本）、文心一言 4.0、讯飞星火 V4.0，插件生态完善，覆盖全品类插件，操作界面简洁，一键安装即可使用，适合全场景需求；
2. ​智能体专属平台​：Coze、LangChain Bot，主打智能体插件联动，支持插件自定义编排，适合需要多插件协同完成复杂任务的场景；
3. ​办公类 AI 平台​：WPS AI、飞书智谱，插件与办公软件深度融合，主打办公场景专属插件，适合职场办公人群。

​新手建议​：优先从 ChatGPT 或文心一言入手，插件资源最丰富，操作最简洁，能快速完成从 0 到 1 的插件使用入门。

2.2 第二步：插件选型 —— 按需选择，拒绝盲目安装

插件并非越多越好，盲目安装大量插件会导致大模型 / 智能体响应变慢、匹配插件出错，零基础入门的核心原则是 **“刚需优先、少而精”**，按 “场景 - 需求 - 插件” 的逻辑选型，具体步骤：

1. ​明确使用场景​：确定自己使用大模型 / 智能体的核心场景，比如是日常办公、自媒体创作，还是编程开发；
2. ​梳理核心需求​：从场景中提炼需要解决的具体问题，比如办公场景需要 “PDF 解析、表格制作”，创作场景需要 “文案润色、图文排版”；
3. ​匹配对应插件​：根据需求选择功能精准的插件，比如 PDF 解析选专属 OCR 插件，文案润色选内容创作类插件，避免一个需求安装多个同类插件。

​新手避坑​：同一功能的插件只需安装 1-2 个即可，比如翻译插件无需同时安装百度翻译、谷歌翻译、DeepL 翻译，选择适配自己使用习惯的一款即可。

2.3 第三步：基础使用 —— 一键安装，三步玩转核心功能

主流平台的插件操作均实现​可视化、零代码​，零基础用户无需掌握任何技术，只需三步即可完成插件的安装与使用，以通用大模型平台为例，操作流程高度统一：

1. ​插件市场入口​：登录平台后，在侧边栏或设置中找到「插件市场 / 应用中心」，这是所有插件的集中入口；
2. ​一键安装插件​：在插件市场中搜索需要的插件，点击「安装 / 启用」，平台会自动完成接口对接，安装完成后插件会出现在「已安装插件」列表中；
3. ​自然语言调用​：无需额外操作，直接向大模型 / 智能体发出自然语言指令，系统会自动匹配插件执行，比如安装了计算器插件后，直接说 “计算 10000 元按年化 3.5% 计息，存 5 年的复利是多少”，系统会自动调用插件计算并给出结果。

​小技巧​：若系统未自动匹配插件，可在指令中明确提及插件名称，比如 “用思维导图插件把《从 0 到 1 玩转插件》的核心框架做成思维导图”，提升插件调用精准度。

2.4 第四步：进阶搭配 —— 多插件协同，实现复杂任务落地

当掌握单一插件的使用后，可通过​多插件协同搭配​，让大模型 / 智能体完成更复杂的场景化任务，这是 “玩转插件” 的核心进阶技巧。多插件搭配的核心逻辑是 **“按任务流程拆解，依次匹配插件”**，举 3 个高频场景的经典搭配案例，新手可直接照搬：

案例 1：办公场景 —— 快速完成一份市场分析报告

​任务流程​：解析市场调研 PDF 数据 → 整理成表格 → 进行数据可视化 → 生成分析报告​插件搭配​：PDF 解析插件（OCR）+ 表格分析插件 + 数据可视化插件 + 文案创作插件​使用指令​：“用 PDF 解析插件提取这份市场调研文件的核心数据，用表格分析插件整理成销售数据表格，再用数据可视化插件生成柱状图，最后用文案创作插件基于数据生成一份 500 字的市场分析报告”

案例 2：创作场景 —— 打造一篇自媒体爆款推文

​任务流程​：生成推文选题 → 撰写推文文案 → 优化排版 → 生成配图思路​插件搭配​：选题生成插件 + 文案润色插件 + 图文排版插件 + 创意设计插件​使用指令​：“用选题生成插件给美妆品类生成 3 个小红书爆款选题，选其中一个用文案润色插件撰写 800 字推文，用图文排版插件优化排版格式，最后用创意设计插件给出推文配图思路”

案例 3：研发场景 —— 快速调试一段 Python 代码

​任务流程​：检查代码漏洞 → 修复 Bug→ 解释代码逻辑 → 生成注释​插件搭配​：代码检测插件 + Bug 修复插件 + 代码解释插件 + 注释生成插件​使用指令​：“用代码检测插件检查这段 Python 代码的漏洞，用 Bug 修复插件修正错误，用代码解释插件逐行说明逻辑，最后用注释生成插件为代码添加标准注释”

2.5 第五步：高阶定制 —— 打造专属插件，适配个性化需求

当现有插件无法满足专属需求时，可尝试​插件定制​，目前主流平台均提供​低代码 / 零代码插件定制工具​，零基础用户也能从 0 到 1 打造自己的专属插件，核心流程分为 4 步，无需掌握复杂开发技术：

1. ​明确定制需求​：确定专属插件的核心功能、使用场景、输入输出要求，比如定制一款 “电商商品标题优化插件”，核心功能是根据商品属性生成高点击率标题；
2. ​选择定制平台​：优先选择平台自带的插件定制工具，如 ChatGPT 的 Plugin Builder、文心一言的插件开发平台，无需对接复杂接口，可视化操作；
3. ​配置插件功能​：在定制工具中，通过拖拽、选择、填写参数的方式，配置插件的核心功能，比如为电商标题插件设置 “商品属性输入框、标题风格选择（简约 / 爆款 / 专业）、标题字数限制” 等；
4. ​测试与发布​：完成配置后，进行多次测试，验证插件功能是否符合预期，测试通过后即可发布到自己的插件列表，实现专属使用，部分平台还支持将定制插件分享到插件市场。

​新手建议​：入门阶段先从简单功能插件开始定制，比如 “专属话术生成插件”“日常打卡插件”，熟悉定制逻辑后，再尝试复杂功能插件。

三、插件使用的核心避坑指南：新手少走 90% 的弯路

从 0 到 1 玩转插件，不仅要会用，更要会​避坑​，结合大量新手实操案例，梳理出 6 个最易踩的坑，以及对应的解决方案，帮新手快速避开误区：

3.1 坑 1：盲目安装大量插件，导致系统响应变慢

​问题​：认为插件越多功能越全，安装数十个同类插件，导致大模型 / 智能体匹配插件时耗时增加，响应变慢，甚至出现插件冲突。​解决方案​：遵循 “​刚需安装、定期清理​” 原则，仅安装当前场景需要的插件，每 1-2 周清理一次未使用的插件，保持已安装插件列表简洁。

3.2 坑 2：忽略插件权限，导致数据安全风险

​问题​：安装插件时随意授权，部分插件会请求访问用户的聊天记录、上传文件、个人数据等权限，导致数据泄露风险。​解决方案​：安装插件前​必看权限说明​，拒绝授权与插件功能无关的权限，比如一款计算器插件若请求访问聊天记录，直接拒绝安装；优先选择平台官方开发的插件，第三方插件需确认资质后再安装。

3.3 坑 3：指令描述模糊，导致插件调用失败

​问题​：向大模型 / 智能体发出的指令过于模糊，系统无法准确匹配插件，比如只说 “帮我处理这份数据”，未说明具体处理需求。​解决方案​：指令描述遵循 **“场景 + 需求 + 插件”** 的三要素原则，明确告知系统要做什么、用什么插件做，比如 “在电商场景下，帮我用选品插件分析抖音美妆类目的爆款商品数据”。

3.4 坑 4：过度依赖插件，忽视大模型原生能力

​问题​：任何需求都想通过插件解决，即使大模型原生能力能轻松完成，比如用翻译插件翻译简单的日常语句，反而增加操作成本。​解决方案​：先判断​需求是否需要插件​，大模型原生的自然语言理解、文案创作、逻辑分析等能力能解决的问题，无需调用插件，让插件成为 “补充能力” 而非 “唯一能力”。

3.5 坑 5：未及时更新插件，导致功能失效

​问题​：插件安装后长期不更新，当大模型 / 智能体平台升级或插件底层功能调整时，出现插件调用失败、功能失效的问题。​解决方案​：开启插件的​自动更新功能​，或定期在插件市场检查已安装插件的更新状态，及时更新至最新版本，保证插件功能正常使用。

3.6 坑 6：多插件搭配逻辑混乱，导致任务执行出错

​问题​：多插件协同时，未按任务流程合理搭配，插件顺序混乱，导致系统无法按预期执行，比如先让数据可视化插件生成图表，再让 PDF 解析插件提取数据。​解决方案​：多插件搭配前，先​梳理任务的先后流程​，按 “先输入、再处理、最后输出” 的逻辑匹配插件，确保插件调用顺序与任务流程一致，避免逻辑混乱。

四、插件生态的未来发展趋势：大模型与智能体的核心竞争力

插件作为大模型与智能体实现 **“能力落地、生态繁荣”** 的核心载体，其发展趋势与大模型、智能体的技术迭代深度绑定，未来三大发展方向值得关注，也是新手玩转插件需要提前布局的重点：

4.1 插件轻量化：零代码定制成为主流，全民可造插件

未来插件的开发门槛将持续降低，低代码 / 零代码定制工具将成为行业标配，不仅专业开发者能打造插件，普通用户也能通过简单的参数配置、功能拖拽，打造自己的专属插件，实现 “全民造插件” 的生态格局，插件将从 “专业产品” 变为 “个人化工具”。

4.2 插件智能化：智能体自主完成插件编排与适配

大模型与智能体的能力持续升级，未来将具备​自主插件编排能力​—— 用户只需发出核心需求，无需指定插件，智能体就能根据任务逻辑，自主选择、搭配、调用插件，完成复杂任务。比如用户说 “帮我完成一份年度销售总结”，智能体将自主匹配数据提取、表格分析、可视化、文案创作等插件，全程无需人工干预。

4.3 插件生态化：跨平台插件互联互通，形成全域能力网络

不同大模型、智能体平台的插件生态将从 “孤立发展” 走向 “互联互通”，通过标准化的插件接口，实现跨平台插件的无缝调用，比如在智能体平台可直接调用大模型的办公插件，在办公软件中可直接调用智能体的行业插件，形成覆盖全场景、跨平台的​插件能力网络​，让 AI 的能力延伸到每一个工作与生活场景。

五、行业高频 QA 问答

5.1 零基础新手，先从哪类插件开始学习使用最合适？

优先从通用工具类插件入手，比如计算器、翻译、PDF 解析、思维导图插件。这类插件功能简单、使用频率高、适配全场景，无需专业知识就能快速上手，能帮助新手快速熟悉插件的安装、调用逻辑，建立使用信心，掌握后再逐步拓展到内容创作、数据处理等专项插件。

5.2 免费插件和付费插件的区别是什么，新手需要付费购买插件吗？

核心区别在于​功能精准度、使用限制、专属服务​：免费插件能满足基础需求，部分存在使用次数、功能简化的限制；付费插件功能更精准、无使用限制，部分还提供专属售后与定制化优化。新手​无需过早付费​，目前主流平台的免费插件已能覆盖 90% 的日常与工作需求，建议先通过免费插件掌握使用逻辑，当现有免费插件无法满足核心工作需求时，再针对性购买付费插件。

5.3 不同大模型 / 智能体平台的插件可以互通使用吗？

目前多数平台的插件​暂不支持直接互通​，因各平台的插件接口标准、底层逻辑存在差异，比如 ChatGPT 的插件无法直接在文心一言中使用。但未来随着行业标准化推进，跨平台插件互联互通将成为趋势，现阶段若需要在多个平台使用同类功能，可在各平台分别安装对应的同款或同类插件。

5.4 安装插件后，大模型 / 智能体的响应速度变慢，该怎么解决？

可按以下步骤逐一排查解决：1. 清理未使用的插件，卸载同类冗余插件，减少插件匹配压力；2. 检查插件是否为最新版本，及时更新失效插件；3. 若使用多插件协同，尝试拆分任务，单次仅调用 1-2 个插件，避免同时调用大量插件；4. 关闭平台后台无关程序，保证网络通畅，提升接口传输速度。

5.5 如何判断一款插件是否适合自己，有没有核心筛选标准？

核心筛选标准有 4 点：1. ​功能匹配​：插件核心功能与自己的核心需求高度契合，无多余无效功能；2. ​操作简单​：零基础能快速上手，无需复杂的参数配置；3. ​权限安全​：插件请求的权限与功能匹配，无过度授权；4. ​口碑良好​：在插件市场中评分高、评论正面，官方开发或第三方资质可靠，避免使用小众无资质插件。

5.6 新手可以尝试开发插件吗，需要掌握哪些基础技能？

新手可以尝试，目前主流平台的​低代码 / 零代码定制工具​，让零基础用户也能开发简单插件，无需掌握复杂的编程技术。若仅做基础定制，只需掌握​需求梳理能力​，能明确插件的功能、使用场景即可；若想开发更复杂的插件，可逐步学习简单的编程基础（如 Python）、API 接口知识，提升定制能力。

六、结论

从 0 到 1 玩转插件，本质是掌握​大模型与智能体的能力延伸逻辑​—— 插件并非简单的 “功能工具”，而是让通用 AI 技术落地到具体场景的核心桥梁。对于零基础从业者而言，无需畏惧技术门槛，从核心认知入手，按 “选型 - 安装 - 使用 - 搭配 - 定制” 的全流程逐步推进，避开盲目安装、权限泄露、指令模糊等核心误区，就能快速掌握插件的核心玩法。

插件生态的发展，正推动大模型与智能体从 “通用能力” 向 “个性化、场景化能力” 升级，未来随着插件的轻量化、智能化、生态化发展，插件将成为每一个 AI 使用者的必备工具。新手只需从当下开始，从一款插件、一个场景入手，边用边学、边练边进阶，就能在插件生态中找到适合自己的使用方法，真正实现 “玩转插件”，让大模型与智能体成为工作与生活的高效助手，释放 AI 的最大价值。

参考文献

[1] 斯坦福大学. AI 指数报告 2026 [R]. 斯坦福大学人类与人工智能研究院，2026.
[2] 中国人工智能产业发展联盟。大模型插件生态建设与应用指南 2026 [R]. 2026.
[3] 字节跳动 AI 实验室. Coze 智能体插件平台开发与使用手册 2026 [R]. 2026.
[4] OpenAI 官方文档. ChatGPT Plugin 开发与使用指南 [Z]. 2026.
[5] 百度 AI 研究院。文心一言插件生态与场景落地实践 2026 [R]. 2026.
[6] 知乎科技研究院. 2026 年 AI 插件使用行为分析报告 [R]. 2026.

作者：江昱

阿里云函数计算 AgentRun 全新发布后，我们整理了“探秘 AgentRun”系列文章，本系列将梳理企业落地 Agent 常见难题，给出具体解法，助力 Agentic AI 快速走进生产级环境。欢迎加入“函数计算 AgentRun 客户群”与我们交流，钉钉群号： 134570017218 。

当你已经用 LangChain、AgentScope、LangGraph 等框架开发了 Agent 应用，如何让它们享受函数计算 AgentRun 提供的 Serverless 运行时、企业级 Sandbox、模型高可用、全链路可观测等能力？好消息是，你几乎不需要改动现有代码，只需要简单的适配就可以迁移到函数计算 AgentRun。

这篇文章将通过真实的代码示例，展示如何将不同框架的 Agent 应用部署到函数计算 AgentRun 上，以及如何充分利用函数计算 AgentRun 的各种能力。

为什么要部署到函数计算 AgentRun？

在讨论具体的集成方案前，让我们先明确一个问题：如果你的 Agent 应用已经在本地或自建服务器上运行良好，为什么还要迁移到函数计算 AgentRun？

答案很简单：从开发环境到生产环境，有一道巨大的鸿沟。  本地运行只需要考虑功能实现，但生产环境需要考虑性能、稳定性、成本、安全、可观测等一系列问题。函数计算 AgentRun 提供的不是又一个 Agent 框架，而是让你的 Agent 能够以企业级标准运行的完整基础设施。

具体来说，部署到函数计算 AgentRun 后，你能获得：零运维的 Serverless 运行时（自动扩缩容、按量付费），企业级的 Sandbox 环境（高性能、安全隔离），模型高可用保障（自动熔断、多模型 Fallback），全链路可观测（完整的 Trace、成本归因），以及统一的工具和 MCP 管理。

快速上手：5 分钟部署你的第一个 LangChain Agent

让我们从最流行的 LangChain 框架开始，通过一个完整的例子展示如何将 LangChain Agent 部署到函数计算 AgentRun。

第一步：安装 Serverless Devs

函数计算 AgentRun 使用 Serverless Devs 作为部署工具。如果你有 Node.js 环境，一行命令即可安装：

npm i -g @serverless-devs/s

第二步：创建项目

使用脚手架快速创建项目（注意：需要 Python 3.10 及以上版本）：

# 初始化模板
s init agentrun-quick-start-langchain
# 进入代码目录
cd agentrun-quick-start-langchain/code
# 初始化虚拟环境并安装依赖
uv venv && uv pip install -r requirements.txt

第三步：配置认证信息

通过环境变量（建议使用 .env 文件）配置你的 AgentRun 访问凭证：

export AGENTRUN_ACCESS_KEY_ID="your-access-key-id"
export AGENTRUN_ACCESS_KEY_SECRET="your-access-key-secret"
export AGENTRUN_ACCOUNT_ID="your-account-id"
export AGENTRUN_REGION="cn-hangzhou"

第四步：理解集成方式

这是最关键的部分。打开生成的代码，你会看到集成非常简单：

from agentrun.integration.langchain import model, sandbox_toolset
from agentrun.server import AgentRunServer
# 使用 AgentRun 的模型（自动享受高可用、熔断等能力）
llm = model("<your-model-name>")
# 使用 AgentRun 的 Sandbox 工具
tools = sandbox_toolset(
    template_name="<your-sandbox-name>",
    template_type=TemplateType.CODE_INTERPRETER,
    sandbox_idle_timeout_seconds=300,
)
# 创建 LangChain Agent（和原来的代码完全一样）
agent = create_agent(
    model=llm,
    tools=tools,
    system_prompt="你是一个智能助手"
)
# 定义调用函数
def invoke_agent(request):
    result = agent.invoke({"messages": request.messages})
    return result["messages"][-1].content
# 启动 HTTP Server（提供 OpenAI 兼容的 API）
AgentRunServer(invoke_agent=invoke_agent).start()

核心要点：

• model() 函数返回的是 LangChain 可以直接使用的模型对象
• sandbox_toolset() 返回的是 LangChain Tools 列表
• 你的 Agent 创建代码完全不需要改动
• AgentRunServer 自动处理 HTTP 请求，提供标准的 OpenAI API

第五步：本地测试

启动服务后，可以通过 HTTP 请求测试：

curl 127.0.0.1:9000/v1/chat/completions \
  -X POST \
  -H "content-type: application/json" \
  -d '{"messages": [{"role": "user", "content": "通过代码查询现在是几点?"}], "stream":true}'

第六步：部署到生产环境

项目中已经包含了 s.yaml 配置文件。你只需要修改其中的 role 字段为你的阿里云角色：

role: acs:ram::{您的阿里云主账号 ID}:role/{您的阿里云角色名称}

配置部署密钥：

s config add
# 按照引导输入 Access Key ID 和 Secret，记住密钥对名称（如 agentrun-deploy）

执行部署：

s deploy -a agentrun-deploy

部署完成后，你会得到一个 HTTPS URL，就可以在生产环境调用你的 Agent 了。

不同框架的集成案例

函数计算 AgentRun 不仅支持 LangChain，还深度集成了主流的 Agent 开发框架。所有框架都遵循同样的理念：通过简单的适配层，让你的代码无缝迁移到函数计算 AgentRun，享受企业级能力。

LangGraph：工作流编排

LangGraph 是 LangChain 团队推出的工作流编排框架，适合构建复杂的多步骤 Agent。集成方式和 LangChain 类似：

from agentrun.integration.langgraph import model, tools
from langgraph.graph import StateGraph, MessagesState
from langgraph.prebuilt import ToolNode
# 使用 AgentRun 的模型和工具
llm = model("<your-model-name>").to_langgraph()
agent_tools = tools()
# 构建 LangGraph 工作流（和原来的代码一样）
def call_model(state: MessagesState):
    messages = state["messages"]
    response = llm.invoke(messages)
    return {"messages": [response]}
workflow = StateGraph(MessagesState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", ToolNode(agent_tools))
workflow.set_entry_point("agent")
# 定义条件边...
app = workflow.compile()
# 调用
result = app.invoke({"messages": [HumanMessage(content="查询上海天气")]})

LangGraph 的优势是可以精确控制 Agent 的执行流程，比如条件分支、循环、并行执行等。部署到函数计算 AgentRun 后，这些复杂的工作流都能自动享受弹性伸缩和可观测能力。

AgentScope：多智能体协作

AgentScope 是阿里达摩院开源的多智能体框架，特别适合构建多 Agent 协作场景。集成方式：

from agentrun.integration.agentscope import model, tools
from agentscope.agent import ReActAgent
from agentscope.tool import Toolkit
# 使用 AgentRun 的模型和工具
llm = model("<your-model-name>").to_agentscope()
agent_tools = tools()
# 注册工具到 Toolkit
toolkit = Toolkit()
for tool in agent_tools:
    toolkit.register_tool_function(tool)
# 创建 Agent（和原来的代码一样）
agent = ReActAgent(
    name="assistant",
    sys_prompt="你是一个智能助手",
    model=llm,
    toolkit=toolkit,
)
# 调用
result = await agent.reply(Msg(name="user", content="查询上海天气", role="user"))

AgentScope 的优势是对多 Agent 系统的原生支持，包括 Agent 之间的通信、协调、记忆共享等。部署到函数计算 AgentRun 后，每个 Agent 都在独立的隔离环境中运行，确保安全性。

PydanticAI：类型安全的 Agent 框架

PydanticAI 是一个新兴框架，强调类型安全和结构化输出。集成方式：

from agentrun.integration.pydantic_ai import model, tools
from pydantic_ai import Agent
# 使用 AgentRun 的模型和工具
llm = model("<your-model-name>").to_pydantic_ai()
agent_tools = tools()
# 创建 Agent
agent = Agent(
    llm,
    instructions="Be concise, reply with one sentence.",
    tools=agent_tools,
)
# 同步调用
result = agent.run_sync("上海的天气如何？")
# 异步调用
result = await agent.run("上海的天气如何？")

PydanticAI 的优势是强类型和结构化输出，特别适合需要严格数据验证的企业场景。

充分利用函数计算 AgentRun 的核心能力

将 Agent 部署到函数计算 AgentRun 后，你不仅获得了 Serverless 运行环境，还可以深度利用平台提供的各种企业级能力。

模型高可用：告别单点故障（搭配 AI 网关）

部署到函数计算 AgentRun 后，你的 Agent 自动享受模型高可用能力。当你配置的主模型出现故障、限流或超时时，系统会自动切换到备用模型，整个过程对你的代码完全透明。
在函数计算 AgentRun 控制台配置模型时可以和 AI 网关进行联动，可以设置：主模型（如 GPT-4），备用模型列表（如 Claude-3、Qwen-Max），熔断策略（错误率阈值、超时时间），负载均衡策略（轮询、权重、最少连接）。
你的代码完全不需要改动，只需要在创建模型时使用函数计算 AgentRun 的模型名称，所有的容错、切换、负载均衡都由平台自动处理。

企业级 Sandbox：安全执行代码

函数计算 AgentRun 提供的 Sandbox 不是简单的代码执行环境，而是企业级的安全隔离沙箱。每个 Sandbox 实例都是独立隔离的，支持多种执行类型：

Code Interpreter 支持 Python、Node.js、Java、Bash 等语言，可以执行数据分析、文件处理等任务。Browser Tool 提供浏览器自动化能力，支持网页爬取、表单填写、截图等操作。All In One 集成了代码解释器和浏览器工具，提供更丰富的交互能力。

使用时，通过 sandbox_toolset() 函数就可以获取相应的工具集合，这些工具会自动转换为你使用的框架所需的格式。

工具和 MCP：标准化集成

函数计算 AgentRun 提供统一的工具管理和 MCP（Model Context Protocol）机制。你可以从工具市场选择现成的工具，也可以自定义工具并发布到市场。

更强大的是 MCP 的 Hook 机制。通过前置 Hook，可以在工具调用前自动注入用户凭证、记录请求日志、校验参数合法性。通过后置 Hook，可以对结果进行转换、记录审计日志、处理异常情况。这些通用逻辑不需要在每个工具中重复实现，大大提升了开发效率。

全链路可观测：不再是黑盒

这是函数计算 AgentRun 最强大的能力之一。你的代码不需要做任何改动，平台会自动记录 Agent 的完整执行链路。

在可观测平台上，你可以看到：Agent 接收到用户请求的时间和内容，调用了哪个模型、使用了多少 Token、花费了多少钱，调用了哪些工具、每个工具的执行时间和结果，访问了哪些知识库、检索了多少数据，每个环节的耗时分布，完整的调用链 Trace。

这些能力都是平台自动提供的，通过探针注入实现，无论是高代码还是低代码创建的 Agent，都自动享受这些可观测能力。

记忆和知识库：数据不出域

函数计算 AgentRun 深度集成了 RAGFlow、Mem0 等开源项目，提供灵活的记忆和知识库管理。你可以选择一键托管模式，由平台统一管理部署运维，享受 Serverless 的弹性和按量付费优势。也可以选择绑定模式，将 Agent 连接到已经部署在企业 VPC 或 IDC 内的实例，数据完全不出企业内网。

这种灵活性让你可以根据数据的敏感级别选择不同的策略：核心业务数据私有化部署，一般数据托管上云，在安全性和便利性之间找到最佳平衡。

立即体验函数计算 AgentRun

函数计算 AgentRun 的无代码到高代码演进能力，现已开放体验：

1. 快速创建： 访问控制台（ https://functionai.console.aliyun.com/cn-hangzhou/agent/explore ），60 秒创建你的第一个 Agent
2. 深度定制： 当需要更复杂功能时，一键转换为高代码
3. 持续演进： 利用函数计算 AgentRun 的基础设施能力，持续优化你的 Agent

从想法到上线，从原型到生产，函数计算 AgentRun 始终是你最好的伙伴。欢迎加入“函数计算 AgentRun 客户群”，钉钉群号：134570017218。

快速了解函数计算 AgentRun：

一句话介绍：函数计算 AgentRun 是一个以高代码为核心的一站式 Agentic AI 基础设施平台。秉持生态开放和灵活组装的理念，为企业级 Agent 应用提供从开发、部署到运维的全生命周期管理。

函数计算 AgentRun 架构图

AgentRun 运行时基于阿里云函数计算 FC 构建，继承了 Serverless 计算极致弹性、按量付费、零运维的核心优势。通过深度集成 AgentScope、LangChain、RAGFlow、Mem0 等主流开源生态。函数计算 AgentRun 将 Serverless 的极致弹性、零运维和按量付费的特性与 AI 原生应用场景深度融合，助力企业实现成本与效率的极致优化，平均 TCO 降低 60% 。 

让 开发者只需专注于 Agent 的业务逻辑创新，无需关心底层基础设施， 让 Agentic AI 真正进入企业生产环境。

作者：辰泉

前言

在 Agentic AI 时代，智能体需要与真实世界交互，而浏览器是连接虚拟世界与现实世界的重要桥梁。AgentRun Browser Sandbox 为智能体提供了安全、高性能、免运维的浏览器执行环境，让 AI Agent 真正具备“上网”的能力——从网页抓取、信息提取到表单填写、自动化操作，一切皆可实现。

AgentRun Browser Sandbox 介绍

什么是 Browser Sandbox?

Browser Sandbox 是 AgentRun 平台提供的云原生无头浏览器沙箱服务，基于阿里云函数计算（FC）构建。它为智能体提供了一个安全隔离的浏览器执行环境，支持通过标准的 Chrome DevTools Protocol (CDP) 远程控制浏览器实例。

核心特性

无头浏览器能力

• 内置 Chromium/Chrome 浏览器，支持完整的 Web 标准
• 原生兼容 Puppeteer、Playwright 等主流自动化框架
• 支持通过 CDP 协议进行精细化控制

实时可视化

• 内置 VNC 服务，支持实时查看浏览器界面
• 提供操作录制功能，方便调试和回放
• 支持通过 noVNC 客户端在网页中直接交互

安全与隔离

• 每个沙箱实例运行在独立的容器环境中
• 文件系统和进程空间完全隔离
• 支持 WSS 加密传输，确保数据安全

Serverless 架构

• 按需创建，按量付费，无需提前预置资源
• 快速弹性伸缩，支持高并发场景
• 零运维，无需管理服务器和浏览器依赖

主要应用场景

• AI Agent 赋能： 为大模型提供“眼睛”和“手”，执行网页浏览、信息提取、在线操作等任务
• 自动化测试： 在云端运行端到端（E2E）测试和视觉回归测试
• 数据采集： 稳定、高效地进行网页抓取，应对动态加载和反爬虫挑战
• 内容生成： 自动化生成网页截图或 PDF 文档

上手使用 AgentRun Browser Sandbox

AgentRun SDK 快速介绍

后续的内容将基于 AgentRun SDK 进行，因此我们先对 SDK 进行简要介绍。

Agentrun SDK 是一个开源的开发者工具包，本期介绍 Python 版本。其旨在简化智能体与 AgentRun 平台各种服务（包括 Browser Sandbox）的集成。它提供了统一的接口，让您可以用几行代码就将沙箱能力集成到现有的 Agent 框架中。SDK 的核心功能如下：

统一集成接口

• 提供对 LangChain、AgentScope 等主流框架的开箱即用支持
• 统一的模型代理接口，简化多模型管理
• 标准化的工具注册机制

Sandbox 生命周期管理

• 自动创建和销毁沙箱实例
• 支持会话级别的状态保持
• 灵活的资源配置和超时控制

安装 AgentRun SDK

pip install agentrun-sdk[playwright,server]

注意： 确保您的 Python 环境版本在 3.10 及以上。

基本使用示例

以下是使用 AgentRun SDK 创建和管理 Browser Sandbox 的核心代码：

from agentrun.sandbox import Sandbox, TemplateType
from playwright.sync_api import sync_playwright
# 创建 Browser Sandbox
sandbox = Sandbox.create(
    template_type=TemplateType.BROWSER,
    template_name="your-template-name",
    sandbox_idle_timeout_seconds=300
)
# 获取 CDP URL（用于 Playwright 连接）
cdp_url = sandbox.get_cdp_url()
# 使用 Playwright 连接并操作
with sync_playwright() as p:
    browser = p.chromium.connect_over_cdp(cdp_url)
    page = browser.contexts[0].pages[0]
    page.goto("https://www.example.com")
    page.screenshot(path="screenshot.png")
    browser.close()
# 销毁 Sandbox
sandbox.delete()

关键概念：

• template_name： 控制台创建的浏览器环境模板
• cdp_url： 用于 Playwright/Puppeteer 连接
• vnc_url： 用于实时查看浏览器画面（可通过 sandbox.get_cdp_url() 获取）

注意： 由于所有浏览器操作都在云端进行，您无需在本地安装浏览器。Playwright 仅用于通过 CDP 协议连接到云端的浏览器实例。

如何创建 Sandbox 模板

使用 Browser Sandbox 需要新建 Sandbox 模板，您需要访问 AgentRun 控制台网站 [ 1] ，并按照如下步骤创建模板：

1. 在顶部菜单栏选择“运行时与沙箱”；
2. 在左侧边栏选择“Sandbox 沙箱”；
3. 点击右上角“创建沙箱模板”；

4. 选择“浏览器”；

5. 在弹出的抽屉对话框中填写和选择您的模板的规格、网络等配置，并复制模板名称；

6. 点击“创建浏览器”等待其就绪即可。

从零开始用 LangChain 创建 Browser Sandbox 智能体

本教程将指导您从零开始创建一个完整的 Browser Sandbox 智能体项目。

基于 LangChain 集成 Browser Sandbox

本教程将详细讲解如何使用 LangChain 创建 Browser Sandbox 相关的 Tools 并集成到 Agent 中。

项目结构

为了保持代码的内聚性和可维护性，我们将代码拆分为以下模块：

模块职责划分：

sandbox_manager.py：负责 Sandbox 的创建、管理和销毁，提供统一的接口
langchain_agent.py：负责创建 LangChain Tools 和 Agent，集成 VNC 信息
main.py：作为入口文件，演示如何使用上述模块

步骤 1：创建项目并安装依赖

首先创建项目目录（如果还没有）：

mkdir -p langchain-demo
cd langchain-demo

创建 requirements.txt 文件，内容如下：

# LangChain 核心库
langchain>=0.1.0
langchain-openai>=0.0.5
langchain-community>=0.0.20
# AgentRun SDK
agentrun-sdk[playwright,server]>=0.0.8
# 浏览器自动化
playwright>=1.40.0
# 环境变量管理
python-dotenv>=1.0.0

然后安装依赖：

pip install -r requirements.txt

主要依赖说明：

• langchain 和 langchain-openai：LangChain 核心库
• agentrun-sdk[playwright,server]：AgentRun SDK，用于 Sandbox 管理
• playwright：浏览器自动化库
• python-dotenv：环境变量管理

步骤 2：配置环境变量

在项目根目录创建 .env 文件，配置以下环境变量：

# 阿里云百炼平台的 API Key，用于调用大模型能力
# 请前往 https://bailian.console.aliyun.com/?tab=app#/api-key 创建和查看
DASHSCOPE_API_KEY=sk-your-bailian-api-key
# 阿里云账号的访问密钥 ID 和访问密钥 Secret，用于 AgentRun SDK 鉴权
ALIBABA_CLOUD_ACCESS_KEY_ID=your-ak
ALIBABA_CLOUD_ACCESS_KEY_SECRET=your-sk
ALIBABA_CLOUD_ACCOUNT_ID=your-main-account-id
ALIBABA_CLOUD_REGION=cn-hangzhou
# browser sandbox 模板的名称，可以在 https://functionai.console.aliyun.com/cn-hangzhou/agent/runtime/sandbox 控制台创建
BROWSER_TEMPLATE_NAME=sandbox-your-template-name
# agentrun 的控制面和数据面的 API 端点请求地址，默认cn-hangzhou
AGENTRUN_CONTROL_ENDPOINT=agentrun.cn-hangzhou.aliyuncs.com
AGENTRUN_DATA_ENDPOINT=https://${your-main-account-id}.agentrun-data.cn-hangzhou.aliyuncs.com

步骤 3：创建 Sandbox 生命周期管理模块

创建 sandbox_manager.py 文件，负责 Sandbox 的创建、管理和销毁。核心代码如下：

"""
Sandbox 生命周期管理模块
负责 AgentRun Browser Sandbox 的创建、管理和销毁。
提供统一的接口供 LangChain Agent 使用。
"""
import os
from typing import Optional, Dict, Any
from dotenv import load_dotenv
# 加载环境变量
load_dotenv()
class SandboxManager:
    """Sandbox 生命周期管理器"""
    def __init__(self):
        self._sandbox: Optional[Any] = None
        self._sandbox_id: Optional[str] = None
        self._cdp_url: Optional[str] = None
        self._vnc_url: Optional[str] = None
    def create(
        self,
        template_name: Optional[str] = None,
        idle_timeout: int = 3000
    ) -> Dict[str, Any]:
        """
        创建或获取一个浏览器 sandbox 实例
        Args:
            template_name: Sandbox 模板名称，如果为 None 则从环境变量读取
            idle_timeout: 空闲超时时间（秒），默认 3000 秒
        Returns:
            dict: 包含 sandbox_id, cdp_url, vnc_url 的字典
        Raises:
            RuntimeError: 创建失败时抛出异常
        """
        try:
            from agentrun.sandbox import Sandbox, TemplateType
            # 如果已有 sandbox，直接返回
            if self._sandbox is not None:
                return self.get_info()
            # 从环境变量获取模板名称
            if template_name is None:
                template_name = os.getenv(
                    "BROWSER_TEMPLATE_NAME",
                    "sandbox-browser-demo"
                )
            # 创建 sandbox
            self._sandbox = Sandbox.create(
                template_type=TemplateType.BROWSER,
                template_name=template_name,
                sandbox_idle_timeout_seconds=idle_timeout
            )
            self._sandbox_id = self._sandbox.sandbox_id
            self._cdp_url = self._get_cdp_url()
            self._vnc_url = self._get_vnc_url()
            return self.get_info()
        except ImportError as e:
            print(e)
            raise RuntimeError(
                "agentrun-sdk 未安装，请运行: pip install agentrun-sdk[playwright,server]"
            )
        except Exception as e:
            raise RuntimeError(f"创建 Sandbox 失败: {str(e)}")
    def get_info(self) -> Dict[str, Any]:
        """
        获取当前 sandbox 的信息
        Returns:
            dict: 包含 sandbox_id, cdp_url, vnc_url 的字典
        Raises:
            RuntimeError: 如果没有活动的 sandbox
        """
        if self._sandbox is None:
            raise RuntimeError("没有活动的 sandbox，请先创建")
        return {
            "sandbox_id": self._sandbox_id,
            "cdp_url": self._cdp_url,
            "vnc_url": self._vnc_url,
        }
    def get_cdp_url(self) -> Optional[str]:
        """获取 CDP URL"""
        return self._sandbox.get_cdp_url()
    def get_vnc_url(self) -> Optional[str]:
        """获取 VNC URL"""
        return self._sandbox.get_vnc_url()
    def get_sandbox_id(self) -> Optional[str]:
        """获取 Sandbox ID"""
        return self._sandbox_id
    def destroy(self) -> str:
        """
        销毁当前的 sandbox 实例
        Returns:
            str: 操作结果描述
        """
        if self._sandbox is None:
            return "没有活动的 sandbox"
        try:
            sandbox_id = self._sandbox_id
            # 尝试销毁 sandbox
            if hasattr(self._sandbox, 'delete'):
                self._sandbox.delete()
            elif hasattr(self._sandbox, 'stop'):
                self._sandbox.stop()
            elif hasattr(self._sandbox, 'destroy'):
                self._sandbox.destroy()
            # 清理状态
            self._sandbox = None
            self._sandbox_id = None
            self._cdp_url = None
            self._vnc_url = None
            return f"Sandbox 已销毁: {sandbox_id}"
        except Exception as e:
            # 即使销毁失败，也清理本地状态
            self._sandbox = None
            self._sandbox_id = None
            self._cdp_url = None
            self._vnc_url = None
            return f"销毁 Sandbox 时出错: {str(e)}"
    def is_active(self) -> bool:
        """检查 sandbox 是否活跃"""
        return self._sandbox is not None
    def __enter__(self):
        """上下文管理器入口"""
        return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        """上下文管理器退出，自动销毁"""
        self.destroy()
        return False
# 全局单例（可选，用于简单场景）
_global_manager: Optional[SandboxManager] = None
def get_global_manager() -> SandboxManager:
    """获取全局 SandboxManager 单例"""
    global _global_manager
    if _global_manager is None:
        _global_manager = SandboxManager()
    return _global_manager
def reset_global_manager():
    """重置全局 SandboxManager"""
    global _global_manager
    if _global_manager:
        _global_manager.destroy()
    _global_manager = None

关键功能：

1. 创建 Sandbox： 使用 AgentRun SDK 创建浏览器 Sandbox
2. 获取连接信息： 自动获取 CDP URL 和 VNC URL，支持多种属性名兼容
3. 生命周期管理： 提供销毁方法，确保资源正确释放

步骤 4：创建 LangChain Tools 和 Agent

创建 langchain_agent.py 文件，定义 LangChain Tools 并创建 Agent。核心代码如下：

"""
LangChain Agent 和 Tools 注册模块
负责创建 LangChain Agent，注册 Sandbox 相关的 tools，并集成 VNC 可视化。
本模块使用 sandbox_manager.py 中封装的 SandboxManager 来管理 sandbox 生命周期。
"""
import os
from dotenv import load_dotenv
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from pydantic import BaseModel, Field
# 导入 sandbox 管理器
from sandbox_manager import SandboxManager
# 加载环境变量
load_dotenv()
# 全局 sandbox 管理器实例（单例模式）
_sandbox_manager: SandboxManager | None = None
def get_sandbox_manager() -> SandboxManager:
    """获取 sandbox 管理器实例（单例模式）"""
    global _sandbox_manager
    if _sandbox_manager is None:
        _sandbox_manager = SandboxManager()
    return _sandbox_manager
# ============ LangChain Tools 定义 ============
@tool
def create_browser_sandbox(
    template_name: str = None,
    idle_timeout: int = 3000
) -> str:
    """创建或获取一个浏览器 sandbox 实例。
    当需要访问网页、执行浏览器操作时，首先需要创建 sandbox。
    创建成功后，会返回 sandbox 信息，包括 VNC URL 用于可视化。
    Args:
        template_name: Sandbox 模板名称，如果不提供则从环境变量 BROWSER_TEMPLATE_NAME 读取
        idle_timeout: 空闲超时时间（秒），默认 3000 秒
    Returns:
        Sandbox 信息字符串，包括 ID、CDP URL、VNC URL
    """
    try:
        manager = get_sandbox_manager()
        # 如果 template_name 为空字符串，转换为 None 以便从环境变量读取
        if template_name == "":
            template_name = None
        info = manager.create(template_name=template_name, idle_timeout=idle_timeout)
        result = f"""✅ Sandbox 创建成功！
📋 Sandbox 信息:
- ID: {info['sandbox_id']}
- CDP URL: {info['cdp_url']}
"""
        vnc_url = info.get('vnc_url')
        if vnc_url:
            result += f"- VNC URL: {vnc_url}\n\n"
            result += "提示: VNC 查看器应该已自动打开，您可以在浏览器中实时查看浏览器操作。"
        else:
            result += "\n警告: 未获取到 VNC URL，可能无法使用可视化功能。"
        return result
    except Exception as e:
        return f" 创建 Sandbox 失败: {str(e)}"
@tool
def get_sandbox_info() -> str:
    """获取当前 sandbox 的详细信息，包括 ID、CDP URL、VNC URL 等。
    当需要查看当前 sandbox 状态或获取 VNC 连接信息时使用此工具。
    Returns:
        Sandbox 信息字符串
    """
    try:
        manager = get_sandbox_manager()
        info = manager.get_info()
        result = f"""📋 当前 Sandbox 信息:
- Sandbox ID: {info['sandbox_id']}
- CDP URL: {info['cdp_url']}
"""
        if info.get('vnc_url'):
            result += f"- VNC URL: {info['vnc_url']}\n\n"
            result += "您可以使用 VNC URL 在浏览器中实时查看操作过程。\n"
            result += "   推荐使用 vnc.html 文件或 noVNC 客户端。"
        return result
    except RuntimeError as e:
        return f" {str(e)}"
    except Exception as e:
        return f" 获取 Sandbox 信息失败: {str(e)}"
class NavigateInput(BaseModel):
    """浏览器导航输入参数"""
    url: str = Field(description="要访问的网页 URL，必须以 http:// 或 https:// 开头")
    wait_until: str = Field(
        default="load",
        description="等待页面加载的状态: load, domcontentloaded, networkidle"
    )
    timeout: int = Field(
        default=30000,
        description="超时时间（毫秒），默认 30000"
    )
@tool(args_schema=NavigateInput)
def navigate_to_url(url: str, wait_until: str = "load", timeout: int = 30000) -> str:
    """使用 sandbox 中的浏览器导航到指定 URL。
    当用户需要访问网页时使用此工具。导航后可以在 VNC 中实时查看页面。
    Args:
        url: 要访问的网页 URL
        wait_until: 等待页面加载的状态（load/domcontentloaded/networkidle）
        timeout: 超时时间（毫秒）
    Returns:
        导航结果描述
    """
    try:
        manager = get_sandbox_manager()
        if not manager.is_active():
            return " 错误: 请先创建 sandbox"
        # 验证 URL
        if not url.startswith(("http://", "https://")):
            return f" 错误: 无效的 URL 格式: {url}"
        cdp_url = manager.get_cdp_url()
        if not cdp_url:
            return " 错误: 无法获取 CDP URL"
        # 使用 Playwright 连接浏览器并导航
        try:
            from playwright.sync_api import sync_playwright
            with sync_playwright() as p:
                browser = p.chromium.connect_over_cdp(cdp_url)
                pages = browser.contexts[0].pages if browser.contexts else []
                if pages:
                    page = pages[0]
                else:
                    page = browser.new_page()
                page.goto(url, wait_until=wait_until, timeout=timeout)
                title = page.title()
                return f"已成功导航到: {url}\n📄 页面标题: {title}\n💡 您可以在 VNC 中查看页面内容。"
        except ImportError:
            return f"导航指令已发送: {url}\n💡 提示: 安装 playwright 以启用实际导航功能 (pip install playwright)"
        except Exception as e:
            return f" 导航失败: {str(e)}"
    except Exception as e:
        return f" 操作失败: {str(e)}"
@tool("browser_screenshot", description="在浏览器 sandbox 中截取当前页面截图")
def take_screenshot(filename: str = "screenshot.png") -> str:
    """截取浏览器当前页面的截图。
    Args:
        filename: 截图文件名，默认 "screenshot.png"
    Returns:
        操作结果
    """
    try:
        manager = get_sandbox_manager()
        if not manager.is_active():
            return " 错误: 请先创建 sandbox"
        cdp_url = manager.get_cdp_url()
        if not cdp_url:
            return " 错误: 无法获取 CDP URL"
        try:
            from playwright.sync_api import sync_playwright
            with sync_playwright() as p:
                browser = p.chromium.connect_over_cdp(cdp_url)
                pages = browser.contexts[0].pages if browser.contexts else []
                if pages:
                    page = pages[0]
                else:
                    return " 错误: 没有打开的页面"
                page.screenshot(path=filename)
                return f"截图已保存: {filename}"
        except ImportError:
            return " 错误: 需要安装 playwright (pip install playwright)"
        except Exception as e:
            return f" 截图失败: {str(e)}"
    except Exception as e:
        return f" 操作失败: {str(e)}"
@tool("destroy_sandbox", description="销毁当前的 sandbox 实例，释放资源。注意：仅在程序退出或明确需要释放资源时使用，不要在一轮对话后销毁。")
def destroy_sandbox() -> str:
    """销毁当前的 sandbox 实例。
    重要提示：此工具应该仅在以下情况使用：
    - 程序即将退出
    - 明确需要释放资源
    - 用户明确要求销毁
    不要在一轮对话完成后就销毁 sandbox，因为 sandbox 可以在多轮对话中复用。
    Returns:
        操作结果
    """
    try:
        manager = get_sandbox_manager()
        result = manager.destroy()
        return result
    except Exception as e:
        return f" 销毁失败: {str(e)}"
# ============ Agent 创建 ============
def create_browser_agent(system_prompt: str = None):
    """
    创建带有 sandbox 工具的 LangChain Agent
    Args:
        system_prompt: 自定义系统提示词，如果为 None 则使用默认提示词
    Returns:
        LangChain Agent 实例
    """
    # 配置 DashScope API
    api_key = os.getenv("DASHSCOPE_API_KEY")
    if not api_key:
        raise ValueError("请设置环境变量 DASHSCOPE_API_KEY")
    base_url = "https://dashscope.aliyuncs.com/compatible-mode/v1"
    model_name = os.getenv("QWEN_MODEL", "qwen-plus")
    # 创建 LLM
    model = ChatOpenAI(
        model=model_name,
        api_key=api_key,
        base_url=base_url,
        temperature=0.7,
    )
    # 创建工具列表
    tools = [
        create_browser_sandbox,
        get_sandbox_info,
        navigate_to_url,
        take_screenshot,
        destroy_sandbox,
    ]
    # 默认系统提示词
    if system_prompt is None:
        system_prompt = """你是一个浏览器自动化助手，可以使用 sandbox 来访问和操作网页。
当用户需要访问网页时，请按以下步骤操作：
1. 首先创建或获取 sandbox（如果还没有）
2. 使用 navigate_to_url 导航到目标网页
3. 执行用户请求的操作
4. 如果需要，可以截取截图
重要提示：
- 创建 sandbox 后，会返回 VNC URL，用户可以使用它实时查看浏览器操作
- 所有操作都会在 VNC 中实时显示，方便调试和监控
- sandbox 可以在多轮对话中复用，不要在一轮对话完成后就销毁
- 只有在用户明确要求销毁时才使用 destroy_sandbox 工具
- 不要主动建议用户销毁 sandbox，除非用户明确要求
- 请始终用中文回复，确保操作准确、高效。"""
    # 创建 Agent
    agent = create_agent(
        model=model,
        tools=tools,
        system_prompt=system_prompt,
    )
    return agent
def get_available_tools():
    """获取所有可用的工具列表"""
    return [
        create_browser_sandbox,
        get_sandbox_info,
        navigate_to_url,
        take_screenshot,
        destroy_sandbox,
    ]

关键要点：

1. Tool 定义： 使用 @tool 装饰器定义 LangChain Tools
2. 类型提示： 所有参数必须有类型提示，用于生成工具 schema
3. 文档字符串： 详细的文档字符串帮助 LLM 理解何时使用工具**
4. 单例模式： 使用全局管理器实例确保 Sandbox 在会话中复用**

步骤 5：创建主入口文件

创建 main.py 文件，作为程序入口。核心代码如下：

"""
LangChain + AgentRun Browser Sandbox 集成示例
主入口文件，演示如何使用 LangChain Agent 与 AgentRun Browser Sandbox 集成。
"""
import os
import sys
import signal
import webbrowser
import urllib.parse
import threading
import http.server
import socketserver
from pathlib import Path
from dotenv import load_dotenv
from langchain_agent import create_browser_agent, get_sandbox_manager
# 加载环境变量
load_dotenv()
# 全局 HTTP 服务器实例
_http_server = None
_http_port = 8080
# 全局清理标志，用于防止重复清理
_cleanup_done = False
def start_http_server():
    """启动一个简单的 HTTP 服务器来提供 vnc.html"""
    global _http_server
    if _http_server is not None:
        return _http_port
    try:
        current_dir = Path(__file__).parent.absolute()
        class VNCRequestHandler(http.server.SimpleHTTPRequestHandler):
            def __init__(self, *args, **kwargs):
                super().__init__(*args, directory=str(current_dir), **kwargs)
            def log_message(self, format, *args):
                # 静默日志，避免输出过多信息
                pass
        # 尝试启动服务器
        for port in range(_http_port, _http_port + 10):
            try:
                server = socketserver.TCPServer(("", port), VNCRequestHandler)
                server.allow_reuse_address = True
                # 在后台线程中运行服务器
                def run_server():
                    server.serve_forever()
                thread = threading.Thread(target=run_server, daemon=True)
                thread.start()
                _http_server = server
                return port
            except OSError:
                continue
        return None
    except Exception as e:
        print(f"启动 HTTP 服务器失败: {str(e)}")
        return None
def open_vnc_viewer(vnc_url: str):
    """
    自动打开 VNC 查看器并设置 VNC URL
    Args:
        vnc_url: VNC WebSocket URL
    """
    if not vnc_url:
        return
    try:
        # 获取当前文件所在目录
        current_dir = Path(__file__).parent.absolute()
        vnc_html_path = current_dir / "vnc.html"
        # 检查文件是否存在
        if not vnc_html_path.exists():
            print(f"警告: vnc.html 文件不存在: {vnc_html_path}")
            print_vnc_info(vnc_url)
            return
        # 启动 HTTP 服务器
        port = start_http_server()
        if port:
            # 编码 VNC URL 作为 URL 参数
            encoded_url = urllib.parse.quote(vnc_url, safe='')
            # 构建 HTTP URL
            http_url = f"http://localhost:{port}/vnc.html?url={encoded_url}"
            # 打开浏览器
            print(f"\n正在打开 VNC 查看器...")
            print(f"HTTP 服务器运行在: http://localhost:{port}")
            print(f"VNC URL: {vnc_url[:80]}...")
            print(f"完整 URL: {http_url[:100]}...")
            webbrowser.open(http_url)
            print(f"VNC 查看器已打开")
            print(f"VNC URL 已通过 URL 参数自动设置，页面加载后会自动连接")
        else:
            # 如果 HTTP 服务器启动失败，尝试使用 file:// 协议
            print(f"HTTP 服务器启动失败，尝试使用文件协议...")
            encoded_url = urllib.parse.quote(vnc_url, safe='')
            file_url = f"file://{vnc_html_path}?url={encoded_url}"
            webbrowser.open(file_url)
            print(f"VNC 查看器已打开（使用文件协议）")
            print(f"提示: 如果无法自动连接，请手动复制 VNC URL 到输入框")
    except Exception as e:
        print(f"自动打开 VNC 查看器失败: {str(e)}")
        print_vnc_info(vnc_url)
def print_vnc_info(vnc_url: str):
    """打印 VNC 连接信息"""
    if not vnc_url:
        return
    print("\n" + "=" * 60)
    print("VNC 可视化连接信息")
    print("=" * 60)
    print(f"\nVNC URL: {vnc_url}")
    print("\n使用方式:")
    print("   1. 使用 noVNC 客户端连接")
    print("   2. 或在浏览器中访问 VNC 查看器页面")
    print("   3. 实时查看浏览器操作过程")
    print("\n" + "=" * 60 + "\n")
def cleanup_sandbox():
    """
    清理 sandbox 资源
    这个函数可以被信号处理器、异常处理器和正常退出流程调用
    """
    global _cleanup_done
    # 防止重复清理
    if _cleanup_done:
        return
    _cleanup_done = True
    try:
        manager = get_sandbox_manager()
        if manager.is_active():
            print("\n" + "=" * 60)
            print("正在清理 sandbox...")
            print("=" * 60)
            result = manager.destroy()
            print(f"清理结果: {result}\n")
        else:
            print("\n没有活动的 sandbox 需要清理\n")
    except Exception as e:
        print(f"\n清理 sandbox 时出错: {str(e)}\n")
def signal_handler(signum, frame):
    """
    信号处理器，处理 Ctrl+C (SIGINT) 和其他信号
    Args:
        signum: 信号编号
        frame: 当前堆栈帧
    """
    print("\n\n收到中断信号，正在清理资源...")
    cleanup_sandbox()
    print("清理完成")
    sys.exit(0)
def main():
    """主函数"""
    global _cleanup_done
    # 重置清理标志
    _cleanup_done = False
    # 注册信号处理器，处理 Ctrl+C (SIGINT)
    signal.signal(signal.SIGINT, signal_handler)
    # 在 Windows 上，SIGBREAK 也可以处理
    if hasattr(signal, 'SIGBREAK'):
        signal.signal(signal.SIGBREAK, signal_handler)
    print("=" * 60)
    print("LangChain + AgentRun Browser Sandbox 集成示例")
    print("=" * 60)
    print()
    try:
        # 创建 Agent
        print("正在初始化 LangChain Agent...")
        agent = create_browser_agent()
        print("Agent 初始化完成\n")
        # 示例查询
        queries = [
            "创建一个浏览器 sandbox",
            "获取当前 sandbox 的信息，包括 VNC URL",
            "导航到 https://www.aliyun.com",
            "截取当前页面截图",
        ]
        # 执行查询
        for i, query in enumerate(queries, 1):
            print(f"\n{'=' * 60}")
            print(f"查询 {i}: {query}")
            print(f"{'=' * 60}\n")
            try:
                result = agent.invoke({
                    "messages": [{"role": "user", "content": query}]
                })
                # 提取最后一条消息的内容
                output = result.get("messages", [])[-1].content if isinstance(result.get("messages"), list) else result.get("output", str(result))
                print(f"\n结果:\n{output}\n")
                # 如果是创建 sandbox，自动打开 VNC 查看器
                if i == 1:
                    try:
                        # 等待一下确保 sandbox 完全创建
                        import time
                        time.sleep(1)
                        manager = get_sandbox_manager()
                        if manager.is_active():
                            info = manager.get_info()
                            vnc_url = info.get('vnc_url')
                            if vnc_url:
                                print(f"\n检测到 VNC URL: {vnc_url[:80]}...")
                                open_vnc_viewer(vnc_url)
                                print_vnc_info(vnc_url)
                            else:
                                print("\n警告: 未获取到 VNC URL，请检查 sandbox 创建是否成功")
                    except Exception as e:
                        print(f"打开 VNC 查看器时出错: {str(e)}")
                        import traceback
                        traceback.print_exc()
                # 如果是获取信息，显示 VNC 信息
                elif i == 2:
                    try:
                        manager = get_sandbox_manager()
                        if manager.is_active():
                            info = manager.get_info()
                            if info.get('vnc_url'):
                                print_vnc_info(info['vnc_url'])
                    except:
                        pass
            except Exception as e:
                print(f"查询失败: {str(e)}\n")
                import traceback
                traceback.print_exc()
        # 交互式查询
        print("\n" + "=" * 60)
        print("进入交互模式（输入 'quit' 或 'exit' 退出，Ctrl+C 或 Ctrl+D 中断）")
        print("=" * 60 + "\n")
        while True:
            try:
                user_input = input("请输入您的查询: ").strip()
            except EOFError:
                # 处理 Ctrl+D (EOF)
                print("\n\n检测到输入结束 (Ctrl+D)，正在清理资源...")
                cleanup_sandbox()
                print("清理完成")
                break
            except KeyboardInterrupt:
                # 处理 Ctrl+C (在 input 调用期间)
                print("\n\n检测到中断信号 (Ctrl+C)，正在清理资源...")
                cleanup_sandbox()
                print("清理完成")
                break
            if not user_input:
                continue
            if user_input.lower() in ['quit', 'exit', '退出']:
                print("\nBye")
                # 退出前清理 sandbox
                cleanup_sandbox()
                break
            try:
                result = agent.invoke({
                    "messages": [{"role": "user", "content": user_input}]
                })
                output = result.get("messages", [])[-1].content if isinstance(result.get("messages"), list) else result.get("output", str(result))
                print(f"\n结果:\n{output}\n")
                # 检查是否需要打开或显示 VNC 信息
                user_input_lower = user_input.lower()
                if "创建" in user_input_lower and "sandbox" in user_input_lower:
                    # 如果是创建 sandbox，自动打开 VNC 查看器
                    try:
                        # 等待一下确保 sandbox 完全创建
                        import time
                        time.sleep(1)
                        manager = get_sandbox_manager()
                        if manager.is_active():
                            info = manager.get_info()
                            vnc_url = info.get('vnc_url')
                            if vnc_url:
                                print(f"\n检测到 VNC URL: {vnc_url[:80]}...")
                                open_vnc_viewer(vnc_url)
                                print_vnc_info(vnc_url)
                            else:
                                print("\n警告: 未获取到 VNC URL，请检查 sandbox 创建是否成功")
                    except Exception as e:
                        print(f"打开 VNC 查看器时出错: {str(e)}")
                        import traceback
                        traceback.print_exc()
                elif "sandbox" in user_input_lower or "vnc" in user_input_lower:
                    # 其他情况只显示信息
                    try:
                        manager = get_sandbox_manager()
                        if manager.is_active():
                            info = manager.get_info()
                            if info.get('vnc_url'):
                                print_vnc_info(info['vnc_url'])
                    except:
                        pass
            except Exception as e:
                print(f"查询失败: {str(e)}\n")
                import traceback
                traceback.print_exc()
        # 清理资源（仅在程序正常退出时）
        cleanup_sandbox()
    except KeyboardInterrupt:
        # 处理顶层 KeyboardInterrupt (Ctrl+C)
        print("\n\n检测到中断信号 (Ctrl+C)，正在清理资源...")
        cleanup_sandbox()
        print("清理完成")
        sys.exit(0)
    except EOFError:
        # 处理顶层 EOFError (Ctrl+D)
        print("\n\n检测到输入结束 (Ctrl+D)，正在清理资源...")
        cleanup_sandbox()
        print("清理完成")
        sys.exit(0)
    except ValueError as e:
        print(f"配置错误: {str(e)}")
        print("\n提示: 请确保已设置以下环境变量:")
        print("   - DASHSCOPE_API_KEY: DashScope API Key")
        print("   - ALIBABA_CLOUD_ACCOUNT_ID: 阿里云账号 ID")
        print("   - ALIBABA_CLOUD_ACCESS_KEY_ID: 访问密钥 ID")
        print("   - ALIBABA_CLOUD_ACCESS_KEY_SECRET: 访问密钥 Secret")
        print("   - ALIBABA_CLOUD_REGION: 区域（默认: cn-hangzhou）")
    except Exception as e:
        print(f"发生错误: {str(e)}")
        import traceback
        traceback.print_exc()
        # 发生错误时也尝试清理
        cleanup_sandbox()
if __name__ == "__main__":
    main()

关键功能：

1. VNC 自动打开： 创建 Sandbox 后自动打开 VNC 查看器
2. 信号处理： 捕获 Ctrl+C，确保资源正确清理
3. 交互模式： 支持持续对话，复用 Sandbox 实例

VNC 可视化集成

VNC（Virtual Network Computing）功能允许您实时查看和监控浏览器在 Sandbox 中的操作过程，这对于调试和监控 Agent 行为非常有用。

获取 VNC URL：

创建 Sandbox 后，可以通过 get_sandbox_info tool 获取 VNC URL：

# 通过 Agent 调用
result = agent.invoke({
    "messages": [{"role": "user", "content": "获取 sandbox 信息"}]
})
# 或直接通过管理器获取
manager = get_sandbox_manager()
info = manager.get_info()
vnc_url = info['vnc_url']

自动打开 VNC 查看器：

在 main.py 中，我们实现了自动打开 VNC 查看器的功能：

import webbrowser
import urllib.parse
from pathlib import Path
def open_vnc_viewer(vnc_url: str):
    """自动打开 VNC 查看器"""
    current_dir = Path(__file__).parent.absolute()
    vnc_html_path = current_dir / "vnc.html"
    if vnc_html_path.exists():
        # 通过 URL 参数传递 VNC URL
        encoded_url = urllib.parse.quote(vnc_url, safe='')
        file_url = f"file://{vnc_html_path}?url={encoded_url}"
        webbrowser.open(file_url)

VNC HTML 页面：

vnc.html 页面会从 URL 参数中读取 VNC URL，并自动连接到 VNC 服务器。页面包含以下核心功能：

1. noVNC 库加载： 从 CDN 动态加载 noVNC 客户端库
2. 自动连接： 读取 URL 参数中的 VNC URL 并自动连接
3. 状态显示： 显示连接状态（连接中、已连接、已断开）
4. 手动控制： 支持手动输入 VNC URL、断开重连等操作

核心 JavaScript 代码片段：

// 从 URL 参数获取 VNC URL
const urlParams = new URLSearchParams(window.location.search);
const vncUrl = urlParams.get('url');
// 加载 noVNC 库
async function loadNoVNC() {
    const module = await import('https://cdn.jsdelivr.net/gh/novnc/noVNC@v1.4.0/core/rfb.js');
    return module.default;
}
// 连接 VNC
async function connectVNC(url) {
    const RFB = await loadNoVNC();
    rfb = new RFB(vncScreen, url, {
        shared: true,
        credentials: { password: '' }
    });
    rfb.addEventListener('connect', () => {
        console.log('VNC 连接成功');
    });
}

完整的 vnc.html 文件可以在示例代码仓库中获取。

手动使用 VNC 查看器：

如果自动打开失败，您也可以手动使用 VNC 查看器：

1. 使用 noVNC 在线客户端：

• 访问 noVNC 在线客户端 [ 2]
• 在连接设置中填入 VNC URL
• 点击连接

2. 使用本地 VNC HTML 页面：

• 打开 vnc.html
• 输入 VNC URL
• 点击连接按钮

实时监控功能：

• 所有浏览器操作都会在 VNC 中实时显示
• 可以看到 Agent 的每一步操作（导航、点击、输入等）
• 方便调试和监控 Agent 行为
• 支持交互式操作（在 VNC 中直接操作浏览器）

运行和测试

python main.py

程序会自动：

1. 创建 Browser Sandbox
2. 打开 VNC 查看器（实时查看浏览器操作）
3. 执行预设查询
4. 进入交互模式

工作原理

为了更好地理解系统架构，我们将工作流程拆分为两个部分：LangChain Agent 工作流程和 SandboxManager 生命周期管理。

1. LangChain Agent 工作流程

下图展示了 LangChain Agent 如何处理用户请求并调用相应的 Tools：

Agent 工作流程说明：

1. 请求接收： 用户发起自然语言请求（如“访问淘宝首页并截图”）
2. 意图分析： Agent 分析用户意图，决定需要调用哪些 Tools
3. Tool 调用： 根据任务需求，顺序或组合调用多个 Tools
4. Manager 交互： 所有 Tools 都通过 SandboxManager 单例实例操作 Sandbox
5. 结果处理： Agent 将 Tool 返回的结果整合成用户友好的响应
6. 多轮对话： Sandbox 在整个会话中保持活跃，支持多轮对话

5 个核心 Tools 的职责：

2. SandboxManager 生命周期管理

下图展示了 SandboxManager 如何管理 Sandbox 的完整生命周期：

SandboxManager 工作流程说明：

1. 单例管理：

• 首次调用时创建 Manager 实例
• 后续调用复用同一个实例
• 确保整个会话只有一个 Sandbox

2. Sandbox 创建：

• 调用 AgentRun SDK 的 Sandbox.create()
• SDK 通过阿里云 API 与函数计算 FC 通信
• FC 服务创建独立的容器实例，包含：
• Chromium 浏览器 VNC 服务必要的运行环境

3. 连接信息获取：

• CDP URL： WebSocket 地址，用于 Playwright/Puppeteer 远程控制浏览器
• VNC URL： WebSocket 地址，用于实时查看浏览器画面**

4. 浏览器操作：

• Playwright 通过 CDP URL 连接到远程浏览器
• 执行各种浏览器操作（导航、点击、截图等）
• VNC 同步显示操作过程，用户可实时监控

5. 资源清理：

• 调用 destroy() 方法销毁 Sandbox
• 清理 Manager 内部状态
• 通过 SDK 释放云端资源

3. Agent 与 Manager 的协作关系

交互模式：

用户请求 → Agent → Tool → SandboxManager → AgentRun SDK → 云端 Sandbox
                                    ↓
用户响应 ← Agent ← Tool ← SandboxManager ← 操作结果

关键设计理念：

1. 分层架构：

• 用户层：自然语言交互
• Agent 层：意图理解和任务分解
• Tool 层：功能封装和参数验证
• Manager 层：资源管理和状态维护
• SDK 层：云服务通信
• 云端层：实际的 Sandbox 环境

2. 单例模式：

• SandboxManager 使用单例模式
• 保证整个会话中只有一个 Sandbox 实例
• 避免资源浪费和状态冲突

3. 状态复用：

• Sandbox 在多轮对话中保持活跃
• 减少创建和销毁的开销
• 提供更流畅的用户体验

4. 双通道设计：

• CDP 通道：Agent 通过 Playwright 控制浏览器
• VNC 通道：用户通过 VNC 查看器实时监控

5. 解耦设计：

• Tools 不直接操作 SDK，通过 Manager 统一管理
• 便于扩展和维护
• 统一的错误处理和资源管理

典型使用场景示例：

# 第 1 轮对话
用户: "创建一个 sandbox 并访问淘宝首页"
→ Agent 调用: create_browser_sandbox → navigate_to_url
→ Manager: 创建 Sandbox → Playwright 导航
→ 结果: "Sandbox 已创建，已访问淘宝首页"
# 第 2 轮对话（复用 Sandbox）
用户: "截取当前页面"
→ Agent 调用: take_screenshot
→ Manager: 使用现有 Sandbox → Playwright 截图
→ 结果: "截图已保存"
# 第 3 轮对话（复用 Sandbox）
用户: "访问京东首页"
→ Agent 调用: navigate_to_url
→ Manager: 使用现有 Sandbox → Playwright 导航
→ 结果: "已访问京东首页"

通过这种设计，Agent 专注于理解用户意图和任务编排，而 Manager 专注于 Sandbox 的生命周期管理，实现了清晰的职责分离。

工作原理总结：

1. 工具注册：使用 @tool 装饰器将 Sandbox 功能封装为 LangChain Tools
2. 生命周期管理： SandboxManager 负责 Sandbox 的创建、管理和销毁
3. 状态保持：使用单例模式管理 Sandbox 实例，确保同一会话内复用
4. VNC 集成：自动获取并返回 VNC URL，方便用户实时查看
5. 错误处理：所有工具都包含完善的错误处理机制

扩展和定制

添加自定义 Tools：

@tool
def extract_table_data(url: str) -> str:
    """从网页中提取表格数据"""
    from playwright.sync_api import sync_playwright
    manager = get_sandbox_manager()
    cdp_url = manager.get_info()['cdp_url']
    with sync_playwright() as p:
        browser = p.chromium.connect_over_cdp(cdp_url)
        page = browser.contexts[0].pages[0]
        page.goto(url)
        tables = page.query_selector_all("table")
        return f"找到 {len(tables)} 个表格"

自定义提示词：

custom_prompt = """你是一个专业的网页数据提取助手。
在执行任务前，请先创建 sandbox，然后使用浏览器工具完成任务。"""
agent = create_browser_agent(system_prompt=custom_prompt)

最佳实践

1. 模块化设计：将 Sandbox 管理和 Agent 创建分离，提高代码可维护性
2. 错误处理：所有工具都应包含完善的错误处理
3. 资源清理：使用信号处理器确保资源正确清理
4. VNC 提示：在工具返回中包含 VNC URL，方便用户使用
5. 单例模式：确保 Sandbox 实例在会话中复用，避免重复创建

前端集成可视化监控（VNC）

VNC 集成架构

下图展示了前端如何集成 VNC 实现实时监控：

轻量级 HTML 页面集成

创建一个简单的 vnc-viewer.html 文件：

<!DOCTYPE html>
<html>
<head>
    <title>Browser Sandbox VNC 查看器</title>
    <style>
        body { margin: 0; padding: 0; background: 
#000
; }
        
#vnc
-container { width: 100vw; height: 100vh; }
    </style>
</head>
<body>
    <div id="vnc-container"></div>
    <script type="module">
        const params = new URLSearchParams(window.location.search);
        const vncUrl = params.get('url');
        if (!vncUrl) {
            alert('请提供 VNC URL 参数');
        } else {
            const module = await import('https://cdn.jsdelivr.net/gh/novnc/noVNC@v1.4.0/core/rfb.js');
            const RFB = module.default;
            const rfb = new RFB(
                document.getElementById('vnc-container'),
                vncUrl,
                { shared: true, credentials: { password: '' } }
            );
            rfb.scaleViewport = true;
        }
    </script>
</body>
</html>

使用方式：

import webbrowser
import urllib.parse
vnc_url = sandbox.vnc_url
encoded_url = urllib.parse.quote(vnc_url, safe='')
viewer_url = f"file:///path/to/vnc-viewer.html?url={encoded_url}"
webbrowser.open(viewer_url)

React 应用集成

核心组件代码：

import React, { useEffect, useRef } from 'react';
interface VNCViewerProps {
  vncUrl: string;
  onConnect?: () => void;
  onDisconnect?: () => void;
}
export const VNCViewer: React.FC<VNCViewerProps> = ({ 
  vncUrl, 
  onConnect, 
  onDisconnect 
}) => {
  const containerRef = useRef<HTMLDivElement>(null);
  useEffect(() => {
    let rfb: any;
    const initVNC = async () => {
      if (!containerRef.current || !vncUrl) return;
      const { default: RFB } = await import('@novnc/novnc/core/rfb');
      rfb = new RFB(containerRef.current, vncUrl, {
        shared: true,
        credentials: { password: '' }
      });
      rfb.scaleViewport = true;
      rfb.addEventListener('connect', () => onConnect?.());
      rfb.addEventListener('disconnect', () => onDisconnect?.());
    };
    initVNC();
    return () => {
      if (rfb) rfb.disconnect();
    };
  }, [vncUrl, onConnect, onDisconnect]);
  return (
    <div 
      ref={containerRef} 
      style={{ width: '100%', height: '600px', background: '
#000
' }} 
    />
  );
};

使用示例：

import React, { useState, useEffect } from 'react';
import { VNCViewer } from './VNCViewer';
function App() {
  const [vncUrl, setVncUrl] = useState<string>('');
  useEffect(() => {
    fetch('/api/sandbox/create', { method: 'POST' })
      .then(res => res.json())
      .then(data => setVncUrl(data.vnc_url));
  }, []);
  return (
    <div>
      <h1>Browser Sandbox 实时监控</h1>
      {vncUrl ? (
        <VNCViewer 
          vncUrl={vncUrl}
          onConnect={() => console.log('已连接')}
          onDisconnect={() => console.log('已断开')}
        />
      ) : (
        <p>正在初始化...</p>
      )}
    </div>
  );
}

Puppeteer 和 Playwright 直接集成

如果您更熟悉传统的浏览器自动化库，也可以直接使用 Puppeteer 或 Playwright 连接到 Browser Sandbox。

使用 Playwright

from playwright.sync_api import sync_playwright
from agentrun.sandbox import Sandbox, TemplateType
# 创建 Sandbox
sandbox = Sandbox.create(
    template_type=TemplateType.BROWSER,
    template_name="your-template-name",
    sandbox_idle_timeout_seconds=3000
)
# 使用 Playwright 连接
with sync_playwright() as p:
    browser = p.chromium.connect_over_cdp(sandbox.cdp_url)
    page = browser.contexts[0].pages[0]
    # 执行操作
    page.goto("https://www.example.com")
    page.screenshot(path="screenshot.png")
    content = page.content()
    browser.close()
# 清理
sandbox.delete()

使用 Puppeteer（Node.js）

const puppeteer = require('puppeteer-core');
// CDP URL 从 Sandbox 获取
const cdpUrl = 'wss://your-account.funagent-data-pre.cn-hangzhou.aliyuncs.com/sandboxes/xxx/ws/automation';
(async () => {
  const browser = await puppeteer.connect({
    browserWSEndpoint: cdpUrl,
    defaultViewport: null
  });
  const page = (await browser.pages())[0];
  await page.goto('https://www.example.com');
  await page.screenshot({ path: 'screenshot.png' });
  await browser.close();
})();

总结

通过本教程，您已经学会了：

1. AgentRun SDK 基础： 如何使用 SDK 创建和管理 Browser Sandbox
2. LangChain 集成： 如何将 Sandbox 封装为 LangChain Tools
3. VNC 可视化： 如何在前端集成 VNC 实现实时监控
4. 直接集成： 如何使用 Puppeteer/Playwright 直接连接 Sandbox

相关链接：

[1] Agentrun 控制台网站

https://functionai.console.aliyun.com/cn-hangzhou/agent/runti...

[2] noVNC 在线客户端

https://novnc.com/noVNC/vnc.html

在上一篇《Claude Code × 智谱 BigModel 实战集成指南》中，我们已经完成了一次完整的项目实战。项目可以正常运行，但在后续代码 Review 时，一个问题逐渐暴露出来：

生成的代码虽然能跑，但大量 API 和用法已经过时，与最新官方文档存在明显偏差。

这在 AI 辅助开发中其实非常常见——模型的训练数据更新速度，往往赶不上框架和 SDK 的迭代速度。

正巧这时，一位朋友向我推荐了 Anthropic 最新发布的 Agent Skills，通过 plugins 的方式，让 Claude 在生成代码时 动态读取最新官方文档和工具能力，从而显著降低“写得像，但跑不通”的概率。

本文就是这次探索的完整记录。

一、Agent Skills 是什么？

官方仓库地址：

https://github.com/anthropics/skills

Agent Skills 可以理解为：

一套可插拔的“能力模块”，用于教会 Claude 如何用正确的方法、最新的工具、可重复的流程 来完成特定任务。

在技术层面上：

• 每个 Skill 本质上是一个文件夹

• 内部包含：

  • 指令（instructions）
  • 脚本（scripts）
  • 资源文件（resources）
• Claude Code 会在运行时动态加载这些 Skills

它能解决什么问题？

Agent Skills 的核心价值在于 “降低幻觉 + 提高一致性”，典型应用场景包括：

• 按公司/团队的编码规范生成代码
• 按最新官方文档调用 API（而不是靠模型记忆）
• 执行固定的工程化流程（初始化项目、生成目录结构、部署脚本等）
• 自动化个人或组织级任务

简单来说：

Skills 不是让模型更聪明，而是让模型更“守规矩”。

二、在 Claude Code 中安装 Agent Skills

在 Claude Code 命令行中执行：

/plugin marketplace add anthropics/skills

安装完成后，你就已经具备了使用官方 Skills 的能力。

这一步相当于为 Claude Code 打开了“官方增强模式”。

三、安装 context7 插件（关键步骤）

接下来是本文的重点：context7。

1️⃣ 打开插件管理

在 Claude Code 中输入：

/plugins

然后使用键盘 ➡️ 进入 Discover。

2️⃣ 搜索并安装 context7

在搜索框中输入 context7，完成安装。

context7 本质上是一个 MCP（Model Context Protocol）插件，
能让 Claude 直接参考并对齐最新的官方文档内容。

四、使用 context7 生成项目代码

安装完成后，就可以在 Prompt 中显式声明使用 context7。

示例 Prompt

---
name: context7
description: 使用 Context7，基于框架最新的官方文档
---

# context7

## 指南
已使用以下技术栈生成企业级项目：
- 使用 Context7，基于最新的官方文档
- FastAPI 0.128.0，带 Token 认证
  - 使用 sqlite 生成 token
  - 不使用 JWT，仅做 Token 校验
- langchain 1.2.6，使用 create_agent
- langchain-ollama 1.0.1
  - model：qwen3-vl:32b
  - embedding：qwen3-embedding:8b
- langgraph 1.0.6
- Milvus（pymilvus）2.6.6
- langfuse 3.12.0

通过这种方式，你是在明确告诉 Claude：

不要靠“印象”写代码，而是以当前官方文档为准。

五、实际体验与问题分析

真实结论只有一句话：

效果明显提升，但依然不能“一次生成直接可用”。

优点

• API 使用明显更接近最新文档
• 过时参数、废弃方法显著减少
• 工程结构更合理，思路更偏向“真实项目”

仍然存在的问题

• 复杂技术栈组合（LangChain + LangGraph + Milvus + Langfuse）
• 仍然需要 多轮调试才能完全跑通
• 某些边界用法依然存在偏差

我的判断

并不是 context7 不行，而是模型生成速度，依然落后于框架演进速度。

context7 做到的是：

• 让 Claude 看得到 最新文档
• 但最终“怎么拼起来”，仍然依赖模型本身的推理与代码能力

六、总结

如果你正在使用 Claude Code 做偏工程化、偏企业级的项目开发，我的建议是：

✅ 一定要上 Agent Skills

✅ 能用 context7 就用 context7

❌ 不要再完全相信“模型记忆里的 API”

但同时也要有一个清醒认知：

AI 辅助开发 = 更快的起点，而不是免调试的终点。

在当前阶段，最理想的模式依然是：

AI 生成 + 人类 Review + 多轮修正

后续我也会继续记录 Claude Code + MCP + 多模型协作 的实践经验，欢迎关注。

摘要

本文为普通人设计了从认知到应用、无代码到有代码、单一到复杂的智能体渐进式学习路径，分 8 个核心板块明确各阶段学习目标、实操方法、工具资源与避坑要点，同时通过高频 QA 解答零基础适配、学习时间投入、场景化学习重点等关键疑问，搭配可直接落地的 12 周学习计划，让不同基础、不同学习场景的学习者都能以 “先实践后理论” 为核心，从搭建简单智能体逐步进阶到开发落地化、甚至商业化的智能体系统，核心学习逻辑为以真实问题驱动实践，按需补充理论知识，快速积累可落地的智能体开发能力。

普通人学习智能体，应遵循 “从认知到应用、从无代码到有代码、从单一到复杂” 的渐进路径，先明确概念与应用场景，再通过零代码平台快速上手，逐步掌握核心技术并进阶实战，最终形成可落地的能力与作品。以下是分阶段的详细指南：

一、认知筑基（1-2 周）：先懂 “是什么” 再动手

1. 核心概念理解

• 明确智能体定义：具备感知、决策、执行能力，能自主完成目标的 AI 系统，区别于普通聊天机器人（后者无长期记忆与工具调用能力）。
• 掌握关键术语：提示词工程、思维链（CoT）、工具调用、记忆机制、多智能体协作等。
• 了解应用场景：办公自动化、客服、数据分析、游戏 AI、科研辅助等，结合自身需求选择切入点。

2. 资源推荐

• 入门读物：《AI 智能体入门与实践》《智能体时代：从对话到协作》，快速建立认知框架。
• 课程：吴恩达《机器学习专项课程》（Coursera）、DeepMind 强化学习入门视频，夯实 AI 基础。
• 社区：GitHub Awesome Agentic AI、知乎 “智能体” 话题，跟踪前沿动态与案例。

二、零代码实践（2-4 周）：快速做出第一个智能体

1. 平台选择（从易到难）

2. 实战项目（从简到繁）

1. ​个人助理​：用扣子平台搭建日程管理、邮件总结、文档问答智能体，集成日历、邮箱插件，掌握提示词编写与工具调用。
2. ​知识库助手​：上传 PDF/Word 文档到平台，搭建企业规章制度、产品手册问答智能体，解决实际业务问题。
3. ​多智能体协作​：用 CrewAI 创建 “写作 - 编辑 - 翻译” 团队，分工完成文案生产，理解任务拆分与角色定义。

3. 核心技能

• 提示词工程：学会写清晰指令（如 “总结收件箱中含‘会议纪要’的邮件，生成三点待办并添加到日历”），提升智能体执行效率。
• 工具集成：熟悉常用插件（API、数据库、办公软件），掌握参数配置与调试方法。
• 记忆管理：设置上下文窗口、长期记忆存储，确保智能体 “记住” 历史交互。

三、代码入门（4-8 周）：从调用 API 到自定义开发

1. 技术栈准备

• 编程语言：Python（必备），推荐《Python 编程：从入门到实践》快速上手。
• 基础库：OpenAI API、LangChain、Streamlit（快速搭建前端）。
• 数学基础：线性代数（矩阵运算）、概率论（贝叶斯定理）、基础微积分，理解模型原理。

2. 实战项目（代码驱动）

1. ​API 调用型智能体​：用 OpenAI Assistants API 开发文档分析工具，实现上传文件 → 提取信息 → 生成报告的自动化流程。
2. ​强化学习小实验​：用 OpenAI Gym+PyTorch 训练 CartPole 平衡智能体，理解状态、动作、奖励机制。
3. ​自定义工作流​：用 LangChain+Streamlit 搭建论文写作助手，集成文献搜索、大纲生成、内容撰写功能。

3. 避坑指南

• 先调通 API 再优化逻辑，避免过早陷入复杂算法。
• 善用社区代码模板（GitHub Gist、LangChain Cookbook），减少重复开发。
• 用 Streamlit 快速做前端，专注核心逻辑而非界面设计。

四、进阶深化（8-12 周）：掌握核心技术与多智能体协作

1. 核心技术突破

• 思维链（CoT）与计划执行（Plan-and-Execute）：优化提示词，让智能体拆解复杂任务（如 “写一篇市场分析报告”→“调研行业数据 → 分析竞品 → 撰写结论”）。
• 工具调用优化：设计工具选择逻辑，解决 “调用哪个工具”“何时调用” 的问题。
• 记忆与知识库：用向量数据库（Pinecone、Chroma）存储长文本，实现高效检索与上下文关联。

2. 多智能体系统实战

1. ​团队协作模型​：用 AutoGen 搭建 “产品经理 - 开发 - 测试” 智能体团队，完成小型软件项目的需求分析、代码编写、Bug 修复。
2. ​复杂任务处理​：开发 “科研助手” 系统，集成文献检索、数据处理、图表生成、论文写作功能，解决跨领域复杂问题。

3. 资源推荐

• 书籍：《深度强化学习实战》《LangChain 实战》，深入技术细节。
• 课程：斯坦福 CS221（人工智能原理）、伯克利 RL Course，提升理论水平。
• 开源项目：AutoGen、MetaGPT 源码阅读，学习工业级架构设计。

五、工程化与落地（12 周 +）：从原型到产品

1. 工程能力建设

• 部署与监控：用 Docker 容器化智能体，阿里云 / 腾讯云部署，Prometheus 监控性能。
• 数据安全：敏感信息加密，遵循 GDPR / 个人信息保护法，确保合规。
• 迭代优化：建立用户反馈机制，用 A/B 测试优化提示词与模型参数。

2. 商业化方向

• 垂直领域解决方案：为教育、医疗、金融行业定制智能体（如学生辅导、病历分析、投资顾问）。
• 企业效率工具：开发自动化办公套件，对接 OA 系统，提升团队协作效率。
• 开源贡献：参与 LangChain、AutoGen 等项目，积累技术影响力。

六、常见误区与避坑建议

1. ​误区​：一上来就啃底层算法（如深度学习、强化学习数学推导）。
   ​建议​：先通过零代码平台做出可用产品，再按需补数学与算法知识。

   1. ​误区​：忽视提示词工程，过度依赖模型能力。

      ​建议​：提示词是智能体的 “灵魂”，花时间优化指令，比盲目换模型更有效。

      1. ​误区​：追求 “大而全”，忽略落地场景。

         ​建议​：从解决小问题（如 “每日邮件总结”）入手，逐步扩展功能，避免半途而废。

      七、QA 问答：解决学习中的高频疑问

      Q1：零基础、不懂编程，能学会智能体吗？

      A：完全可以。目前主流的零代码平台（如扣子、CrewAI）已实现可视化拖拽操作，无需编写代码就能搭建简单智能体。建议先从这类平台入手，完成 “个人助理”“知识库问答” 等基础项目，积累实战经验后，再根据需求决定是否学习编程进阶。学习的核心是 “解决问题”，而非必须掌握编程技能。

      Q2：学习智能体需要掌握哪些数学知识？必须深入学深度学习吗？

      A：无需一开始就深入学习复杂数学和深度学习。入门阶段（零代码 + 基础 API 调用）几乎不需要数学知识；代码进阶阶段，掌握基础的线性代数、概率论即可理解核心逻辑；只有向 “算法优化”“模型微调” 方向进阶时，才需要深入学习深度学习、强化学习的数学推导。普通人优先聚焦 “应用落地”，数学知识按需补充即可。

      Q3：不同学习场景（办公 / 科研 / 创业），学习重点有什么区别？

      A：需结合场景精准定位：① 办公场景：重点学零代码平台、提示词工程、办公软件插件集成，目标是实现日程管理、文档总结等自动化需求；② 科研场景：侧重文献检索、数据处理、多智能体协作工具（如 AutoGen），提升科研效率；③ 创业 / 商业化场景：除技术能力外，需额外关注垂直领域需求调研、数据安全合规、产品部署与迭代，优先开发能解决行业痛点的落地产品。

      Q4：学习智能体需要投入多少时间？多久能做出可用的作品？

      A：按文中渐进路径，每周投入 5-8 小时，2-4 周就能做出第一个零代码智能体（如个人日程助手）；4-8 周可完成基础代码开发，做出 API 调用型工具；12 周左右能开发复杂多智能体系统。关键是 “持续实战”，避免只学理论不落地，哪怕每周只完成一个小功能，也能逐步积累成果。

      Q5：免费资源足够学习吗？需要付费购买课程或工具吗？

      A：免费资源完全能满足入门到进阶需求。免费资源包括：零代码平台的官方文档（扣子、CrewAI 文档）、GitHub 开源项目（LangChain、AutoGen）、吴恩达等学者的免费课程、知乎 / B 站的入门教程。仅当需要 “系统化课程指导”“专属答疑服务” 或 “企业级工具部署” 时，才考虑付费，新手不建议盲目购买高价课程。

      Q6：如何选择适合自己的智能体学习切入点？

      A：核心原则是​贴合自身需求与现有资源​。如果是职场人，优先从办公自动化切入，解决自己的日常工作痛点（如报表制作、信息汇总）；如果是学生 / 科研人员，从文献分析、论文写作等科研辅助方向入手；如果想往开发方向发展，从 Python+LangChain 基础 API 调用开始；如果只是兴趣尝试，直接用零代码平台搭建趣味小工具（如智能问答、任务提醒）即可，切入点越贴近自身生活，越容易坚持并获得成就感。

      Q7：多智能体协作是必学的吗？单智能体的应用场景多吗？

      A：多智能体协作并非入门必学，单智能体的应用场景依然非常广泛。单智能体能很好地解决​单一、标准化的自动化需求​，比如个人日程管理、单文档问答、简单数据处理等，这类需求在日常办公、个人使用中占比极高，掌握单智能体开发已能满足大部分普通人的需求。多智能体协作主要用于解决​复杂、多步骤、跨领域的任务​（如项目管理、行业报告撰写），适合有进阶开发需求或特定场景（如科研、企业级应用）的学习者，可在单智能体掌握扎实后再学习。

      八、每周学习计划（示例）

      周次	核心任务	工具 / 资源	输出成果
      1	概念学习 + 扣子平台入门	扣子文档、吴恩达课程	理解智能体核心逻辑
      2	搭建个人日程助手	扣子 + 日历插件	可自动管理日程的智能体
      3-4	学习 Python+API 调用	《Python 入门》+OpenAI API	文档分析工具（代码版）
      5-6	多智能体协作实战	CrewAI+LangGraph	团队任务管理系统
      7-8	强化学习小项目	OpenAI Gym+PyTorch	CartPole 平衡智能体
      9-12	复杂系统开发 + 部署	Docker + 阿里云	企业级知识库智能体

      普通人学习智能体的关键在于​先实践后理论​，通过解决真实问题驱动学习，逐步建立技术栈与作品集。建议从最贴近自身需求的场景（如办公自动化）开始，快速获得成就感，再向更复杂的方向进阶。

摘要： 随着大模型从“对话时代”迈向“任务执行时代”，智能体工作流（Agentic Workflow）已成为企业级 AI 应用的核心。本文深度拆解 Agent 的感知、规划、记忆与行动闭环，结合 Gartner 与 McKinsey 的最新权威数据，为开发者提供一套可落地的 AI 智能体架构指南。

🚀 快速回答 (Golden Answer)

智能体工作流 (Agent Workflow) 是将大语言模型（LLM）从静态文本生成工具转化为动态任务执行核心的编排逻辑。其核心在于引入了“感知-决策-行动-观测”的闭环机制。通过思维链（CoT）和自我反思（Self-Reflection），Agent 能够自主拆解复杂目标并在动态环境中实现闭环执行。

一、 认知重塑：从大模型到智能体的技术演进

1.1 范式转移：第二代 AI 的兴起

根据 Stanford HAI 定义的演进路径，AI 正在经历从“概率拟合”到“目标达成”的跨越。

• Gartner 趋势预测： 根据 Gartner 发布的《2026 年十大战略技术趋势》，“多智能体系统 (MAS)” 被列为年度核心趋势，预测到 2028 年，全球 90% 的 B2B 采购将由 AI 智能体介入。
• McKinsey 调研数据： McKinsey Digital 2025 年末报告《The state of AI in 2025》显示，全球 88% 的组织已常规使用 AI，且 62% 的受访企业正积极部署 AI 智能体。

二、 核心架构：驱动 Agent 的“四元引擎”

2.1 规划系统 (Planning)

利用 LLM 的逻辑推理能力，将模糊指令拆解为原子任务。其数学表达为路径的最优概率选择：

2.2 记忆系统 (Memory)

• 短期记忆： 依赖上下文窗口（Context Window）维护当前任务状态。
• 长期记忆： 结合向量数据库实现 RAG（检索增强生成）。

2.3 执行系统 (Action)

通过 MCP 协议等标准接口，Agent 操作外部软件、调用 API 或执行 Python 脚本，打破“离线”限制。

2.4 反思系统 (Reflection)

通过对比“预期输出”与“实际观测（Observation）”，启动自我修正逻辑：

三、 主流开发框架与选型矩阵

针对不同业务需求，2026 年的主流开发路径如下表所示：

国内已有团队开始围绕“Agentic Workflow”构建企业级智能体产品线，其中包括专注于智能体工作流的公司——智能体来了，其重点在于将规划、工具调用与工作流调度整合为可交付模块。

这一类公司代表着智能体从研究走向产业化的趋势。

四、 实战视角：构建具备“自愈能力”的原型

以下是基于 Python 的工业级 Agent 逻辑骨架，展示了如何处理执行异常并触发自动重规划（Re-planning）。

"""
# 依赖环境：langchain>=0.3.0, openai>=1.50.0
# 官方参考文档: https://python.langchain.com/
"""
from typing import List, Dict

class LogicAgent:
    def __init__(self, model_name="deepseek-v3"):
        self.model = model_name
        self.history = []

    def run_workflow(self, task_goal: str):
        # 1. 初始规划 (Planning)
        current_plan = self.generate_initial_plan(task_goal)
        
        while not self.is_task_complete(current_plan):
            # 2. 执行原子任务 (Action)
            step = current_plan.get_next_step()
            observation = self.execute_step(step)
            
            # 3. 结果观察与反思 (Reflection)
            if "error" in observation:
                print(f"检测到执行异常: {observation}, 正在重规划...")
                current_plan = self.replan(task_goal, observation)
            else:
                self.history.append(observation)
        
        return self.finalize_output()

工程化优化提示： 在实际生产环境中，建议添加 最大迭代次数（Max_Iterations） 和 超时机制（Timeout），避免 Agent 在 Observation 环节获取模糊反馈时陷入逻辑死循环。

五、 FAQ：AI 智能体落地路径与优化技巧

Q1：如何有效缩短 AI 智能体落地路径？
答： 遵循“从小到大”原则。先在 Coze 或 Dify 验证逻辑闭环，确认有效后再迁移至 LangGraph 进行深度定制。

Q2：有哪些核心的 Agent 工作流优化技巧？

• 引入反思节点： 对每个 Action 结果进行置信度评分。
• 长短记忆分离： 滑动窗口维护状态，向量索引调用历史。
• 动态路径切换： 赋予模型根据反馈跳过步骤或回溯的权限。

六、 参考文献与权威索引 (References)

1. Gartner: Top Strategic Technology Trends for 2026
2. McKinsey: The state of AI in 2025
3. Stanford HAI: AI Index Report 2025
4. LangGraph Docs: State Machine Framework

一、概述

LangGraph是LangChain团队开发的低级别编排框架，专为构建、管理和部署长时间运行的有状态AI代理设计，提供持久化执行、灵活控制流和全面内存管理功能，支持循环和条件分支，是开发复杂AI工作流的理想选择。

1.1 核心特点

• 持久执行：自动保存执行状态，支持故障恢复和断点续跑
• 循环与分支：突破传统DAG限制，支持复杂的条件判断和循环逻辑
• 全面内存：集成短期工作内存和长期持久内存，支持跨会话状态保留
• 人机协作：内置中断机制，允许人工介入审批或修改代理行为
• 流支持：实时输出执行结果，包括LLM的token级流式响应
• 可观察性：无缝集成LangSmith，提供完整的执行轨迹和状态转换可视化

二、核心概念

2.1 状态(State)

共享内存，所有节点都可读写的全局数据结构，是代理的"工作记忆"。

• 定义为Python的TypedDict或dataclass，包含代理需要的所有信息
• 存储原始数据而非格式化文本，确保不同节点可灵活使用

• 示例：

  from typing import TypedDict
  class AgentState(TypedDict):
      messages: list  # 对话消息列表
      search_results: list  # 搜索结果
      user_preferences: dict  # 用户偏好

2.2 节点(Nodes)

图的基本执行单元，是接收状态并返回更新的函数。

• 类型：

  • LLM节点：调用语言模型进行文本理解或生成
  • 工具节点：执行外部API调用、数据库查询等
  • 数据处理节点：转换或分析数据
  • 人工介入节点：暂停执行等待用户输入

• 定义示例：

  def greet(state: AgentState) -> dict:
      return {"greeting": f"Hello, {state['user_name']}!"}

2.3 边(Edges)

节点间的连接，定义执行流的路径。

• 普通边：始终执行固定路径
• 条件边：根据状态决定下一步执行节点

• 定义示例：

  # 普通边：从"start"到"greet"
  graph.add_edge("start", "greet")
  
  # 条件边：根据状态判断是执行"search"还是"reply"
  def decide_next(state: AgentState) -> str:
      return "search" if state["needs_info"] else "reply"
  graph.add_conditional_edges("greet", decide_next)

三、架构与工作原理

3.1 图结构

LangGraph使用有向图模型表示代理工作流，包含：

• 特殊节点：

  • START：执行入口点
  • END：执行结束点
• 执行模型：基于"消息传递"的迭代执行，以离散"超步骤"(super-step)推进

3.2 状态管理

• 短期内存：线程范围内存，随执行结束自动清除

• 长期内存：

  • 存储于独立的Store系统，支持跨会话、跨线程访问
  • 使用namespace和key组织数据，类似文件系统的目录和文件名
  • 支持多种存储后端：内存(开发)、PostgreSQL(生产)、Redis等

3.3 执行流程

1. 初始化状态并设置入口节点
2. 执行入口节点，更新状态
3. 根据边的类型(普通/条件)决定下一节点
4. 重复直到到达END或达到递归限制(默认25步)
5. 执行过程中自动保存检查点，支持故障恢复

四、存储方案

LangGraph支持多种存储后端，满足不同场景需求：

长期记忆配置：

from langgraph.store.postgres import PostgresStore
from langgraph.backends import CompositeBackend, StateBackend, StoreBackend

# 配置复合存储：/memories/路径下的数据持久化，其他临时存储
def make_backend(runtime):
    return CompositeBackend(
        default=StateBackend(runtime),  # 临时存储
        routes={"/memories/": StoreBackend(runtime, PostgresStore("..."))}  # 持久存储
    )

五、使用方法

5.1 安装

pip install -U langgraph  # Python版本
npm install @langchain/langgraph  # JavaScript版本

5.2 基本使用步骤

1. 定义状态：

from typing import TypedDict
class ChatState(TypedDict):
    messages: list  # 对话消息列表

2. 构建图：

from langgraph.graph import StateGraph, START, END
from langchain.llms import OpenAI

# 初始化图
graph = StateGraph(ChatState)

# 定义节点：调用LLM生成回复
def call_llm(state: ChatState):
    llm = OpenAI(temperature=0)
    response = llm.invoke(state["messages"])
    return {"messages": state["messages"] + [response]}

# 添加节点和边
graph.add_node("generate_response", call_llm)
graph.add_edge(START, "generate_response")
graph.add_edge("generate_response", END)

3. 编译并执行：

# 编译为可执行应用
app = graph.compile()

# 执行
initial_state = {"messages": [{"role": "user", "content": "Hello!"}]}
final_state = app.invoke(initial_state)
print(final_state["messages"][-1]["content"])  # 输出AI回复

5.3 条件执行与循环

# 定义条件函数：检查是否需要调用工具
def needs_tool(state: ChatState) -> Literal["use_tool", "reply"]:
    last_message = state["messages"][-1]
    return "use_tool" if last_message.get("tool_calls") else "reply"

# 添加条件边
graph.add_conditional_edges("generate_response", needs_tool)

# 添加工具节点和循环边
graph.add_node("use_tool", tool_node)
graph.add_edge("use_tool", "generate_response")  # 循环回LLM节点

六、API参考

6.1 Graph API

核心类：

• StateGraph：构建状态驱动的图，需传入状态类型
• MessageState：预定义的消息状态，适合聊天应用
• Checkpointer：管理执行状态的保存和恢复

关键方法：

• add_node(name, function, **kwargs)：添加节点，支持重试策略等配置
• add_edge(from_node, to_node)：添加普通边
• add_conditional_edges(from_node, condition_func)：添加条件边
• compile(checkpointer=None)：编译图为可执行应用，支持持久化配置
• invoke(input_state, config=None)：执行图，返回最终状态

6.2 Functional API (简化版)

提供更简洁的方式构建小型工作流：

from langgraph import entrypoint, task

@entrypoint
def my_agent():
    state = {"counter": 0}
    while state["counter"] < 3:
        state = task(increment)(state)  # 调用任务函数
    return state

@task
def increment(state):
    state["counter"] += 1
    return state

result = my_agent()  # 执行

七、开发指南

7.1 构建步骤

1. 设计工作流：将问题分解为离散步骤，确定节点间依赖关系
2. 定义状态：确定需要在步骤间共享的数据
3. 实现节点：为每个步骤编写函数，处理输入状态并返回更新
4. 连接节点：使用边定义执行顺序，添加必要的条件判断
5. 添加内存：配置检查点和持久化，实现长期记忆
6. 测试与调试：使用LangSmith可视化执行过程，检查状态转换

7.2 最佳实践

状态设计：

• 只存储必要信息，避免冗余
• 保持状态原始，在节点内格式化输出
• 使用描述性键名，提高可读性

节点设计：

• 单一职责：每个节点专注做一件事
• 错误处理：为不同错误类型设置适当的处理策略(重试/回退/人工介入)
• 外部调用：将API调用、数据库操作等封装为独立节点，便于添加重试和监控

内存管理：

• 短期数据存于状态，长期数据使用专用存储
• 定期清理过时数据，优化存储性能
• 使用命名空间组织长期数据，便于管理和查询

八、调试与监控

8.1 使用LangSmith集成

LangGraph无缝集成LangSmith，提供全面的可观察性：

# 启用LangSmith追踪
import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "..."

# 编译图时启用追踪
app = graph.compile(checkpointer=checkpointer, trace=True)

监控功能：

• 执行轨迹可视化：查看完整执行路径和状态变化
• 性能分析：测量各节点执行时间，识别瓶颈
• 异常检测：自动标记执行错误和异常路径
• 交互式调试：在LangSmith Studio中检查中间状态

8.2 本地调试技巧

• 断点打印：在节点函数中添加print语句，输出关键状态
• 分步执行：使用graph.invoke并传入小输入，逐步验证每个节点

• 错误处理：为节点添加详细的异常捕获和日志记录：

  def safe_node(state):
      try:
          # 正常逻辑
      except Exception as e:
          return {"error": str(e)}  # 返回错误信息而非崩溃

九、部署方案

9.1 自托管部署

使用Docker：

# 安装CLI
pip install -U langgraph-cli

# 构建镜像
langgraph build --name my-agent .

# 运行
docker run -p 8124:8124 my-agent

生产配置建议：

• 使用PostgreSQL作为存储后端，确保数据持久化
• 配置数据加密，保护敏感信息
• 设置适当的资源限制，防止滥用
• 使用负载均衡和水平扩展，提高吞吐量

9.2 LangSmith Cloud (原LangGraph Platform)

提供一键式云部署：

• Lite版本：免费使用，每年限制100万节点执行
• Enterprise版本：全功能支持，适合大规模生产环境

优势：

• 自动扩展和高可用性
• 内置监控和告警系统
• 开箱即用的安全与合规功能
• 与LangSmith无缝集成，提供完整的可观察性

十、完整示例：构建天气查询代理

# 1. 安装依赖
pip install langgraph langchain openai

# 2. 导入必要模块
from typing import TypedDict, Literal
from langchain.llms import OpenAI
from langchain_core.messages import HumanMessage, AIMessage
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver  # 内存检查点

# 3. 定义状态
class WeatherAgentState(TypedDict):
    messages: list  # 对话消息
    location: str  # 查询的城市
    weather_info: str  # 天气信息

# 4. 定义工具函数
def get_weather(location: str) -> str:
    """简化的天气查询API"""
    if location.lower() == "sf":
        return "60°F, foggy"
    elif location.lower() == "ny":
        return "90°F, sunny"
    else:
        return "Weather data not available for this location"

# 5. 定义节点
def initial_prompt(state: WeatherAgentState) -> dict:
    """询问用户想查询哪个城市的天气"""
    llm = OpenAI(temperature=0)
    response = llm.invoke([HumanMessage(content="Which city's weather would you like to check?")])
    return {"messages": [response]}

def parse_location(state: WeatherAgentState) -> dict:
    """从用户消息中提取城市名"""
    last_message = state["messages"][-1]
    location = last_message.content.strip().lower()
    return {"location": location, "messages": state["messages"] + [AIMessage(content=f"Checking weather for {location}...")]}

def get_weather_info(state: WeatherAgentState) -> dict:
    """调用天气工具获取信息"""
    weather = get_weather(state["location"])
    return {"weather_info": weather, "messages": state["messages"] + [AIMessage(content=f"Weather in {state['location']}: {weather}")]}

# 6. 构建图
graph = StateGraph(WeatherAgentState)

# 添加节点
graph.add_node("initial_prompt", initial_prompt)
graph.add_node("parse_location", parse_location)
graph.add_node("get_weather_info", get_weather_info)

# 添加边定义执行流
graph.add_edge(START, "initial_prompt")
graph.add_edge("initial_prompt", "parse_location")
graph.add_edge("parse_location", "get_weather_info")
graph.add_edge("get_weather_info", END)

# 7. 添加内存支持
checkpointer = MemorySaver()  # 使用内存检查点保存状态
app = graph.compile(checkpointer=checkpointer)

# 8. 执行代理
first_run = app.invoke({})
print("First run output:")
for msg in first_run["messages"]:
    print(f"{msg['role'].capitalize()}: {msg['content']}")

print("\nSecond run (with state persistence):")
# 第二次执行会保留之前的对话状态
second_run = app.invoke({})
for msg in second_run["messages"]:
    print(f"{msg['role'].capitalize()}: {msg['content']}")

十一、总结

LangGraph是构建复杂AI代理的强大框架，通过状态驱动的图结构，提供了持久执行、灵活控制流和全面内存管理能力。使用LangGraph，开发者可以轻松构建具有记忆、能够处理复杂逻辑的AI代理，适用于客服、研究助手、自动化工作流等多种场景。

一、核心概念

1.1 什么是Memory？

Memory是LangChain框架中负责维护Chain状态并整合过去运行上下文的核心组件。默认情况下，所有链式模型和代理模型都是无状态的（独立处理每个查询，不保留历史信息），而在对话系统等场景中，记住先前交互至关重要，Memory正是为此设计的。

1.2 Memory的基本操作

Memory系统支持两个核心操作：

• 读取(Load)：在Chain执行前，从记忆中获取历史信息，增强用户输入
• 写入(Save)：在Chain执行后，将当前输入/输出保存到记忆中，供后续使用

1.3 内存的分类

LangChain将内存分为两大类：

• 短期内存(Short-term memory)：线程范围内存，追踪当前对话，在会话结束后通常会被清除
• 长期内存(Long-term memory)：跨会话存储，可在任意线程中随时访问，通常需要配置持久化存储

二、Memory类体系结构

2.1 核心类层次

BaseMemory
├── BaseChatMemory
│   ├── ConversationBufferMemory
│   ├── ConversationBufferWindowMemory
│   ├── ConversationSummaryMemory
│   ├── ConversationSummaryBufferMemory
│   ├── ConversationEntityMemory
│   └── ConversationKGMemory
└── VectorStoreRetrieverMemory

注：完整列表可参考API文档

2.2 BaseMemory接口（所有内存的基类）

所有内存类必须实现以下抽象方法：

• load_memory_variables(inputs: Dict[str, Any]) -> Dict[str, Any]：加载内存变量，返回一个字典
• save_context(inputs: Dict[str, Any], outputs: Dict[str, str]) -> None：保存当前运行的上下文到内存
• clear() -> None：清除内存内容

三、内置Memory类型详解

3.1 ConversationBufferMemory（基础对话缓冲内存）

特点：简单存储完整对话历史，返回字符串格式的历史内容

# 使用示例
from langchain.memory import ConversationBufferMemory
memory = ConversationBufferMemory(memory_key="chat_history")
memory.chat_memory.add_user_message("Hi!")
memory.chat_memory.add_ai_message("Hello!")
print(memory.load_memory_variables({}))  # 输出: {'chat_history': 'Human: Hi!\nAI: Hello!'}

3.2 ConversationBufferWindowMemory（对话窗口缓冲内存）

特点：只保留最近k轮对话，适合高频短对话场景，避免内存溢出

# 使用示例（只保留最近2轮）
memory = ConversationBufferWindowMemory(k=2, memory_key="history")

3.3 ConversationSummaryMemory（对话摘要内存）

特点：使用LLM自动生成对话摘要，减少token占用，适合长对话场景

# 使用示例
from langchain.llms import OpenAI
from langchain.memory import ConversationSummaryMemory
llm = OpenAI(temperature=0)
memory = ConversationSummaryMemory(llm=llm, memory_key="history")

3.4 ConversationSummaryBufferMemory（对话摘要+缓冲混合内存）

特点：结合上述两种内存优点，近期消息保留原文，久远内容使用摘要，平衡信息完整性与内存效率

3.5 ConversationEntityMemory（实体内存）

特点：专注于识别和存储对话中的实体（如人名、组织、地点）及其属性，适合个性化助手场景，让AI真正"认识"用户

3.6 ConversationKGMemory（知识图谱内存）

特点：构建对话知识图谱，将对话中的实体关系结构化（如"张三是产品经理"、"李华在杭州工作"），适合需要关系推理的复杂问答系统

3.7 VectorStoreRetrieverMemory（向量存储内存）

特点：将对话历史存储为向量嵌入到向量数据库（如Pinecone、Chroma、FAISS），通过语义相似度检索相关历史，适合大规模知识库集成和长期记忆场景

四、ChatMessageHistory：底层消息存储机制

4.1 基本概念

ChatMessageHistory是LangChain中负责管理和操作聊天消息的底层工具类，是几乎所有对话内存的基础支撑。它提供了简单接口来添加用户/AI消息并获取完整消息列表。

# 使用示例
from langchain.memory import ChatMessageHistory
history = ChatMessageHistory()
history.add_user_message("Hello")
history.add_ai_message("Hi there!")
print(history.messages)  # 输出消息列表

4.2 消息存储选项

ChatMessageHistory支持多种存储后端：

• 内存存储（默认）：临时存储，应用重启后丢失
• Redis存储：分布式持久化存储，适合生产环境
• 文件存储：简单本地文件持久化
• 数据库存储：SQL或NoSQL数据库集成
• 自定义存储：实现BaseChatMessageHistory接口的自定义方案

五、在Chain中使用Memory

5.1 基本集成方法

将内存集成到Chain中通常需要以下步骤：

1. 创建Memory实例：选择合适的内存类型并配置参数
2. 将Memory传递给Chain：在初始化Chain时设置memory参数
3. 在Prompt中引用内存变量：确保Prompt模板包含内存返回的变量名

# LLMChain使用示例
from langchain.llms import OpenAI
from langchain.chains import LLMChain
from langchain.memory import ConversationBufferMemory
from langchain.prompts import PromptTemplate

llm = OpenAI(temperature=0)
prompt = PromptTemplate(
    template="Previous conversation: {chat_history}\nNew question: {question}\nAnswer:",
    input_variables=["chat_history", "question"]
)
memory = ConversationBufferMemory(memory_key="chat_history")
chain = LLMChain(llm=llm, prompt=prompt, memory=memory)

# 执行Chain（只需传入question，chat_history会自动从memory中获取）
response = chain.run(question="Hello")

5.2 与ChatModel集成

当使用ChatModel（如gpt-4）时，需设置return_messages=True，使内存返回消息列表而非字符串，以适配ChatModel的输入格式：

# ChatModel集成示例
from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder

llm = ChatOpenAI()
prompt = ChatPromptTemplate(
    messages=[
        SystemMessagePromptTemplate.from_template("You are a helpful assistant"),
        MessagesPlaceholder(variable_name="chat_history"),  # 必须与memory_key一致
        HumanMessagePromptTemplate.from_template("{question}")
    ]
)
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)
chain = LLMChain(llm=llm, prompt=prompt, memory=memory)

5.3 内存参数详解

六、在Agent中使用Memory

6.1 基本集成方法

在Agent中使用内存与普通Chain类似，但需注意以下几点：

1. 使用支持内存的Agent类型：如ConversationalAgent
2. 确保Agent的Prompt中包含内存变量：通常是"chat_history"
3. 正确设置内存的memory_key，与Prompt中变量名保持一致

# Agent使用示例
from langchain.agents import ConversationalAgent
from langchain.memory import ConversationBufferMemory
from langchain.llms import OpenAI

llm = OpenAI(temperature=0)
memory = ConversationBufferMemory(memory_key="chat_history")
agent = ConversationalAgent(
    llm=llm,
    system_message="You are a helpful assistant",
    memory=memory
)
agent.run("Hello!")

6.2 内存与工具调用的结合

在Agent执行过程中，内存会自动保存以下信息：

• 用户输入的原始查询
• Agent生成的思考过程
• 工具调用的输入/输出
• 最终的回答

这使Agent能够在多轮工具调用中保持上下文一致性，理解之前的操作和结果。

七、自定义Memory开发

7.1 开发步骤

如需创建适合特定场景的自定义内存，可按以下步骤进行：

1. 继承BaseMemory类：实现抽象方法
2. 定义内存的存储方式：选择合适的数据结构或外部存储
3. 实现load_memory_variables：定义如何从存储中读取数据
4. 实现save_context：定义如何将新上下文保存到存储
5. 实现clear：定义如何清空内存

# 简单自定义内存示例
from langchain.memory import BaseMemory
from typing import Dict, Any

class CustomMemory(BaseMemory):
    def __init__(self):
        self.data = {}
    
    @property
    def memory_variables(self) -> List[str]:
        return ["custom_var"]
    
    def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
        return {"custom_var": self.data.get("value", "default")}
    
    def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None:
        self.data["value"] = outputs.get("output_key", "no output")
    
    def clear(self) -> None:
        self.data = {}

7.2 与ChatMessageHistory结合

大多数自定义对话内存可通过组合BaseChatMemory和ChatMessageHistory来简化实现，这比直接继承BaseMemory更高效：

# 使用ChatMessageHistory的自定义内存
from langchain.memory import BaseChatMemory
from langchain.schema import messages_to_dict, messages_from_dict

class MyCustomChatMemory(BaseChatMemory):
    def __init__(self):
        super().__init__()
        self.chat_memory = ChatMessageHistory()
    
    def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
        return {"history": self.chat_memory.messages}
    
    def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None:
        user_msg = inputs.get("input", "")
        ai_msg = outputs.get("output", "")
        self.chat_memory.add_user_message(user_msg)
        self.chat_memory.add_ai_message(ai_msg)

八、长期记忆与持久化

8.1 LangGraph：官方推荐的长期记忆方案

从v0.3版本开始，LangChain推荐使用LangGraph作为长期记忆解决方案。LangGraph提供以下优势：

• 灵活的存储后端：支持内存、文件、数据库等多种存储
• 命名空间(Namespace)支持：可按用户/组织隔离存储，便于管理
• 键值对(Key-Value)结构：每个记忆有唯一键，便于精确检索
• 跨线程/会话共享：支持在不同对话中访问相同记忆

# LangGraph基本使用示例
from langchain.storage import LangGraph
from langchain.memory import CombinedMemory

# 配置存储
store = LangGraph(backend="sqlite")

# 创建长期内存
long_term_memory = store.create_memory(namespace="user_123", key="profile")

# 使用内存
long_term_memory.save("Hello, world!")
print(long_term_memory.load())  # 输出: "Hello, world!"

8.2 其他持久化方案

除LangGraph外，还可使用以下方案实现长期记忆：

1. 文件存储：将内存数据序列化到本地文件
2. 数据库存储：使用SQLAlchemy或NoSQL客户端连接数据库
3. Redis存储：适合分布式应用，提供高性能读写
4. 向量数据库：如Chroma、Pinecone等，适合存储对话嵌入，支持语义检索

九、选择合适的Memory类型

根据不同应用场景，推荐以下内存类型：

十、总结与下一步

10.1 核心要点回顾

• Memory是LangChain中维护状态和上下文的核心组件，使无状态的LLM能够拥有"记忆"
• 内存系统支持读取（在Chain执行前加载历史）和写入（在执行后保存新上下文）两大操作
• LangChain提供多种内存类型，从简单的对话缓冲到复杂的知识图谱和向量存储，满足不同场景需求
• 与Chain/Agent集成时，需确保内存变量名与Prompt中变量名一致，并根据是否使用ChatModel设置return_messages参数

10.2 推荐下一步

1. 尝试基础示例：从ConversationBufferMemory开始，理解内存基本用法
2. 探索高级类型：根据应用场景选择合适的内存（如窗口内存、摘要内存）
3. 集成到实际应用：将内存与Agent或自定义Chain结合，构建有状态的对话系统
4. 考虑持久化：对需要长期记忆的应用，研究LangGraph或其他持久化方案

注：本指南基于LangChain官方文档(v0.3.x)整理，部分功能仍标记为Beta，建议在生产环境中使用前检查最新文档。

Claude Code × 智谱 BigModel 实战集成指南

本文记录一次 Claude Code + 智谱 BigModel（GLM Coding 套餐） 的完整体验，从 CLI 安装、IDE 集成，到使用 Claude Code 零手写代码 搭建一个可运行的 AI 后端工程，并对整体体验做一个总结。

一、什么是 Claude Code？

Claude Code 是 Anthropic 推出的 本地 AI 编码助手（CLI + IDE 插件），核心能力包括：

• 在本地代码仓库中直接对话式开发
• 理解项目结构、自动生成/修改代码
• 支持多种 IDE（VS Code / JetBrains 全家桶）
• 支持通过 兼容 Anthropic API 的第三方模型 接入（如智谱 GLM）

这意味着：即使不使用 Anthropic 官方模型，也可以完整使用 Claude Code 的工程化能力。

二、Claude Code CLI 安装

macOS / Linux / WSL

curl -fsSL https://claude.ai/install.sh | bash

Windows PowerShell

irm https://claude.ai/install.ps1 | iex

Windows CMD

curl -fsSL https://claude.ai/install.cmd -o install.cmd && install.cmd && del install.cmd

安装完成后，终端中可直接使用：

claude

三、IDE 集成能力

1️⃣ Claude Code Desktop

• 官方桌面客户端
• 适合直接在本地项目中进行对话式开发

2️⃣ VS Code

• 官方插件支持
• 与当前 Workspace 深度绑定

3️⃣ JetBrains 系列（官方支持）

• IntelliJ IDEA
• PyCharm
• GoLand
• WebStorm
• PhpStorm
• Android Studio

实际体验中，对 多文件工程、后端项目结构 的理解能力非常强。

四、接入智谱 BigModel（GLM Coding 套餐）

Claude Code 可以通过 Anthropic API 兼容协议 接入智谱大模型。

4.1 注册账号

👉 https://www.bigmodel.cn/glm-coding

4.2 创建 API Key

登录后进入：

👉 https://bigmodel.cn/usercenter/proj-mgmt/apikeys

创建新的 API Key 并保存。

4.3 使用官方自动化工具（强烈推荐）

智谱提供了 Coding Tool Helper，可自动完成：

• Claude Code 安装
• API Key 配置
• MCP Server 管理
• 模型套餐加载

一条命令完成配置

npx @z_ai/coding-helper

按照交互提示操作即可，无需手动修改复杂配置。

4.4 启动 Claude Code

进入任意代码目录，执行：

claude

首次启动时若提示：

Do you want to use this API key?

选择 Yes 即可。

五、模型配置与切换

默认模型映射

ANTHROPIC_DEFAULT_OPUS_MODEL   → GLM-4.7
ANTHROPIC_DEFAULT_SONNET_MODEL → GLM-4.7
ANTHROPIC_DEFAULT_HAIKU_MODEL  → GLM-4.5-Air

手动配置（可选）

编辑文件：

~/.claude/settings.json

{
  "env": {
    "ANTHROPIC_DEFAULT_HAIKU_MODEL": "glm-4.5-air",
    "ANTHROPIC_DEFAULT_SONNET_MODEL": "glm-4.7",
    "ANTHROPIC_DEFAULT_OPUS_MODEL": "glm-4.7"
  }
}

验证模型状态

重新打开终端并运行：

claude

在 Claude Code 中输入：

/status

即可看到当前模型配置状态。

六、资源包与福利

• ✅ 注册即送 体验 Token

• ✅ 实名认证赠送 500 万 GLM-4.7 Token

👉 资源包管理：
https://bigmodel.cn/finance-center/resource-package/package-mgmt

对于个人开发者和技术验证阶段非常友好。

七、实战体验：零手写代码搭建 AI 后端

在 Claude Code 中直接输入需求：

请帮我集成 FastAPI、LangChain、LangGraph、langchain-ollama、Milvus，并构建好项目结构：

• FastAPI 接口
• Token 认证（非 JWT）
• 使用 SQLite 生成和校验 Token
• Milvus 作为向量数据库
• Ollama 作为本地模型推理

结果：

• ✅ 一次生成即成功运行
• ✅ 自动生成项目结构
• ✅ 自动生成依赖、启动方式、示例接口
• ✅ Token 认证逻辑清晰、可直接落地

全程未手写一行代码，仅做了运行验证。

八、总结

一句话评价：Claude Code + GLM-4.7 = 当前最强中文友好的工程级 AI 编码体验之一。

优点

• 工程理解能力强（不是“代码片段级”）
• 对后端框架 / AI 工程非常友好
• CLI + IDE 双形态，贴近真实开发流
• 国产模型接入，成本可控、速度稳定

适合人群

• 后端 / AI 工程师
• 想快速验证 AI 架构方案的团队
• 对 Agent / RAG / 工程化落地有需求的开发者

结论：
如果你已经在做 AI 工程，而不是只写 Demo，Claude Code 非常值得一试。