The AI Agent Tech Stack Explained

解释 AI Agent 技术栈 - MachineLearningMastery.com

解释 AI Agent 技术栈

By

Shittu Olumide

on

June 27, 2026

in

Artificial Intelligence

0

Share

Post

在本文中，你将学习到生产级 AI Agent 技术栈的七层是如何协同工作的，从基础模型到部署基础设施。

我们将涵盖的主题包括：

• 从基础模型、编排框架，到记忆、检索、工具、可观测性和部署，每一层的功能。
• 如何通过实际代码实现每一层，包括一个有状态的代理、记忆系统、RAG 管道、自定义工具和追踪。
• 根据你是原型设计、扩展初创公司，还是在企业环境中运行，每一层应使用哪些技术组合。

介绍

想象一下：你让一个 AI Agent 去研究三个竞争对手，从每个竞争对手的网站上提取价格数据，将研究结果总结成结构化报告，并在上午9点前发送到 Slack 频道。你按下回车键。30秒后，报告就出现了。

刚刚在幕后发生的事情并不是魔法，也不是单一因素。它是七层不同的技术依次协作，每一层负责特定的任务，每一层都有其特定的故障方式。最顶层的模型获得了所有关注。它下面的六层决定了代理是否真的能正常工作。

根据 Gartner 的数据，到 2026 年底，40% 的企业应用程序将集成特定任务的 AI Agent，而 2025 年这一比例不到 5%。这不是一个渐进的曲线。这是一条几乎垂直的采用曲线，负责这些部署的工程师和技术负责人需要理解整个技术栈，而不仅仅是他们恰好拥有的那一层。

本文将依次介绍每一层，从基础模型到部署基础设施。到文章结束时，你将了解每一部分的功能、存在的原因、各层之间的连接方式，以及在每一层应该使用什么。

“标准”模型和“推理”模型家族之间不再存在明确的硬性划分，这与2025年的情况不同；OpenAI、Anthropic和Google都已经将推理能力整合到单一模型中，该模型决定需要思考多长时间。GPT-5.5提供了可调节的推理努力级别（从无到xhigh），Claude的effort参数和Gemini的思考级别也是如此。对于大多数代理工作流程，默认或低努力设置是最佳选择：快速且成本低。对于需要仔细规划或数学推理的任务，提高努力级别可以以正确性换取成本。

第二层：编排框架

如果基础模型是大脑，编排框架就是神经系统。它处理控制流：决定代理下一步应该做什么，何时调用工具，如何处理结果，以及如何在整个推理循环中保持多个步骤之间的一致性。

大多数框架实现的模式称为ReAct（推理和行动）。代理生成一个想法，决定一个行动，通过工具执行该行动，观察结果，然后再次思考。这个循环会一直重复，直到代理生成最终答案。听起来很简单。实际上，这是大多数生产失败发生的地方：代理调用了错误的工具，陷入循环，或者未能识别出何时已拥有足够的信息来停止。

ReAct循环的循环图（点击放大）

• LangChain是最广泛采用的框架。它提供了一个庞大的集成生态系统和良好的文档。在原型阶段，它添加了过多抽象的批评是合理的，但一旦你需要抽象提供的功能，这种批评就变得不那么相关了。由同一团队开发的LangGraph更适合需要对执行图进行细粒度控制的状态多代理工作流程。如果你的代理涉及多个专家协调完成任务，LangGraph是更清晰的选择。

• CrewAI专门设计用于多代理协调。它允许你为代理定义角色，分配任务，并在结构化的工作流程中让他们协作。它比LangGraph更高级，启动速度更快，但对执行细节的控制较少。来自微软的AutoGen对多代理系统采用对话式方法。代理通过消息传递接口进行交互，这使得交互逻辑非常清晰。

• Semantic Kernel是微软面向企业环境的选择，支持C#、Python和Java的生产就绪功能。如果你已经在运行微软堆栈的企业环境中操作，它自然地契合。LlamaIndex最初是一个文档摄入和检索框架，此后已发展为一个完整的代理框架，尤其在以RAG为主的流程中具有强大的支持。

正确的选择取决于你的代理需要做什么。对于单代理任务执行器：LangGraph或LangChain。对于协调的专家代理团队：CrewAI或AutoGen。对于企业环境：Semantic Kernel。对于以文档为主的检索流程：LlamaIndex。

以下是使用LangGraph实现的最小工作代理，它可以处理工具使用并维护状态。

前提条件：

pip install langgraph langchain-openai langchain-community python-dotenv

1

pip

install

langgraph

langchain

-

openai

community

python

dotenv

如何运行：保存为 agent.py，将你的 OPENAI_API_KEY 添加到 .env 文件中，然后运行 python agent.py

