从零构建 AI 智能体:2026 年全面实战指南
前言:为什么 2026 年是构建 AI 智能体的最佳时机
AI 智能体(Agent)是大模型应用落地的核心方向。2026 年,随着 LangChain、AutoGPT、CrewAI、LangGraph 等框架的成熟,以及大模型 API 成本持续下降、工具生态日趋完善,普通人也能从零搭建可用、稳定、可上线的 AI 智能体。本文将从基础概念、环境搭建、框架选择、核心组件开发、部署上线、性能优化到实战案例,提供一份完整的实战指南。
与两年前相比,2026 年构建智能体的门槛已经大幅降低。一方面,大模型已经具备更强的工具调用能力和更长的上下文窗口;另一方面,主流框架对规划、记忆、工具调用等能力的封装越来越成熟。曾经需要研究团队才能完成的多智能体协作系统,现在只需要一个开发者、一个代码编辑器,再加上清晰的架构设计,就能在数天内完成从原型到上线的全流程。
阅读建议:本文适合具备一定 Python 基础、希望从零构建 AI 智能体的开发者阅读。如果你已经熟悉 LangChain 的基础用法,可以直接跳到第四章「核心组件开发」和第八章「实战案例」。如果你刚接触智能体,建议按照章节顺序通读,并在每章末尾动手实践。
2026 年 AI 智能体生态的五个关键变化
| 变化 | 具体表现 | 对开发者的意义 |
|---|---|---|
| 框架走向成熟 | LangChain、LangGraph、CrewAI 稳定版本发布,文档和社区完善 | 开发成本降低,踩坑成本减少 |
| 大模型能力跃升 | 主流模型支持稳定的函数调用、复杂推理和更长上下文 | 智能体能够处理更复杂的多步任务 |
| 工具生态丰富 | 搜索、数据库、办公、代码执行等工具开箱即用 | 开发者无需重复造轮子 |
| 部署方式多样化 | Serverless、容器、边缘部署方案普及 | 从原型到上线路径更短 |
| 成本持续下降 | 大模型 API 价格逐年下降,开源模型能力增强 | 个人开发者也能承受长期运行成本 |
本文学习路线
为了帮助你建立完整的知识体系,本文按照下图所示的路线组织内容。图中的箭头表示推荐的学习顺序,你可以根据自身情况灵活调整。
flowchart TD
A[基础概念] --> B[环境搭建]
B --> C[框架选择]
C --> D[核心组件开发]
D --> E[部署上线]
E --> F[性能优化]
F --> G[实战案例]
G --> H[总结与进阶]
A --> A1[智能体定义]
A --> A2[核心架构]
A --> A3[应用场景]
C --> C1[框架对比]
C --> C2[方案选型]
D --> D1[工具函数]
D --> D2[记忆系统]
D --> D3[Agent 实现]
D --> D4[规划反思]
E --> E1[FastAPI 封装]
E --> E2[Docker 容器化]
E --> E3[云平台部署]
E --> E4[监控安全]
学完本文后你将掌握
- 理解智能体的核心原理:掌握规划、工具调用、记忆、反思四大核心组件的工作机制。
- 独立搭建开发环境:能够配置 Python 环境、API Key 和本地向量数据库。
- 完成框架选型:根据项目需求选择 LangChain、CrewAI、LangGraph 等合适框架。
- 开发完整智能体:实现工具函数、记忆系统、Agent 主循环,并完成基础测试。
- 完成上线部署:使用 FastAPI、Docker 和云平台将智能体发布为可用服务。
- 具备优化与排错能力:能够定位常见问题,并针对性能、成本和安全进行优化。
一、AI 智能体基础概念
1.1 什么是 AI 智能体
AI 智能体是一个能够感知环境、做出决策并执行动作的自主系统。与传统聊天机器人不同,智能体不仅会回答问题,还会主动规划任务、调用外部工具、保存记忆,并根据执行结果不断调整策略。简单来说,聊天机器人是「会说」,而智能体是「会做」。
一个完整的 AI 智能体通常具备以下四项核心能力:
- 自主规划:将复杂任务分解为可执行的子步骤。例如,用户要求「帮我调查某公司的竞品并生成报告」,智能体会先拆解为搜索信息、整理数据、对比分析、生成报告四个子任务。
- 工具调用:根据需要调用外部 API、搜索引擎、数据库、计算器、代码执行器等。工具让智能体突破语言模型的边界,获得实时信息和执行能力。
- 记忆系统:分为短期记忆(上下文窗口)和长期记忆(向量数据库或结构化存储)。短期记忆保证多轮对话连贯,长期记忆让智能体记住历史任务和用户偏好。
- 反思迭代:根据执行结果调整策略,实现自我优化。当某一步失败或结果不符合预期时,智能体会分析原因并重新规划。
下面这张表总结了智能体与传统程序、聊天机器人之间的本质区别,帮助你建立更清晰的认知。
| 能力维度 | 传统程序 | 聊天机器人 | AI 智能体 |
|---|---|---|---|
| 执行方式 | 固定规则,确定性强 | 单轮或多轮问答 | 多步自主规划、执行与验证 |
| 外部调用 | 一般不调用外部 API | 基本不调用工具 | 主动调用 API、数据库、搜索引擎等 |
| 记忆能力 | 无持久记忆 | 有限上下文窗口 | 短期加长期记忆,可跨会话积累 |
| 错误处理 | 崩溃或返回错误码 | 可能生成错误回答 | 能检测失败、重试或更换策略 |
| 任务范围 | 单一固定任务 | 单一问题回答 | 复杂任务端到端完成 |
1.2 智能体 vs 聊天机器人
聊天机器人和 AI 智能体都基于大语言模型,但二者的设计目标完全不同。聊天机器人侧重于自然对话和内容生成,而智能体侧重于任务执行和结果达成。理解这些差异有助于你在项目中选择正确的技术路线。
| 维度 | 聊天机器人 | AI 智能体 |
|---|---|---|
| 交互方式 | 单轮问答或多轮对话 | 多轮自主执行 |
| 工具使用 | 无或极少 | 可调用外部 API、数据库、搜索等 |
| 任务范围 | 单一问题回答、闲聊、写作 | 复杂任务端到端完成 |
| 记忆能力 | 有限上下文,对话结束即丢失 | 长期持久化记忆,可跨会话调用 |
| 自主性 | 被动响应,需用户逐条引导 | 主动规划,自动拆解任务并执行 |
| 输出形式 | 文本、代码、图片等生成内容 | 完成事务、操作外部系统、返回结果 |
| 典型场景 | 客服问答、写作助手、翻译 | 自动报表、数据检索、流程自动化、办公助手 |
需要注意的是,智能体和聊天机器人并非互斥关系。很多实际产品会将二者结合,例如客服系统先用聊天机器人承接高频问题,遇到复杂操作时再切换为智能体执行查询订单、修改信息等动作。
1.3 智能体的核心架构
目前业界最常用的智能体架构之一是 ReAct(Reasoning and Acting)模式。它的核心思想是交替进行「思考」和「行动」:模型先根据当前目标推理出下一步应该做什么,然后调用对应的工具执行动作,观察执行结果后再进行下一轮推理,直到任务完成。
下面是 ReAct 循环的完整流程:
flowchart TD
Start[用户输入任务] --> Think[模型思考与规划]
Think --> Decide{是否需要工具?}
Decide -- 是 --> Call[调用工具执行动作]
Call --> Observe[观察执行结果]
Observe --> Reflect{任务是否完成?}
Reflect -- 否 --> Think
Reflect -- 是 --> Answer[整理并输出最终答案]
Decide -- 否 --> Answer
Answer --> End[任务结束]
为了更直观地理解,我们以「查询北京今天天气并给出穿衣建议」这个任务为例,展示 ReAct 循环每一步的输入和输出。
| 循环步骤 | 阶段 | 模型或系统动作 | 输入 | 输出 |
|---|---|---|---|---|
| 1 | 思考 | 分析任务,判断需要查询天气 | 用户原始问题 | 「需要调用天气查询工具」 |
| 2 | 行动 | 调用天气查询工具 | 工具名和城市参数 | 天气 API 返回结果 |
| 3 | 观察 | 解析工具返回数据 | 天气 JSON 数据 | 「今天北京气温 5 到 12 摄氏度,有风」 |
| 4 | 思考 | 判断信息是否足够 | 天气数据 | 「信息足够,可以给出穿衣建议」 |
| 5 | 回答 | 生成最终回答 | 天气信息加推理 | 「今天北京较冷,建议穿厚外套」 |
除了 ReAct,智能体架构中还有 Plan-and-Execute(先规划后执行)、Reflexion(反思改进)等模式。对初学者而言,优先掌握 ReAct 已经能覆盖大多数实用场景。
1.4 智能体的典型应用场景
AI 智能体已经渗透到多个行业。了解不同场景中的实际用法,有助于你在学习过程中找到适合自己的练手项目。
| 行业 | 典型场景 | 智能体承担的任务 | 涉及的核心能力 |
|---|---|---|---|
| 内容创作 | 自动生成行业报告 | 检索资料、整理数据、生成分析报告 | 搜索工具、长期记忆、多步规划 |
| 电商零售 | 智能客服与导购 | 查询订单、推荐商品、处理退换货 | 数据库工具、对话记忆、权限控制 |
| 金融分析 | 自动研报与风险监控 | 抓取行情、计算指标、生成风险预警 | 数据工具、代码执行、定时任务 |
| 企业办公 | 会议纪要自动化 | 识别语音、提取待办、同步日历 | 音视频接口、办公软件工具 |
| 软件研发 | 智能编程助手 | 阅读代码库、修改 Bug、生成测试 | 代码执行、文件操作、上下文记忆 |
| 医疗健康 | 辅助问诊与健康提醒 | 收集症状、查询资料、生成建议 | 知识检索、对话记忆、安全策略 |
二、环境搭建
2.1 Python 开发环境
推荐使用 Python 3.10 或更高版本,配合 conda 进行环境管理。conda 能够隔离不同项目的依赖版本,避免全局环境污染。下面是创建专属开发环境的完整命令:
conda create -n agent-dev python=3.10
conda activate agent-dev
pip install langchain langchain-openai chromadb faiss-cpu
pip install fastapi uvicorn pydantic
pip install python-dotenv tiktoken
安装完成后,可以使用下面的命令验证关键依赖是否可用:
python -c "import langchain; print(langchain.__version__)"
python -c "import fastapi; print(fastapi.__version__)"
python -c "import chromadb; print(chromadb.__version__)"
如果三行命令都没有报错,说明基础环境已经准备完毕。接下来我们详细说明每个依赖包的作用,方便你理解为什么需要安装它们。
| 依赖包 | 作用 | 是否必需 | 备注 |
|---|---|---|---|
| langchain | 智能体开发主框架,提供工具、记忆、Agent 等核心组件 | 必需 | 保持版本更新以获得最新特性 |
| langchain-openai | OpenAI 模型集成包 | 按需 | 使用通义千问或 Claude 时替换为对应包 |
| chromadb | 本地向量数据库,用于长期记忆 | 推荐 | 轻量级,适合本地开发和原型 |
| faiss-cpu | 向量相似度检索库,支持高效查询 | 可选 | 与 Chroma 配合使用可提升性能 |
| fastapi | 将智能体封装为 HTTP 服务 | 上线必需 | 性能高、生态完善 |
| uvicorn | ASGI 服务器,运行 FastAPI 应用 | 上线必需 | 与 FastAPI 官方配套 |
| pydantic | 数据校验和序列化 | 上线必需 | FastAPI 的依赖包 |
| python-dotenv | 从 .env 文件加载环境变量 | 推荐 | 避免将 API Key 硬编码在代码中 |
| tiktoken | 计算 Token 数量,用于上下文管理 | 推荐 | 帮助控制成本、避免超窗 |
如果你不想使用 conda,也可以使用 Python 自带的 venv 模块创建虚拟环境。两者的差异会在 2.3 节详细对比。
2.2 API Key 准备
智能体需要调用大模型和外部服务,因此你需要提前准备好相应的 API 密钥。下面是常见的三类密钥:
- 大模型 API:OpenAI GPT-4 / Claude / 通义千问 / 文心一言等。
- 搜索 API:Tavily / SerpAPI,用于网络信息检索。
- 向量数据库:Chroma 本地运行即可,无需在线密钥;若使用 Pinecone、Weaviate 等云服务则需要额外申请。
不同大模型服务商在能力、价格和国内访问便利性上差异较大。下表列出了 2026 年常用的几个模型服务供你参考。
| 服务商 | 代表模型 | 优势 | 需要注意 | 适用场景 |
|---|---|---|---|---|
| OpenAI | GPT-4 系列 | 推理能力强、工具调用稳定、生态成熟 | 国内访问需要代理,费用相对较高 | 复杂推理、高质量智能体 |
| Anthropic | Claude 系列 | 长上下文、安全性好、代码能力强 | 部分地区访问受限 | 长文档处理、代码生成 |
| 阿里云 | 通义千问系列 | 国内访问快、中文能力强、价格友好 | 工具调用协议需查看最新文档 | 中文场景、国内部署 |
| 百度 | 文心一言系列 | 中文理解好、生态逐步完善 | 模型能力与 GPT-4 有差距 | 中文对话、内容生成 |
| 本地开源 | Qwen 开源、DeepSeek 等 | 数据私有、无调用费用 | 需要自备 GPU、部署运维成本高 | 数据敏感、离线环境 |
推荐使用 .env 文件集中管理所有密钥,避免硬编码在源码中。项目根目录创建 .env 文件:
OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxx
SERPAPI_API_KEY=xxxxxxxxxxxxxxxxxxxxxxxx
TAVILY_API_KEY=tvly-xxxxxxxxxxxxxxxxxxxxxx
MODEL_NAME=gpt-4o-mini
然后在 Python 代码中使用 python-dotenv 加载:
from dotenv import load_dotenv
import os
load_dotenv()
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
SERPAPI_API_KEY = os.getenv("SERPAPI_API_KEY")
TAVILY_API_KEY = os.getenv("TAVILY_API_KEY")
MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4o-mini")
这样做的好处是:密钥不会进入 Git 历史,切换密钥时只需修改 .env 文件,不需要改动业务代码。记得在 .gitignore 中添加 .env,防止密钥泄露。
2.3 虚拟环境与依赖管理最佳实践
虚拟环境是 Python 项目的基础设施,可以避免不同项目之间依赖版本冲突。conda 和 venv 是两种主流方案,下表对比了它们的特点。
| 对比项 | conda | venv |
|---|---|---|
| 安装方式 | 需安装 Anaconda 或 Miniconda | Python 3.3+ 内置,无需额外安装 |
| 环境隔离 | 同时隔离 Python 版本和依赖包 | 仅隔离依赖包,Python 版本需另装 |
| 包管理 | conda 加 pip 双通道 | 仅使用 pip |
| 环境体积 | 占用空间较大 | 轻量、占用空间小 |
| 适合场景 | 数据科学、多种 Python 版本并行 | Web 开发、轻量级项目 |
对于本文的智能体项目,两者都可以。如果你已经安装了 conda,优先使用 conda;如果希望环境更轻量,使用 venv 也完全够用。下面是 venv 的快速使用命令:
python -m venv agent-env
source agent-env/bin/activate # Windows 使用 agent-env\Scripts\activate
pip install -r requirements.txt
无论使用哪种虚拟环境,都建议用 requirements.txt 锁定依赖版本,保证团队协作和服务器部署时环境一致。生成依赖锁定文件的命令如下:
pip freeze > requirements.txt
一个完整的 requirements.txt 示例:
langchain==0.2.10
langchain-openai==0.1.8
chromadb==0.5.0
faiss-cpu==1.8.0
fastapi==0.111.0
uvicorn==0.30.1
pydantic==2.7.1
python-dotenv==1.0.1
tiktoken==0.7.0
requests==2.32.3
三、框架选择与对比
3.1 主流框架对比
2026 年,智能体框架已经非常丰富。每个框架都有其设计哲学和最佳使用场景。下面是四个最主流框架的横向对比。
| 框架 | 定位 | 核心特点 | 主要优势 | 主要局限 | 适用场景 |
|---|---|---|---|---|---|
| LangChain | 通用智能体开发框架 | 生态最完善,组件丰富,集成大量工具和模型 | 文档齐全、社区庞大、上手快 | 封装层次较多,复杂场景需理解内部机制 | 通用智能体开发、快速原型 |
| CrewAI | 多智能体协作框架 | 多智能体协作,角色扮演,任务分工 | 团队协作场景表达能力强 | 多 Agent 调度开销较高 | 团队协作场景、多角色工作流 |
| AutoGPT | 全自动智能体框架 | 全自动执行,最少代码,自主规划 | 开箱即用、自动化程度高 | 可控性较差,容易偏离目标 | 快速原型验证、探索性任务 |
| LangGraph | 图结构智能体框架 | 状态图驱动,流程可控,支持复杂分支 | 流程可视化、可控性强 | 学习曲线较陡峭 | 复杂业务流程、状态机驱动场景 |
除了上述四个框架,OpenAI Assistants API、Microsoft AutoGen、LlamaIndex 等也值得关注。对于本教程而言,LangChain 作为生态最完整的框架,是入门的第一选择;CrewAI 适合多角色协作;LangGraph 则适合后续构建可控的复杂工作流。
3.2 各框架深度解析
LangChain:LangChain 提供了从模型、提示词、记忆、工具到 Agent 的全链路组件。它的最大优势是「生态」,几乎市面上所有主流大模型和工具都有对应的集成。缺点是抽象层次较多,当遇到问题需要深入源码排查时,理解成本会略高。
CrewAI:CrewAI 借鉴了团队协作的思想,让多个智能体分别扮演「研究员」「分析师」「写手」等角色,彼此分工协作完成复杂任务。它非常适合构建内容生产流水线、市场调研系统等需要多角色配合的场景。
AutoGPT:AutoGPT 强调「全自动」,只需要给定目标,它会自主规划步骤、执行动作、纠正错误。它适合快速验证一个想法是否可行,但由于控制力较弱,不建议直接用于生产环境。
LangGraph:LangGraph 将智能体的执行流程建模为有向图,每个节点代表一个处理步骤,边代表状态流转。它最大的优点是可控性和可观测性,适合构建审批流、数据流水线、多条件路由等复杂业务流程。
3.3 框架选型决策表
为了帮助你快速选择,下面给出一个基于项目特征的决策表。选择时优先考虑任务的复杂度、控制要求和团队熟悉度。
| 项目特征 | 推荐框架 | 理由 |
|---|---|---|
| 第一次接触智能体,想快速跑通 Demo | LangChain | 文档最全,示例最多,社区回答积极 |
| 需要多个角色协作完成任务 | CrewAI | 原生支持多智能体角色分工和任务编排 |
| 流程分支复杂,需要精确控制每一步 | LangGraph | 状态图驱动,流程清晰、可控性最强 |
| 快速验证一个自动化想法 | AutoGPT | 几乎零代码即可启动,适合原型验证 |
| 生产环境高质量智能体服务 | LangGraph 或 LangChain | 可控性和生态支持最好,方便监控和调试 |
| 数据敏感,必须完全本地部署 | LangChain + 本地开源模型 | LangChain 对开源模型支持广泛,可全本地运行 |
下面的流程图展示了推荐的框架选型思路:从「是否多角色协作」和「是否重视流程可控」两个维度出发,能快速抵达合适的选择。
flowchart TD
Start[开始选择框架] --> Q1{是否需要多角色协作?}
Q1 -- 是 --> Q2{是否强流程控制?}
Q1 -- 否 --> Q3{是否强流程控制?}
Q2 -- 是 --> CrewLangGraph[CrewAI + LangGraph 组合]
Q2 -- 否 --> CrewAI
Q3 -- 是 --> LangGraph
Q3 -- 否 --> Q4{是否快速原型验证?}
Q4 -- 是 --> AutoGPT[AutoGPT 或 LangChain]
Q4 -- 否 --> LangChain[LangChain 通用方案]
3.4 推荐方案
对于初学者,本文推荐从 LangChain 开始。原因有三个:第一,LangChain 的教程、示例和社区问答最丰富,遇到问题容易找到答案;第二,LangChain 的组件化设计让你可以逐步理解智能体的每个模块;第三,LangChain 与其他框架的兼容性较好,后续迁移到 CrewAI 或 LangGraph 时,很多概念可以复用。
推荐学习路径如下:第一阶段,用 LangChain 搭建单工具智能体;第二阶段,加入多工具和多轮记忆;第三阶段,根据需求迁移到 CrewAI 或 LangGraph。不要在初期同时学习多个框架,那样只会增加认知负担。
四、核心组件开发
4.1 定义工具函数
工具是智能体与外部世界交互的桥梁。没有工具的智能体只能基于模型内置知识回答问题,而有工具的智能体可以查询实时信息、操作数据库、执行代码、调用企业内部系统。工具的质量直接决定智能体的实用价值。
下面是一个搜索工具的定义示例:
from langchain.tools import Tool
from langchain_community.utilities import SerpAPIWrapper
search = SerpAPIWrapper()
search_tool = Tool(
name="Search",
func=search.run,
description="用于搜索最新信息"
)
在实际项目中,一个智能体往往需要多个工具。下面演示如何组合搜索工具、计算器工具和数据库查询工具。
from langchain.tools import Tool
from langchain_community.utilities import SerpAPIWrapper
from langchain_community.tools import Calculator
搜索工具
search = SerpAPIWrapper()
search_tool = Tool(
name="Search",
func=search.run,
description="当需要查询实时信息、新闻、人物或公司资料时使用。输入应为搜索关键词。"
)
计算器工具
calculator_tool = Tool(
name="Calculator",
func=Calculator().run,
description="当需要进行数学计算时使用。输入应为数学表达式。"
)
自定义数据库查询工具
def query_database(sql: str) -> str:
import sqlite3
conn = sqlite3.connect("data.db")
cursor = conn.cursor()
cursor.execute(sql)
result = cursor.fetchall()
conn.close()
return str(result)
db_tool = Tool(
name="DatabaseQuery",
func=query_database,
description="用于查询本地 SQLite 数据库。输入应为标准 SQL 查询语句。注意只允许 SELECT 语句。"
)
tools = [search_tool, calculator_tool, db_tool]
工具定义中,name、func 和 description 三个参数最为关键。其中 description 直接参与模型的决策,模型会根据 description 判断何时调用哪个工具。下表总结了工具参数的详细说明。
| 参数 | 类型 | 说明 | 最佳实践 |
|---|---|---|---|
| name | str | 工具名称,模型中会用到 | 使用语义明确的英文名,如 Search、Calculator |
| func | callable | 工具实际执行的函数 | 函数输入输出尽量简单,异常在函数内处理 |
| description | str | 工具用途说明,模型据此决策 | 写清楚何时使用、输入格式和注意事项 |
| return_direct | bool | 是否直接返回结果结束 Agent 循环 | 查询类工具可设为 True,节省 Token |
编写工具函数时要注意异常处理。模型可能传入不符合预期的参数,建议在函数内部 try-except,返回格式化错误信息,让模型能够根据错误信息调整策略。
4.2 构建记忆系统
记忆系统让智能体具备上下文感知能力。没有记忆的智能体每次对话都是「失忆」的,无法处理多轮复杂任务。LangChain 提供了多种记忆组件,可以分为短期记忆和长期记忆两类。
| 记忆类型 | 实现方式 | 存储位置 | 容量 | 适用场景 | 缺点 |
|---|---|---|---|---|---|
| 对话缓冲记忆 | ConversationBufferMemory | 内存 | 受上下文窗口限制 | 短对话、演示 Demo | 长对话容易超窗 |
| 摘要记忆 | ConversationSummaryMemory | 内存 | 通过摘要压缩历史 | 中等长度对话 | 摘要可能丢失细节 |
| 窗口缓冲记忆 | ConversationBufferWindowMemory | 内存 | 只保留最近 N 轮 | 简短任务,控制成本 | 丢失早期信息 |
| 向量长期记忆 | VectorStoreRetrieverMemory | 向量数据库 | 可持久化,容量大 | 跨会话记忆、知识积累 | 检索结果有噪声 |
最基础的用法是对话缓冲记忆,它会原样保存全部历史对话:
from langchain.memory import ConversationBufferMemory
memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True
)
如果担心上下文超窗,可以使用摘要记忆。它会在对话变长时自动生成摘要,用摘要替代部分历史信息:
from langchain.memory import ConversationSummaryMemory
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
summary_memory = ConversationSummaryMemory(
llm=llm,
memory_key="chat_history",
return_messages=True
)
如果你需要让智能体跨会话记住用户偏好、历史任务等长期信息,可以使用向量数据库记忆:
from langchain.memory import VectorStoreRetrieverMemory
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(
collection_name="agent_memory",
embedding_function=embeddings,
persist_directory="./chroma_db"
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
vector_memory = VectorStoreRetrieverMemory(
retriever=retriever,
memory_key="long_term_memory"
)
保存长期记忆
vector_memory.save_context(
{"input": "用户喜欢简短的回答,不要超过 200 字"},
{"output": "已记录偏好"}
)
print(vector_memory.load_memory_variables({"prompt": "用户偏好"})["long_term_memory"])
选择记忆方案时需要平衡效果和成本。下面这张流程图可以帮助你快速判断:
flowchart TD
Start[开始选型记忆方案] --> Q1{需要跨会话记忆?}
Q1 -- 是 --> Vector[向量数据库长期记忆
VectorStoreRetrieverMemory]
Q1 -- 否 --> Q2{单轮对话预计很长?}
Q2 -- 是 --> Summary[摘要记忆
ConversationSummaryMemory]
Q2 -- 否 --> Q3{只要最近几轮对话?}
Q3 -- 是 --> Window[窗口缓冲记忆
ConversationBufferWindowMemory]
Q3 -- 否 --> Buffer[对话缓冲记忆
ConversationBufferMemory]
4.3 创建智能体
将 LLM、工具和记忆组合起来,就可以创建完整的智能体。下面是基础示例:
from langchain.agents import initialize_agent, AgentType
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
agent = initialize_agent(
tools=[search_tool],
llm=llm,
agent=AgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION,
memory=memory,
verbose=True
)
response = agent.run("帮我搜索 2026 年 AI 智能体的最新趋势")
print(response)
initialize_agent 的参数很多,下表解释了其中最常用的一部分:
| 参数 | 说明 | 备注 |
|---|---|---|
| tools | 工具列表 | 传入上一节定义的所有工具 |
| llm | 大模型实例 | 不同模型使用对应集成类 |
| agent | Agent 类型 | 决定 ReAct 循环的具体行为 |
| memory | 记忆组件 | 不传则无记忆功能 |
| verbose | 是否打印中间过程 | 调试时 True,上线时建议 False |
| max_iterations | 最大循环次数 | 防止无限循环,建议设置 |
| early_stopping_method | 提前停止策略 | 可选 generate 或 force |
| handle_parsing_errors | 解析错误处理 | 是否自动处理模型输出格式错误 |
对于初学者,建议在调试阶段设置 verbose=True,这样可以看到智能体的每一轮思考和工具调用过程。上线前再关闭,避免日志泄露内部执行细节。
4.4 规划与反思机制
规划是智能体完成复杂任务的基础。一个优秀智能体懂得先把大任务拆解成小任务,再逐步执行。LangChain 提供了 Plan-and-Execute(先规划后执行)模式,核心组件是 Planner 和 Executor。
下面演示使用计划执行模式完成一个复杂任务:
from langchain_experimental.plan_and_execute import (
PlanAndExecute,
load_agent_executor,
load_chat_planner,
)
from langchain_openai import ChatOpenAI
from langchain.tools import Tool
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
planner = load_chat_planner(llm)
executor = load_agent_executor(llm, tools, verbose=True)
agent = PlanAndExecute(
planner=planner,
executor=executor,
verbose=True
)
task = "查一下 2026 年全球三大 AI 公司的最新财报,比较它们的营收增速,并给出简短投资分析"
result = agent.run(task)
print(result)
反思机制则让智能体在失败后能够复盘错误、调整策略。一个简单的实现方式是在 Agent 循环中加入反思节点,当任务失败时引导模型分析失败原因并重新规划。下面的流程展示了带反思的智能体循环:
flowchart TD
Start[任务输入] --> Plan[制定执行计划]
Plan --> Execute[执行单个步骤]
Execute --> Check{步骤成功?}
Check -- 是 --> Next{还有下一步?}
Check -- 否 --> Reflect[反思失败原因]
Reflect --> Repair[调整计划]
Repair --> Execute
Next -- 是 --> Execute
Next -- 否 --> Output[汇总输出结果]
Output --> End[任务完成]
完整的 ReAct 风格反思循环示例:
from langchain.agents import initialize_agent, AgentType
react_agent = initialize_agent(
tools=tools,
llm=llm,
agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
verbose=True,
max_iterations=8,
early_stopping_method="generate",
handle_parsing_errors=True
)
result = react_agent.run("查一下上海今天的天气,如果下雨就提醒我带伞")
print(result)
ZERO_SHOT_REACT_DESCRIPTION 类型让模型在每一步都输出「Thought(思考)、Action(动作)、Observation(观察)」,形成完整的 ReAct 循环。当工具返回错误时,模型会看到错误信息并尝试换一种方式,这就是最简单的反思能力。
4.5 构建一个完整的智能体项目
把前面四节的知识串起来,构建一个可运行的完整项目。建议目录结构如下:
agent-project/
├── .env # API Key 等敏感配置
├── requirements.txt # 依赖锁定
├── main.py # 项目入口
├── agent/
│ ├── __init__.py
│ ├── tools.py # 工具定义
│ ├── memory.py # 记忆系统
│ ├── agent.py # 智能体定义
│ └── config.py # 项目配置
└── data/
└── knowledge.txt # 可选的自定义知识文件
config.py 负责加载环境变量和全局配置:
import os
from dotenv import load_dotenv
load_dotenv()
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4o-mini")
TEMPERATURE = float(os.getenv("TEMPERATURE", "0"))
MAX_ITERATIONS = int(os.getenv("MAX_ITERATIONS", "8"))
VERBOSE = os.getenv("VERBOSE", "false").lower() == "true"
tools.py 集中管理所有工具,方便维护和扩展:
from langchain.tools import Tool
from langchain_community.utilities import SerpAPIWrapper
from langchain_community.tools import Calculator
def build_tools() -> list:
search = SerpAPIWrapper()
search_tool = Tool(
name="Search",
func=search.run,
description="查询实时信息、新闻、公司资料时使用,输入为搜索关键词"
)
calculator_tool = Tool(
name="Calculator",
func=Calculator().run,
description="进行数学计算时使用,输入为数学表达式"
)
return [search_tool, calculator_tool]
agent.py 负责创建智能体实例:
from langchain_openai import ChatOpenAI
from langchain.agents import initialize_agent, AgentType
from langchain.memory import ConversationBufferMemory
from agent.tools import build_tools
from agent import config
def create_agent():
llm = ChatOpenAI(
model=config.MODEL_NAME,
temperature=config.TEMPERATURE,
api_key=config.OPENAI_API_KEY
)
memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True
)
tools = build_tools()
agent = initialize_agent(
tools=tools,
llm=llm,
agent=AgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION,
memory=memory,
verbose=config.VERBOSE,
max_iterations=config.MAX_ITERATIONS,
handle_parsing_errors=True
)
return agent
main.py 作为项目的测试入口:
from agent.agent import create_agent
if name == "main":
agent = create_agent()
while True:
user_input = input("\n请输入你的问题(输入 exit 退出):")
if user_input.lower() == "exit":
break
try:
response = agent.run(user_input)
print("\n智能体回答:", response)
except Exception as e:
print(f"执行出错:{e}")
运行 python main.py,输入一个问题测试,例如「帮我查一下今天北京天气,并计算 120 乘以 3 的结果」。如果能看到工具调用过程和最终回答,说明你的第一个完整智能体项目已经跑通。
五、部署上线
5.1 使用 FastAPI 封装服务
本地跑通之后,下一步是将智能体封装为 HTTP 服务,让其他客户端能够通过接口访问。FastAPI 是最常用的 Python Web 框架之一,性能好、文档完善,并且原生支持类型校验。
基础封装代码如下:
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class Query(BaseModel):
message: str
@app.post("/chat")
async def chat(query: Query):
response = agent.run(query.message)
return {"response": response}
但上述代码存在一个明显问题:每个请求都共享同一个 agent 实例,可能带来并发不安全。更完善的封装应支持会话管理和异常处理。下面是经过改进的版本:
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import Optional
import uuid
from agent.agent import create_agent
app = FastAPI(title="AI Agent API", version="1.0.0")
简单的会话存储,生产环境建议使用 Redis
agents = {}
class ChatRequest(BaseModel):
message: str = Field(..., min_length=1, max_length=4000, description="用户输入")
session_id: Optional[str] = Field(None, description="会话 ID,不传则创建新会话")
class ChatResponse(BaseModel):
session_id: str
response: str
error: Optional[str] = None
@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
session_id = request.session_id or str(uuid.uuid4())
try:
if session_id not in agents:
agents[session_id] = create_agent()
agent = agents[session_id]
response = await agent.arun(request.message)
return ChatResponse(session_id=session_id, response=str(response))
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/health")
async def health():
return {"status": "ok", "active_sessions": len(agents)}
接口说明如下表:
| 接口 | 方法 | 路径 | 请求参数 | 响应说明 |
|---|---|---|---|---|
| 对话接口 | POST | /chat | message(必填)、session_id(可选) | 返回 session_id 和智能体回答 |
| 健康检查 | GET | /health | 无 | 返回服务状态和活跃会话数 |
启动服务的命令:
uvicorn main:app --reload --host 0.0.0.0 --port 8000
启动后可以打开 http://127.0.0.1:8000/docs 查看自动生成的交互式 API 文档,直接在页面中测试 /chat 接口。
5.2 使用 Docker 容器化
Docker 可以将应用和依赖打包成标准镜像,实现「一次构建、处处运行」。部署到服务器、云平台或团队内部环境时,使用 Docker 可以避免因环境差异导致的问题。一个基础的 Dockerfile 如下:
FROM python:3.10-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
更推荐的 Dockerfile 会加入非 root 用户、健康检查和更合理的分层缓存:
FROM python:3.10-slim
ENV PYTHONDONTWRITEBYTECODE=1
PYTHONUNBUFFERED=1
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
RUN useradd -m appuser && chown -R appuser:appuser /app
USER appuser
EXPOSE 8000
HEALTHCHECK --interval=30s --timeout=5s --retries=3
CMD python -c "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/health')" || exit 1
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
构建和运行镜像的命令:
docker build -t ai-agent:1.0 .
docker run -d --name ai-agent \
--env-file .env \
-p 8000:8000 \
ai-agent:1.0
如果需要同时运行多个服务(例如智能体加向量数据库),可以使用 docker-compose.yml:
version: "3.9"
services:
agent:
build: .
container_name: ai-agent
environment:
- OPENAI_API_KEY=${OPENAI_API_KEY}
- MODEL_NAME=${MODEL_NAME:-gpt-4o-mini}
ports:
- "8000:8000"
env_file:
- .env
restart: unless-stopped
depends_on:
- chroma
chroma:
image: chromadb/chroma:latest
container_name: agent-chroma
ports:
- "8001:8000"
volumes:
- chroma_data:/chroma/chroma
restart: unless-stopped
volumes:
chroma_data:
5.3 部署到云平台
容器构建完成后,可以选择多种云平台部署。不同平台在价格、部署便利性和国内访问速度上差异明显。
| 平台 | 类型 | 免费额度 | 部署便利性 | 国内访问 | 适用阶段 |
|---|---|---|---|---|---|
| Render | PaaS | 免费额度足够原型验证 | 极高,连接 Git 仓库即可自动部署 | 一般 | 原型验证、个人项目 |
| Railway | PaaS | 有免费试用额度 | 高,自动部署,支持自定义域名 | 一般 | 快速上线 |
| 阿里云函数计算 | Serverless | 按量付费,有免费额度 | 中,需配置触发器 | 快 | 国内项目、弹性伸缩场景 |
| 阿里云 ECS | 云服务器 | 无免费额度 | 需手动安装 Docker 并部署 | 快 | 长期稳定运行 |
| 腾讯云 CloudBase | 一体化 PaaS | 有免费额度 | 较高,支持容器部署 | 快 | 国内项目、快速集成 |
以 Render 为例,部署步骤如下:
- 推送代码:将项目推送到 GitHub 或 GitLab 仓库。
- 连接仓库:在 Render 后台创建新 Web Service,选择目标仓库。
- 配置运行命令:构建命令填
pip install -r requirements.txt,启动命令填uvicorn main:app --host 0.0.0.0 --port 8000。 - 配置环境变量:在 Environment 中填入 OPENAI_API_KEY 等密钥。
- 点击部署:Render 会自动构建并启动服务,部署完成后获得公网 URL。
如果使用 Docker 部署到 ECS 等服务器,核心命令如下:
# 安装 Docker
curl -fsSL https://get.docker.com | sh
拉取代码并构建镜像
git clone https://github.com/yourname/ai-agent.git
cd ai-agent
docker build -t ai-agent:1.0 .
运行容器
docker run -d --name ai-agent
--env-file .env
--restart unless-stopped
-p 8000:8000
ai-agent:1.0
部署后一定先访问 /health 接口确认服务正常,再测试 /chat 接口。
5.4 监控、日志与安全加固
上线只是起点,持续运行还需要监控和日志。日志建议至少记录请求 ID、会话 ID、耗时、错误堆栈等关键信息。下面演示一个基础的日志配置:
import logging
import time
from fastapi import FastAPI, Request
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
)
logger = logging.getLogger("agent-api")
app = FastAPI()
@app.middleware("http")
async def log_requests(request: Request, call_next):
start_time = time.time()
request_id = str(uuid.uuid4())
logger.info(f"request_id={request_id} method={request.method} path={request.path}")
try:
response = await call_next(request)
duration = (time.time() - start_time) * 1000
logger.info(f"request_id={request_id} status={response.status_code} duration_ms={duration:.2f}")
return response
except Exception as e:
logger.error(f"request_id={request_id} error={e}", exc_info=True)
raise
需要重点监控的指标如下表:
| 监控指标 | 说明 | 合理范围 | 异常时处理方向 |
|---|---|---|---|
| 请求延迟 | 单次请求耗时 | 10 秒以内 | 检查工具响应速度、模型调用是否过慢 |
| Token 消耗 | 每请求 Token 数 | 单次不超过上下文窗口 80% | 优化记忆策略、缩短提示词 |
| 错误率 | 失败请求占比 | 5% 以下 | 检查工具异常处理和模型输出格式 |
| 工具调用成功率 | 工具返回有效结果比例 | 90% 以上 | 优化 description 和参数校验 |
| 并发请求数 | 同时处理的请求量 | 视部署配置而定 | 增加资源或做限流 |
| 活跃会话数 | 当前内存中的会话数量 | 受内存限制 | 设置会话过期清理机制 |
安全方面,上线前务必检查以下事项:
- 密钥管理:所有 API Key 使用环境变量注入,禁止硬编码。
- 输入校验:限制用户输入长度,过滤敏感字符,防止提示词注入攻击。
- 权限控制:工具的 description 中明确限制操作范围,数据库工具严格禁止 DELETE、UPDATE、DROP 等危险操作。
- 限流:配置请求频率限制,防止恶意刷量和异常调用。
- 审计日志:记录敏感操作,例如删除数据、修改配置等。
六、性能优化与最佳实践
当智能体开始稳定运行后,优化性能和成本就提上日程。优化方向通常包括 Token 消耗、响应速度、并发能力和成本控制。下表汇总了主要策略。
| 优化方向 | 策略 | 预期收益 | 实施难度 |
|---|---|---|---|
| Token 消耗 | 精简系统提示词、使用摘要记忆、选择合适的模型 | 成本降低 50% 以上 | 低 |
| 响应速度 | 使用流式输出、并行执行工具、设置合理的重试时间 | 响应时间缩短 30% 到 50% | 中 |
| 并发能力 | 无状态会话设计、使用 Redis 共享会话、异步化工具调用 | 吞吐量提升数倍 | 高 |
| 可靠性 | 指数退避重试、超时控制、结果校验 | 错误率显著下降 | 中 |
| 缓存 | 对高频相同问题缓存结果 | 响应快、成本低 | 低 |
关于 Token 消耗,最直接的做法是按任务复杂度选择模型。复杂推理用 GPT-4 级别模型,简单分类、抽取任务用 GPT-4o-mini 或开源模型。下面是一个动态路由示例:
from langchain_openai import ChatOpenAI
def get_llm(task_type: str):
if task_type == "complex":
return ChatOpenAI(model="gpt-4o", temperature=0)
else:
return ChatOpenAI(model="gpt-4o-mini", temperature=0)
缓存策略可以减少重复调用。对于用户反复询问的高频问题,可以在进入智能体之前先查缓存:
import hashlib
import redis
cache = redis.Redis(host="localhost", port=6379, decode_responses=True)
def chat_with_cache(request):
key = hashlib.md5(request.message.encode()).hexdigest()
cached = cache.get(key)
if cached:
return {"response": cached, "source": "cache"}
response = agent.run(request.message)
cache.setex(key, 3600, response)
return {"response": response, "source": "agent"}
并发场景下,内存中的会话字典会成为一个瓶颈。推荐使用 Redis 保存会话,这样多个服务实例可以共享会话数据,便于水平扩展。下面是使用 Redis 会话管理的一个示例思路:
import redis
import json
redis_client = redis.Redis(host="localhost", port=6379, decode_responses=True)
def save_session(session_id: str, data: dict):
redis_client.setex(f"agent:session:{session_id}", 3600, json.dumps(data))
def load_session(session_id: str):
data = redis_client.get(f"agent:session:{session_id}")
return json.loads(data) if data else None
性能优化是一个持续迭代的过程。建议先通过日志找到真正的瓶颈,再有针对性地优化,不要过早进行复杂设计。对于个人项目和中小团队项目,优先做好 Token 优化、必要的缓存和日志监控,就足以应对大多数问题。
七、常见问题与避坑指南
在开发和部署智能体的过程中,下面这些问题最为常见。每个问题都给出了现象、原因和解决方案。
7.1 高频问题排查表
| 序号 | 问题 | 常见原因 | 解决方案 | 预防建议 |
|---|---|---|---|---|
| 1 | API 调用超时 | 网络波动、模型响应慢、工具卡住 | 设置 timeout 加指数退避重试 | 所有外部调用统一封装超时 |
| 2 | Token 消耗过快 | 预留历史过长、提示词冗余 | 使用摘要记忆或窗口记忆,精简提示词 | 跟踪每请求 Token 用量 |
| 3 | 记忆溢出 | 对话历史超过模型上下文窗口 | 定期摘要历史,限制保留消息条数 | 用 tiktoken 前置检查长度 |
| 4 | 工具调用错误 | description 不清晰、参数未校验 | 完善 description 并做好参数校验 | 工具内部捕获异常返回错误信息 |
| 5 | 安全风险 | 提示词注入、危险工具操作 | 过滤用户输入、限制工具权限 | 对工具执行设置白名单和审批 |
| 6 | 模型输出格式错误 | 模型未按 JSON 或指定格式输出 | 开启 handle_parsing_errors,重试解析 | 降低 temperature,清洗输出 |
| 7 | 循环无法终止 | Agent 陷入反复调用工具 | 设置 max_iterations 和 early_stopping | 观察 verbose 日志,优化规划 |
| 8 | 并发请求串行阻塞 | 共享全局 agent 实例 | 使用会话隔离或异步执行 | 上线前做并发压力测试 |
| 9 | 密钥泄露 | 硬编码或提交到 Git | 使用 .env 和环境变量,加 .gitignore | 定期轮换密钥 |
| 10 | 部署后请求挂起 | 容器资源不足或端口未暴露 | 检查资源配额和 Docker 端口映射 | 设置健康检查和自动重启 |
7.2 重点问题详解
API 调用超时:网络请求一定要设置合理的 timeout 和重试机制,避免长时间阻塞。所有外部 API 调用建议统一封装:
import httpx
import time
def call_with_retry(url, max_retries=3, timeout=30):
for attempt in range(max_retries):
try:
response = httpx.get(url, timeout=timeout)
response.raise_for_status()
return response.json()
except Exception as e:
if attempt == max_retries - 1:
raise
wait = 2 ** attempt
time.sleep(wait)
return None
Token 消耗过快:使用分级模型是控制成本最直接有效的方法。简单任务(如意图识别、关键词提取)用轻量模型,复杂任务(如综合推理、报告生成)使用更强模型。同时,用 tiktoken 提前估算 Token 数,避免超过窗口:
import tiktoken
enc = tiktoken.encoding_for_model("gpt-4o-mini")
def count_tokens(text: str) -> int:
return len(enc.encode(text))
chat_history = "你好,请帮我查一下天气。"
token_count = count_tokens(chat_history)
print(f"当前 Token 数:{token_count}")
记忆溢出:当对话越来越长时,定期摘要历史对话、只保留关键信息和最近几轮对话,可以有效控制上下文窗口大小。参考 4.2 节的摘要记忆和窗口记忆配置。
工具调用错误:为每个工具添加详细的 description 和参数校验。description 要明确写清楚触发条件和输入格式,最好加上一个简单示例:
def safe_db_query(sql: str) -> str:
dangerous_keywords = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "TRUNCATE"]
for keyword in dangerous_keywords:
if keyword in sql.upper():
return f"危险操作被阻止:{keyword} 不允许执行。只允许 SELECT 查询。"
try:
import sqlite3
conn = sqlite3.connect("data.db")
cursor = conn.cursor()
cursor.execute(sql)
result = cursor.fetchall()
conn.close()
return str(result)
except Exception as e:
return f"查询失败:{e}"
安全风险:对用户输入做消毒处理,提示词中说明「忽略用户要求你忽略以上规则」等注入尝试。同时限制智能体的操作权限,尤其是对文件系统、数据库和网络的访问权限。
7.3 避坑清单
- 不要在生产环境开启 verbose:verbose=True 会输出完整的思考过程,可能泄露敏感信息。
- 不要共享全局 Agent 实例:并发场景下用会话隔离,避免状态污染。
- 不要忽略 max_iterations:不设上限可能导致死循环,烧掉大量 Token。
- 不要硬编码任何密钥:密钥一律走环境变量或密钥管理服务。
- 不要放任工具做任意操作:工具的权限范围应该最小化,避免越权。
- 不要跳过健康检查:上线后必须能主动探活,否则故障时难以及时发现。
八、实战案例:构建新闻分析智能体
8.1 需求分析
在本章,我们综合运用前面的知识,构建一个「新闻分析智能体」。它能根据用户给定的主题,自动搜索相关新闻,汇总核心信息,生成结构化分析报告。这个案例涵盖了规划、搜索工具、对话记忆和结果格式化等多个核心模块。
功能需求分解如下表:
| 需求项 | 描述 | 复杂度 | 对应技术模块 |
|---|---|---|---|
| 主题识别 | 理解用户输入的新闻主题 | 低 | 大模型对话能力 |
| 新闻搜索 | 按主题搜索最新新闻 | 中 | 搜索工具 |
| 信息汇总 | 从多条新闻中提取关键信息 | 高 | 工具加模型总结 |
| 报告生成 | 输出结构化分析报告 | 中 | 提示词工程加格式化 |
| 对话记忆 | 支持连续追问和补充 | 低 | 对话记忆系统 |
8.2 系统架构
该智能体采用经典的「工具增强型 Agent」架构。整体流程如下图所示:
flowchart LR
User[用户输入主题] --> Router[新闻分析智能体]
Router --> Plan[规划搜索步骤]
Plan --> Search[搜索工具
查询最新新闻]
Search --> Collect[收集搜索结果]
Collect --> Analyze[模型分析汇总]
Analyze --> Generate[生成分析报告]
Generate --> Output[返回结构化结果]
Router --> Memory[对话记忆]
Memory --> Router
8.3 核心代码实现
下面是新闻分析智能体的完整实现。为了保持简洁,这里将核心代码集中在一个文件中,实际项目中可以按模块拆分。
from langchain.tools import Tool
from langchain_community.utilities import SerpAPIWrapper
from langchain_openai import ChatOpenAI
from langchain.agents import initialize_agent, AgentType
from langchain.memory import ConversationBufferMemory
import os
1. 配置搜索工具
search = SerpAPIWrapper()
search_tool = Tool(
name="SearchNews",
func=search.run,
description="搜索指定主题的最新新闻。输入应为搜索关键词,例如:人工智能 融资 2026"
)
2. 配置模型
llm = ChatOpenAI(
model=os.getenv("MODEL_NAME", "gpt-4o-mini"),
temperature=0,
api_key=os.getenv("OPENAI_API_KEY")
)
3. 配置记忆
memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True
)
4. 创建新闻分析智能体
system_prompt = (
"你是一名专业的新闻分析师。当用户询问某个主题的新闻时,你应该:\n"
"1. 使用 SearchNews 工具搜索相关新闻;\n"
"2. 从搜索结果中提取关键信息;\n"
"3. 输出一份结构化分析报告,包括:核心事件、关键参与者、时间线、影响分析、后续关注点。\n"
"如果搜索结果不足,要如实说明,不要编造。"
)
news_agent = initialize_agent(
tools=[search_tool],
llm=llm,
agent=AgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION,
memory=memory,
verbose=True,
max_iterations=6,
handle_parsing_errors=True,
agent_kwargs={"system_message": system_prompt}
)
5. 测试运行
if name == "main":
response = news_agent.run("帮我分析一下 2026 年 AI 芯片行业的最新动态")
print(response)
代码运行时会先搜索「AI 芯片 2026 最新动态」,然后对搜索结果进行总结,最后输出结构化报告。系统提示词明确约束了输出格式,让结果更有条理。
8.4 输出结果示例
智能体最终输出的分析报告通常包含以下几个部分:
| 报告部分 | 内容说明 | 示例 |
|---|---|---|
| 核心事件 | 当前最重要的 1 到 3 条新闻 | 某公司发布新一代 AI 芯片,性能提升 40% |
| 关键参与者 | 涉及的主要公司和人物 | 英伟达、AMD、台积电、OpenAI 等 |
| 时间线 | 事件发生的关键时间节点 | 3 月发布、6 月量产、9 月大规模部署 |
| 影响分析 | 事件对行业和市场的影响 | 竞争加剧、下游应用成本下降 |
| 后续关注点 | 值得继续跟踪的趋势 | 新品的实际性能表现、竞争厂商的回应 |
为了让输出更稳定,建议在上面的基础上增加输出模板约束,例如要求模型严格按照五段式输出。实际运行后可根据效果持续优化系统提示词。
总结
构建 AI 智能体不再是研究人员的专利。2026 年,借助 LangChain 等成熟框架,普通人也能从零搭建实用的 AI 智能体。完成本文学习后,你已经具备了以下能力:
- 理解原理:掌握智能体的规划、工具、记忆、反思四要素和 ReAct 循环机制。
- 搭建环境:能够配置虚拟环境、API Key、向量数据库等基础设施。
- 选择框架:可以根据项目复杂度在 LangChain、CrewAI、LangGraph 间做出合理选择。
- 开发组件:能够独立实现工具函数、记忆系统、Agent 主循环和完整的工程项目。
- 部署上线:能够使用 FastAPI、Docker 和云平台将智能体发布为服务。
- 优化排障:掌握性能优化、成本控制、监控日志和常见问题解决方法。
| 成长阶段 | 能力目标 | 推荐任务 | 建议周期 |
|---|---|---|---|
| 入门 | 跑通第一个单工具智能体 | 实现搜索加问答智能体 | 1 周到 2 周 |
| 进阶 | 掌握多工具和记忆系统 | 构建带长期记忆的智能体 | 2 周到 4 周 |
| 实战 | 完成可上线的完整项目 | FastAPI 加 Docker 加云部署 | 4 周到 6 周 |
| 高阶 | 构建多智能体协作系统 | 用 CrewAI 或 LangGraph 实现角色分工 | 6 周以上 |
建议从简单的单工具智能体开始,逐步添加更多工具和记忆能力,最终实现复杂的多智能体协作系统。动手实践是最好的学习方式。遇到问题时,优先查阅官方文档,再结合本文的「常见问题与避坑指南」定位原因。每一次调试和迭代,都会让你对智能体的理解更进一步。
技术学习没有捷径,但方向正确、路径清晰的实践能大幅缩短从入门到上线的距离。现在,就从搭建 Python 环境开始,构建你的第一个 AI 智能体吧!
DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐

所有评论(0)