agent.py # 基于 LangGraph 构建的最小状态化代理，支持工具使用 # Python 3.10+ | LangGraph 0.2+ | LangChain 0.3+ import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langchain_community.tools import DuckDuckGoSearchRun from langchain_core.messages import HumanMessage from langgraph.prebuilt import create_react_agent # 从 .env 文件加载 API 密钥 load_dotenv() # 初始化语言模型 # temperature=0 用于代理任务中确定性、专注的响应 llm = ChatOpenAI( model="gpt-5.5", temperature=0, api_key=os.getenv("OPENAI_API_KEY") ) # 注册代理可以使用的工具 # DuckDuckGoSearchRun 不需要 API 密钥，适合开发 tools = [DuckDuckGoSearchRun()] # LangGraph 的 create_react_agent 将 LLM、工具和内置的 ReAct 循环连接在一起，无需任何样板代码 agent = create_react_agent(llm, tools) # 使用示例查询运行代理 # 代理将根据问题决定是否使用工具 result = agent.invoke({ "messages": [HumanMessage(content="Nvidia 的当前市值是多少？")] }) # 最终响应是消息列表中的最后一个消息 print(result["messages"][-1].content)

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

agent.py

基于 LangGraph 构建的最小状态化代理，支持工具使用

Python 3.10+ | LangGraph 0.2+ | LangChain 0.3+

import

os

from

load_dotenv

langchain_openai

ChatOpenAI

langchain_community

.

tools

DuckDuckGoSearchRun

langchain_core

messages

HumanMessage

prebuilt

create_react

_

agent

从 .env 文件加载 API 密钥

(

)

初始化语言模型

temperature=0 用于代理任务中确定性、专注的响应

llm

=

model

"gpt-5.5"

,

temperature

api_key

getenv

"OPENAI_API_KEY"

注册代理可以使用的工具

DuckDuckGoSearchRun 不需要 API 密钥，适合开发

[

]

LangGraph 的 create_react_agent 将 LLM、工具和内置的 ReAct 循环连接在一起，无需任何样板代码

create_react_agent

使用示例查询运行代理

代理将根据问题决定是否使用工具

result

invoke

{

"messages"

:

content

"Nvidia 的当前市值是多少？"

}

最终响应是消息列表中的最后一个消息

print

这是它的工作方式：create_react_agent 会自动处理完整的 ReAct 循环。代理接收到问题，决定需要当前数据，调用 DuckDuckGo 搜索工具，读取结果，并综合出最终答案。输出中的消息列表包含该推理过程的完整追踪。

第三层：记忆系统

状态无性是任何 LLM 的默认行为。每次调用都从头开始，除非你显式传递上下文，否则不会知道之前发生了什么。对于一次性问题，这没有问题。但对于需要跟踪对话、记住用户的偏好或基于昨天的工作进行构建的代理，这是一个根本性的问题。

根据 Atlan 对 AI 代理记忆的研究，2025 年 95% 的企业生成式 AI 试点项目没有可衡量的 ROI，失败归因于上下文准备不足，而不是模型质量。代理的失败不是因为模型错误，而是因为记忆层不存在。

生产代理中有四种类型的内存，每种内存负责不同的任务：

• 工作内存（上下文内） 是当前的上下文窗口。它保存当前对话、你传入的任何文档以及最近工具调用的结果。它速度快且不需要基础设施，但它是会话绑定的。当会话结束时，它就会消失。

• 情景内存 是先前交互的日志。如记忆类型的研究中所述，情景内存存储了发生了什么：时间戳、任务、采取的行动、结果。这使得代理能够回答诸如“我们上周二做了什么？”或“用户在三会话前对这个项目说了什么？”之类的问题。

• 语义内存 是存储在外部的事实性知识，包括定义、实体关系以及模型未经过训练的特定领域事实。这是你的 RAG 管道注入的地方（更多内容将在下一层中介绍）。

• 程序性内存 编码了工作流程和工具使用模式，是代理应始终遵循的可重复行为。它存在于系统提示或版本控制的指令文件中，并影响代理生成的每一个响应。

以下是使用 LangChain 0.3+ 推荐模式实现工作内存和情景内存的方法：

前提条件：

pip install langchain langchain-openai python-dotenv

如何运行： 保存为 memory.py，确保你的 .env 文件中有 OPENAI_API_KEY，然后运行 python memory.py。

memory.py

用于持久化代理上下文的工作记忆 + 事件记忆

使用当前 LangChain 0.3+ 模式（旧版 ConversationBufferMemory 已弃用）

import os import json from datetime import datetime from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, trim_messages

load_dotenv()

llm = ChatOpenAI( model="gpt-5.5", temperature=0.2, api_key=os.getenv("OPENAI_API_KEY") )

── 事件记忆存储 ─────────────────────────────────────────────────────

在生产环境中，应将此列表替换为数据库（SQLite、Postgres、Redis）。

此处的结构：每个事件是一个包含时间戳、用户输入和代理响应的字典。

episodic_store: list[dict] = []

def save_episode(user_input: str, agent_response: str) -> None: """将对话轮次保存到事件存储中。""" episodic_store.append({ "timestamp": datetime.now().isoformat(), "user": user_input, "agent": agent_response })

def load_recent_episodes(n: int = 5) -> str: """检索最近的 N 个事件，并以格式化字符串形式注入到上下文中。""" if not episodic_store: return "没有之前的对话历史。" recent = episodic_store[-n:] return "\n".join( f"[{ep['timestamp']}] 用户: {ep['user']} | 代理: {ep['agent']}" for ep in recent )

── 工作记忆（上下文内） ───────────────────────────────────────────────

我们自行管理消息列表，并在每次调用 LLM 之前通过 trim_messages 进行修剪

以保持在模型的上下文限制内。max_tokens=4000 为模型的响应留出空间。

working_memory: list = []

def chat(user_input: str) -> str: """向代理发送消息。事件历史被注入到系统提示中。 在每次调用之前修剪工作记忆以防止上下文溢出。"""

将事件记忆注入到系统提示中，使模型具有长期上下文

system = SystemMessage(content=( "你是一个有帮助且具有上下文感知的助手。\n\n" "最近的对话历史：\n" f"{load_recent_episodes()}" ))

将新的用户消息添加到工作记忆中

working_memory.append(HumanMessage(content=user_input))

修剪工作记忆以保持在上下文窗口内

这会压缩较旧的消息，而不是完全删除它们

trimmed = trim_messages( working_memory, max_tokens=4000, strategy="last", # 保留最近的消息 token_counter=llm, # 使用模型的分词器进行准确计数 include_system=True, allow_partial=False )

使用系统上下文和修剪后的工作记忆调用模型

response = llm.invoke([system] + trimmed) reply = response.content

将对话保存到事件记忆中，并将回复添加到工作记忆中

save_episode(user_input, reply) working_memory.append(AIMessage(content=reply))

return reply

── 示例 ──────────────────────────────────────────────────────────────────────

if __name__ == "__main__": print(chat("我的名字是 Alex，我正在为法律文件构建一个 RAG 管道。")) print(chat("对于我的用例，最好的向量数据库是什么？")) print(chat("我告诉过你我正在构建什么吗？"))

使用当前 LangChain 0.3+ 模式（旧版的 ConversationBufferMemory 已弃用）

json

datetime

AIMessage

SystemMessage

trim_messages

0.2

── 事件记忆存储 ─────────────────────────────────────────────────────

在生产环境中，应将此列表替换为数据库（SQLite、Postgres、Redis）。

此处的结构：每个事件是一个字典，包含时间戳、用户输入和代理响应。

episodic_store

list

dict

def

save_episode

user_input

str

agent_response

->

None

""

"将对话轮次保存到事件存储中。"

append

"timestamp"

now

isoformat

"user"

"agent"

response

load_recent_episodes

n

int

"检索最近的 N 个事件，并以格式化字符串形式返回，用于注入上下文。"

if

not

return

"没有之前的对话历史。"

recent

"\n"

join

f

"[{ep['timestamp']}] 用户: {ep['user']} | 代理: {ep['agent']}"

for

ep

── 工作记忆（上下文内） ───────────────────────────────────────────────

我们自行管理消息列表，并在每次 LLM 调用前通过 trim_messages 进行处理，以保持在模型上下文限制内。

max_tokens=4000 为模型的响应预留了空间。

working_memory

chat

"

向代理发送一条消息。

将事件历史注入系统提示中。

在每次调用前修剪工作记忆，以防止上下文溢出。

将事件记忆注入系统提示中，使模型具有长期上下文

system

"你是一个有帮助且上下文感知的助手。\n\n"

"最近的对话历史:\n"

"{load_recent_episodes()}"

将新的用户消息添加到工作记忆中

修剪工作记忆以保持在上下文窗口内

这会压缩较旧的消息，而不是完全删除它们

trimmed

max_tokens

4000

strategy

"last"

保留最近的消息

token_counter

使用模型的分词器进行准确计数

include_system

True

allow_partial

False

使用系统上下文 + 修剪后的工作记忆调用模型

+

reply

将对话保存到事件记忆中，并将回复添加到工作记忆中

── 示例 ──────────────────────────────────────────────────────────────────────

__name__

==

"__main__"

"我的名字是 Alex，我正在为法律文件构建一个 RAG 管道。"

"我用例的最佳向量数据库是什么？"

"我告诉过你我正在构建什么？"

测试事件回忆

这是做什么的：事件存储作为轻量级持久日志，在每次调用时被汇总到系统提示中。工作记忆列表保存会话中的消息历史，并在每次 LLM 调用前通过 trim_messages 进行修剪，以防止令牌溢出。最后的测试问题，“我告诉过你我正在构建什么？”验证了事件回忆是否在上下文窗口移动后仍能正常工作。

第四层：向量数据库和检索（RAG）

基础模型知道很多，但它们并不知道你的文档。它们没有在你的内部知识库、客户支持历史、专有研究或自训练截止日期以来发生的事情上进行训练。检索增强生成（RAG）就是用来解决这个问题的方法。

这个概念非常简单：与其试图将整个知识库放入上下文窗口中，不如将文档转换为数值表示（嵌入），存储到向量数据库中，并在查询时仅检索最相关的内容块。代理获得的上下文窗口中只包含精确所需的信息，而不是你曾经写过的所有内容。

全球向量数据库市场在2025年达到32亿美元，并以每年24%的速度增长，这反映了检索在生产AI系统中的核心地位。

主要选项各有不同的使用场景：

• Pinecone 是一个完全托管的解决方案，没有基础设施开销。你只需付费、向其推送向量并进行查询。在拥有1亿个向量时，它无需调整即可保持召回率。当你希望快速部署而不必考虑基础设施时，Pinecone 是最佳选择。

• Weaviate 是开源的，并提供托管云选项，它在混合搜索方面领先，可以在一个查询中结合向量相似性、关键词匹配（BM25）和元数据过滤。如果你的检索需求需要的不仅仅是纯语义搜索，Weaviate 可以原生处理这些需求。

• Chroma 以开发者为中心，可在本地运行且无需基础设施。2025年的Rust重写使其显著加快。对于原型设计和中小型生产工作负载，当开发者体验比规模更重要时，Chroma 是最佳选择。

• pgvector 是一个 PostgreSQL 扩展，为数据库添加了向量搜索功能，你可能已经在运行该数据库。如果你的团队已经在使用PostgreSQL，pgvector 是实现生产RAG的最低摩擦路径。它使用HNSW索引处理数百万个向量，并且对于大多数生产工作负载，它仍保持在单节点PostgreSQL的限制内。

一个水平的三步流程图展示了RAG管道：文档 → 嵌入模型 → 向量数据库（点击放大）

下面是一个使用Chroma和OpenAI嵌入的可运行RAG管道。

pip install langchain langchain-openai langchain-chroma langchain-text-splitters chromadb python-dotenv

chroma

text

splitters

chromadb

如何运行：保存为 rag_pipeline.py，将 OPENAI_API_KEY 添加到你的 .env 文件中，然后运行 python rag_pipeline.py。

rag_pipeline.py

最小的 RAG 管道：导入文档 → 嵌入 → 存储到 Chroma → 检索并回答

Python 3.10+ | ChromaDB 0.5+ | LangChain 0.3+

import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_chroma import Chroma from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_core.documents import Document from langchain_core.prompts import ChatPromptTemplate

load_dotenv()

── 第 1 步：示例文档 ────────────────────────────────────────────────────────

用知识库中的真实文档替换这个列表。

在生产环境中，可以从 PDF、数据库、API 或文件系统加载文档。

documents = [ Document(page_content="Pinecone 是一个托管的向量数据库，针对大规模快速、低延迟相似性搜索进行了优化。它自动处理基础设施，适合生产环境中的 RAG，当你不想管理服务器时。", metadata={"source": "vector_db_guide", "topic": "pinecone"}), Document(page_content="Weaviate 是一个开源的向量数据库，原生支持混合搜索，可以在单个查询中结合 BM25 关键字搜索和密集向量搜索。它可以通过自托管或 Weaviate Cloud 使用。", metadata={"source": "vector_db_guide", "topic": "weaviate"}), Document(page_content="Chroma 是一个开发者友好、本地优先的向量数据库，非常适合原型设计。2025 年的 Rust 重写显著提升了性能。最适合中小型生产工作负载和本地开发。", metadata={"source": "vector_db_guide", "topic": "chroma"}), Document(page_content="pgvector 是一个 PostgreSQL 扩展，为现有的 Postgres 数据库添加了向量相似性搜索功能。通过 HNSW 索引，它可以在低延迟下处理数百万个向量。如果团队已经在生产环境中运行 PostgreSQL，这是最佳选择。", metadata={"source": "vector_db_guide", "topic": "pgvector"}), ]

── 第 2 步：对文档进行分块 ───────────────────────────────────────────────────

在嵌入之前，将大文档拆分成较小的块。

chunk_size=500 字符；chunk_overlap=50 保留块边界之间的上下文。

splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50) chunks = splitter.split_documents(documents)

── 第 3 步：嵌入并存储到 Chroma ───────────────────────────────────────────────

OpenAIEmbeddings 将每个块转换为高维向量。

Chroma 将这些向量本地存储在 ./chroma_db 目录中。

在后续运行中，将加载现有存储而不是重新构建。

embeddings = OpenAIEmbeddings( model="text-embedding-3-small", # 对大多数 RAG 任务快速且成本效益高 api_key=os.getenv("OPENAI_API_KEY") )

vectorstore = Chroma.from_documents( documents=chunks, embedding=embeddings, persist_directory="./chroma_db" # 持久化到磁盘，避免每次运行都重新嵌入 )

── 第 4 步：检索 ─────────────────────────────────────────────────────────────

将查询转换为嵌入并找到最相似的块。

k=3 返回最相关的前 3 个块。

retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

── 第 5 步：生成 ─────────────────────────────────────────────────────────────

llm = ChatOpenAI( model="gpt-5.5", temperature=0, api_key=os.getenv("OPENAI_API_KEY") )

提示告诉模型只使用检索到的上下文。

这可以防止模型生成知识库中没有的事实。

rag_prompt = ChatPromptTemplate.from_messages([ ("system", "仅使用提供的上下文回答问题。" "如果答案不在上下文中，请明确说明。\n\n" "上下文：\n{context}"), ("human", "{question}") ])

def answer(question: str) -> str: """检索相关块并生成基于上下文的答案。"""

检索与该问题最相关的文档块

retrieved_docs = retriever.invoke(question)

将检索到的块合并为一个上下文块

context = "\n\n".join(doc.page_content for doc in retrieved_docs)

使用问题和检索到的上下文构建并调用提示

prompt = rag_prompt.invoke({"context": context, "question": question}) response = llm.invoke(prompt) return response.content

── 示例 ─────────────────────────────────────────────────────────────────────

if __name__ == "__main__": q = "如果我已经在运行 PostgreSQL，我应该使用哪个向量数据库？" print(f"Q: {q}\nA: {answer(q)}")

92

93

94

95

96

97

98

99

100

rag_pipeline.py

最小的 RAG 管道：导入文档 → 嵌入 → 存储到 Chroma → 检索并回答

Python 3.10+ | ChromaDB 0.5+ | LangChain 0.3+

OpenAIEmbeddings

langchain_chroma

langchain_text_splitters

RecursiveCharacterTextSplitter

documents

Document

prompts

ChatPromptTemplate

── 第 1 步：示例文档 ──────────────────────────────────────────────────

请将此列表替换为知识库中的真实文档。

在生产环境中，可以从 PDF、数据库、API 或文件系统中加载。

page_content

"Pinecone 是一个托管的向量数据库，专为大规模的快速、低延迟相似性搜索进行了优化。它自动处理基础设施，"

"在不需要管理服务器的生产 RAG 场景中表现最佳。"

metadata

"source"

"vector_db_guide"

"topic"

"pinecone"

"Weaviate 是一个开源的向量数据库，原生支持混合搜索，"

"将 BM25 关键字搜索与密集向量搜索结合在一个查询中。"

"它可以自托管，也可以通过 Weaviate Cloud 使用。"

"weaviate"

"Chroma 是一个开发者友好、以本地优先的向量数据库，非常适合原型设计。"

"2025 年的 Rust 重写显著提升了性能。"

"最适合中小型生产工作负载和本地开发。"

"chroma"

"pgvector 是一个 PostgreSQL 扩展，为现有的 PostgreSQL 数据库添加了向量相似性搜索功能。"

"通过 HNSW 索引，它可以在低延迟下处理数百万个向量。"

"如果团队已经在生产环境中运行 PostgreSQL，这是最佳选择。"

"pgvector"

── 第 2 步：对文档进行分块 ───────────────────────────────────────────────

大型文档在嵌入之前会被拆分成较小的块。

chunk_size=500 字符；chunk_overlap=50 保留块边界之间的上下文。

splitter

chunk_size

500

chunk_overlap

chunks

split_documents

── 第 3 步：嵌入并存储到 Chroma ────────────────────────────────────────

OpenAIEmbeddings 将每个块转换为高维向量。

Chroma 将这些向量本地存储在 ./chroma_db 目录中。

在后续运行中，会加载现有存储，而不是重新构建。

embeddings

"text-embedding-3-small"

对大多数 RAG 任务来说，速度快且成本效益高

vectorstore

from_documents

embedding

persist_directory

"./chroma_db"

持久化到磁盘，这样就不需要在每次运行时重新嵌入

── 第 4 步：检索 ──────────────────────────────────────────────────────────

将查询转换为嵌入，并找到最相似的块。

k=3 返回最相关的前 3 个块。

retriever

as_retriever

search_kwargs

"k"

── 第 5 步：生成 ────────────────────────────────────────────────────────

提示告诉模型只使用检索到的上下文。

这可以防止模型生成知识库中没有的事实。

rag_prompt

from_messages

"system"

"仅使用提供的上下文回答问题。"

"如果答案不在上下文中，请明确说明。\n\n"

"上下文：\n{context}"

"human"

"{question}"

answer

question

"检索相关块并生成一个基于上下文的答案。"

检索与此问题最相关的文档块

retrieved_docs

将检索到的块合并成一个上下文块

context

"\n\n"

doc

使用问题和检索到的上下文构建并调用提示

prompt

"context"

"question"

q

"如果我已经在运行 PostgreSQL，我应该使用哪种向量数据库？"

"Q: {q}\nA: {answer(q)}"

这做了什么：该流程分为两个阶段。在索引阶段，文档会被分割成块，通过 OpenAI 的 text-embedding-3-small 模型转换为嵌入向量，并存储在本地的 Chroma 数据库中。在检索阶段，查询会使用相同的模型进行嵌入，从 Chroma 中提取出最相似的三个块，然后 LLM 使用这些块（仅这些块）来回答问题。persist_directory 参数意味着 Chroma 会将向量保存到磁盘，因此你不需要在每次运行时都重新嵌入文档，从而节省成本。

第五层：工具和外部集成

没有工具的代理只是一个非常昂贵的文本预测器。工具赋予代理在世界上采取行动的能力，而不仅仅是谈论它。

从技术上讲，工具是模型可以选择调用的函数。你用自然语言描述该函数的功能，用模式定义其输入参数，模型决定何时调用该函数有助于回答问题。模型本身不会执行该函数；你的代码会执行。模型只是决定何时调用以及使用哪些参数。

在生产环境中，对代理最重要的工具类别包括：网络搜索（用于获取当前信息）、代码执行（用于计算和数据处理）、文件 I/O（用于读取和写入文档）、API 调用（用于连接外部服务）以及浏览器使用（用于与没有 API 的网页界面进行交互）。

一个值得关注的发展是 Model Context Protocol（MCP），由 Anthropic 于 2024 年底引入。MCP 是一种标准化的方式，使模型能够与外部工具和数据源进行通信。而不是每个团队为每个工具编写自定义的集成代码，MCP 提供了一种共享协议。Amazon Bedrock Agents 在 2025 年增加了对 MCP 的原生支持，生态系统中的采用率正在迅速增长。

工具设计中最重要的事情是模式。模型根据描述决定是否使用工具，并根据参数模式决定传递哪些参数。描述模糊会导致错误的工具调用。而一个类型明确、参数描述清晰的模式会产生可靠的调用。

pip install langchain langchain-openai langchain-community python-dotenv

如何运行：保存为 tools.py，将 OPENAI_API_KEY 添加到你的 .env 文件中，然后运行 python tools.py。

tools.py

使用 LangChain 代理定义、注册和使用工具

Python 3.10+ | LangChain 0.3+

import os
import json
import requests
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_core.messages import HumanMessage
from langgraph.prebuilt import create_react_agent

load_dotenv()
llm = ChatOpenAI(model="gpt-5.5", temperature=0, api_key=os.getenv("OPENAI_API_KEY"))

── 工具 1：网络搜索 ────────────────────────────────────────────────────────

内置的 DuckDuckGo 工具，不需要 API 密钥。

search = DuckDuckGoSearchRun()

── 工具 2：天气查询 ────────────────────────────────────────────────────

`@tool` 装饰器有三个作用：

1. 将函数注册为可调用工具

2. 使用函数名称作为工具名称

3. 使用文档字符串作为工具描述（这是模型读取的内容）

描述非常重要 —— 模糊的描述会导致错误的工具调用。

@tool
def get_weather(city: str) -> str:
    """获取给定城市的当前天气。当用户询问天气状况、温度或预报时使用此工具。

    输入：城市名称作为字符串（例如，'London', 'Tokyo', 'New York'）。
    """
    try:
        # 使用 open-meteo（免费，无需 API 密钥）进行地理编码和天气查询
        geo_url = f"https://geocoding-api.open-meteo.com/v1/search?name={city}&count=1"
        geo = requests.get(geo_url, timeout=5).json()
        if not geo.get("results"):
            return f"无法找到 '{city}' 的位置数据。"
        lat = geo["results"][0]["latitude"]
        lon = geo["results"][0]["longitude"]
        weather_url = (
            f"https://api.open-meteo.com/v1/forecast"
            f"?latitude={lat}&longitude={lon}"
            f"¤t_weather=true"
        )
        weather = requests.get(weather_url, timeout=5).json()
        current = weather.get("current_weather", {})
        return (
            f"{city} 的天气："
            f"{current.get('temperature', 'N/A')}°C, "
            f"风速 {current.get('windspeed', 'N/A')} km/h。"
        )
    except Exception as e:
        # 即使在失败时，工具也应始终返回字符串。
        # 从工具中引发异常可能会导致代理循环崩溃。
        return f"获取 '{city}' 的天气信息失败：{str(e)}"

── 工具 3：JSON 计算器 ───────────────────────────────────────────────────

@tool
def calculate(expression: str) -> str:
    """评估一个数学表达式并返回结果。用于算术、百分比计算或任何数值计算。

    输入：一个有效的 Python 数学表达式字符串（例如，'(150 * 1.08) / 12'）。
    不要用于复杂代码执行，仅用于简单的数学表达式。
    """
    try:
        # `eval` 仅允许数学操作 —— 不允许内置函数，不允许导入
        result = eval(expression, {"__builtins__": {}}, {})
        return f"结果：{result}"
    except Exception as e:
        return f"计算错误：{str(e)}"

── 注册工具并构建代理 ────────────────────────────────────────────

tools = [search, get_weather, calculate]
agent = create_react_agent(llm, tools)

── 示例 ──────────────────────────────────────────────────────────────────────

if __name__ == "__main__":
    queries = [
        "Lagos 当前的天气如何？",
        "如果我每年赚 85,000 美元，我的月毛收入是多少？",
        "谁赢得了最近的 FIFA 世界杯？"
    ]
    for query in queries:
        print(f"\n问题：{query}")
        result = agent.invoke({"messages": [HumanMessage(content=query)]})
        print(f"答案：{result['messages'][-1].content}")

Python 3.10+ | LangChain 0.3+

requests

tool

── 工具 1: 网络搜索 ────────────────────────────────────────────────────────

内置的 DuckDuckGo 工具 -- 不需要 API 密钥。

search

── 工具 2: 天气查询 ────────────────────────────────────────────────────────

@tool 装饰器有三个作用：

1. 将函数注册为可调用的工具

2. 使用函数名称作为工具名称

3. 使用文档字符串作为工具描述（这是模型读取的内容）

描述至关重要 -- 模糊的描述会导致错误的工具调用。

@

get_weather

city

获取给定城市的当前天气。

当用户询问天气状况、温度或预报时使用此工具。

输入：城市名称作为字符串（例如 'London', 'Tokyo', 'New York'）。

try

使用 open-meteo（免费，无需 API 密钥）进行地理编码和天气查询

geo_url

"https://geocoding-api.open-meteo.com/v1/search?name={city}&count=1"

geo

get

timeout

"results"

"无法为 '{city}' 找到位置数据。"

lat

"latitude"

lon

"longitude"

weather_url

"https://api.open-meteo.com/v1/forecast"

"?latitude={lat}&longitude={lon}"

"¤t_weather=true"

weather

current

"current_weather"

"{city} 的天气： "

"{current.get('temperature', 'N/A')}°C, "

"风速 {current.get('windspeed', 'N/A')} km/h。"

except

Exception

as

e

即使失败，工具也应始终返回字符串。

从工具中引发异常可能会导致代理循环崩溃。

"为 '{city}' 的天气查询失败：{str(e)}"

── 工具 3: JSON 计算器 ───────────────────────────────────────────────────────

calculate

expression

评估一个数学表达式并返回结果。

用于算术、百分比计算或任何数值计算。

输入：一个有效的 Python 数学表达式作为字符串（例如 '(150 * 1.08) / 12'）。

不要用于复杂代码执行 -- 仅用于简单的数学表达式。

eval 仅允许数学运算，不包含内置函数，不包含导入

eval

"__builtins__"

"结果：{result}"

"计算错误：{str(e)}"

── 注册工具并构建代理 ───────────────────────────────────────────────────────

queries

"What is the weather in Lagos right now?"

"If I earn $85,000 a year, what is my monthly gross salary?"

"Who won the most recent FIFA World Cup?"

query

"\nQuery: {query}"

"Answer: {result['messages'][-1].content}"

这是它所做的：注册了三个工具：一个用于当前事件的网络搜索工具，一个调用免费 API（无需密钥）的天气工具，以及一个安全评估数学表达式的计算器。代理接收每个查询，推理使用哪个工具，调用它，并从结果中综合出答案。需要注意的关键设计细节是在文档字符串中；每个工具的描述都明确说明了工具的功能、使用时机以及输入应采用的格式。

第六层：可观测性和评估

这里是一个在生产环境中很少被提及但非常重要的事实：LLMs 会静默失败。正如 Kanerika 团队所说，一个幻觉的答案仍然会返回 HTTP 200。一个标准的基础设施监控工具会看到一个成功的请求。你不会看到任何异常。与此同时，你的代理已经连续三天自信地给出错误答案。

传统监控是为一个“正确”是二元的世界而设计的：函数返回了正确的类型，API 返回了 200，查询在 100ms 内完成。LLM 的正确性是语义层面的。响应可能是结构上有效的、语法流畅的，但完全错误。这需要一个完全不同的可观测性层。

一个良好的 LLM 可观测性设置应跟踪三件事。追踪会跟踪代理执行的每一步：LLM 调用、工具调用、检索查询、中间推理步骤以及每一步所花费的时间。评估会根据重要的指标对输出进行评分：忠实性（是否保持在检索到的上下文中？）、相关性（是否回答了所提出的问题？）和幻觉率。监控会跟踪随时间推移的行为偏移，无论模型和提示如何演变，代理在某一类输入上的表现是变好还是变差。

领先的平台各有其优势。LangSmith 与 LangChain 和 LangGraph 的集成最深入。如果你已经在该生态系统中，它就是实现追踪的最快途径。Langfuse 是开源的，GitHub 上有超过 19,000 颗星，采用 MIT 许可证，可自托管，并且适用于任何框架。Arize Phoenix 带来了机器学习级别的评估严谨性，附带了超过 50 个研究支持的指标，涵盖忠实性、相关性、安全性和幻觉检测。

根据 MLflow 对可观测性平台的分析，正确的选择通常取决于你使用的框架：使用 LangChain 的团队从 LangSmith 中获益最多，而使用 LlamaIndex 或原始 API 调用的团队则更适合使用 Phoenix 或 Langfuse。

以下是如何在现有代理中以最小的改动添加 Langfuse 追踪。

pip install langfuse langchain langchain-openai python-dotenv

langfuse

在 langfuse.com 注册一个免费账户，并将 LANGFUSE_PUBLIC_KEY 和 LANGFUSE_SECRET_KEY 添加到你的 .env 文件中。如果你希望在自己的基础设施上保留数据，也可以选择自托管。

如何运行：将代码保存为 observability.py 并运行 python observability.py。打开你的 Langfuse 仪表板查看追踪。

observability.py

将 Langfuse 追踪添加到 LangChain 代理

Langfuse 会自动捕获每个 LLM 调用、工具调用和令牌计数。

Langfuse 通过 CallbackHandler 模式进行集成。

它会拦截每个 LangChain 事件，并将其发送到你的 Langfuse 仪表板。

CallbackHandler

── LANGFUSE 配置 ─────────────────────────────────────────────────────────────

CallbackHandler 从环境变量中读取 LANGFUSE_PUBLIC_KEY 和 LANGFUSE_SECRET_KEY。

session_id 将所有相关追踪分组到一个会话中 —— 适用于调试对话。

user_id 将追踪与特定用户关联，用于按用户进行性能分析。

langfuse_handler

session_id

"demo_session_001"

user_id

"demo_user"

── 代理配置 ────────────────────────────────────────────────────────────────

callbacks

在此处附加处理程序 —— 这是唯一需要更改的地方

── 带追踪的运行 ─────────────────────────────────────────────────────────────

在配置中传递处理程序，以便追踪工具调用以及 LLM 调用。

没有这个，只有 LLM 调用会被追踪 —— 工具调用将不可见。

"What is the latest version of Python?"

config

"callbacks"

刷新确保在脚本退出之前所有追踪都被发送到 Langfuse。

在长期运行的服务器中，这会自动处理。

flush

"\n追踪已发送到 Langfuse。请在 https://cloud.langfuse.com 查看你的仪表板"

这会带来什么变化：与标准代理设置相比，有两个变化：CallbackHandler 是使用会话和用户 ID 初始化的，并且它被附加到 LLM 和 agent.invoke 配置中。这已经足够让 Langfuse 捕获每个 LLM 调用、每个工具调用、令牌计数、延迟以及每一步的完整输入/输出。这些是你调试生产故障或跟踪随时间推移的质量变化所需的一切。

第 7 层：部署基础设施

你可以在开发阶段拥有一个完美的代理，但在生产环境中却变成一个维护问题。基础设施层就是这个差距所在。

至少，你的代理应该使用 Docker 进行容器化。容器可以让你在不同环境中行为保持一致，简化依赖管理，并提供一条清晰的路径通往任何云部署目标。另一种选择——使用 requirements.txt 发送 Python 脚本并希望环境匹配——会带来一类浪费工程时间的错误，其代价远大于容器化所需的投入。

对于大多数生产代理，服务层有两种架构选择：同步 API 或异步队列。同步 API（如 Flask 或 FastAPI）适用于代理在几秒内完成的情况，并且你可以承担保持 HTTP 连接打开的开销。

当你的代理涉及多个工具调用、耗时较长的检索流程或可能需要 30 到 60 秒的文档处理时，异步队列（如 Celery、AWS SQS 或 Google Pub/Sub）是更好的选择。客户端提交一个任务，立即获得一个任务 ID，然后轮询结果。

在云方面，目前三大平台都提供了托管代理基础设施。Amazon 的 AgentCore 于 2025 年 10 月正式发布，提供在 AWS 上的专用代理基础设施，用于内存管理、工具执行和会话处理，无需部署服务器。Google Vertex AI Agent Builder 是已经在 GCP 生态系统中的团队的自然选择，具有原生的 Gemini 集成和内置的可观测性。Azure OpenAI 服务与 Semantic Kernel 是微软企业环境的默认选择。

在成本管理方面，有三种做法能产生最大的影响：缓存（对重复的相同查询返回存储的响应，而不是再次调用模型）、请求批处理（将非紧急任务分组以减少每次调用的开销），以及在代理执行器中设置 max_iterations 以防止无限制循环消耗无限制的令牌。

一个垂直堆栈图，从上到下标记了所有 7 层：基础模型、编排框架、内存系统、向量数据库和 RAG、工具和集成、可观测性和评估、部署基础设施（点击放大）

综合来看

每一层的正确选择取决于你在项目生命周期中的位置。以下是一个实用的参考指南，反映了上述研究和权衡。

原型（快速移动，基础设施最少）：

| 层级 | 选择 | 原因 | |------|------|------| | 基础模型 | GPT-5.5 | 可靠的工具调用，成熟的生态系统 | | 编排 | 快速设置，良好的文档 | | | 内存 | 仅上下文 | 无需基础设施 | | 向量数据库 | 本地，无需运维，良好的开发体验 | DuckDuckGo + 自定义 @tool 函数 | | 可观测性 | Langfuse（云免费层） | 一行代码即可设置 | | 部署 | 本地 / Docker | 快速部署 |

可靠性与冗余

LangGraph 或 CrewAI

状态管理与多代理支持

情景记忆（Postgres）+ 语义记忆（RAG）

完整的持久化上下文

Weaviate 或 Pinecone

可扩展性与混合搜索

完整的工具套件，包含 MCP

标准化集成

Langfuse 自托管或 Arize Phoenix

数据控制 + 机器学习级别的评估

Docker + Kubernetes + 异步队列

生产级，成本可控

企业级：

Azure OpenAI 或 AWS Bedrock

合规性、数据驻留、SLA

语义内核（Semantic Kernel）或 LangGraph

企业语言支持、治理

带有审计追踪的托管内存

监管要求

Weaviate 或 pgvector

可自托管、符合合规要求

基于 MCP、内部批准

安全审查与访问控制

Langfuse 自托管或 Datadog LLM 模块

现有基础设施集成

AWS AgentCore / Vertex AI Agent Builder

完全托管、受控、可审计

结论

基础模型是这个堆栈中被广泛讨论的部分。其他六层决定了你构建的内容是否真的能在生产环境中正常运行。

当 ReAct 循环陷入停滞时，代理在编排层失败。当它忘记需要的上下文时，代理在记忆层失败。当检索层返回了错误的片段，模型产生一个听起来有根据但错误的答案时，代理在检索层失败。当工具层的模式过于模糊，模型调用了错误的函数时，代理在工具层失败。当无法知道这些情况是否发生时，代理在可观测性层失败。当基础设施无法处理真实流量的延迟或成本要求时，代理在部署层失败。

Gartner 估计，到 2027 年，超过 40% 的代理 AI 项目因价值不明确、成本上升和治理薄弱而面临被取消的风险。这些失败大多不是由于选择了错误的模型，而是由于在构建堆栈时，每一层之间如何连接缺乏清晰的认识。

理解整个堆栈并不意味着你必须构建所有部分。这意味着你知道你在做哪些决策，以及在做出这些决策时你所做出的权衡。这就是一个在演示中能工作的代理与一个能实际部署的代理之间的区别。

关于此主题的更多信息

• 项目聚焦：使用……的 Stack Exchange 聚类

• 三难理解：AI 代理的记忆

• 为初学者解释的六个语言模型概念

• 损失函数解释：理解数学中的……

• 七个大型语言模型背后的概念解释……

• 十个关键机器学习术语的解释