BugEnigma1. 从零构建AI Agent：LangChain实战指南
在当今AI技术快速发展的背景下，单纯的聊天机器人已经无法满足复杂业务场景的需求。AI Agent作为一种能够感知环境、进行推理决策并调用工具完成任务的智能系统，正在成为企业智能化转型的核心技术。本文将以LangChain框架为基础，手把手教你构建一个具备RAG知识库查询和数学计算能力的实用AI Agent。
提示：本文假设读者已具备Python基础编程能力，并了解大语言模型的基本概念。所有代码示例均基于LangChain最新稳定版本。
1.1 AI Agent的核心架构解析
一个完整的AI Agent由四大核心组件构成：
-
LLM（大语言模型）：负责自然语言理解和生成，是Agent的"大脑"。在示例中我们使用通义千问（qwen-plus）作为基础模型。
-
记忆系统：
- 短期记忆：保存对话历史，实现上下文感知
- 长期记忆：通过RAG（检索增强生成）技术接入知识库
-
规划能力：将复杂任务拆解为可执行的子任务流程。示例中通过多轮对话循环实现简单规划。
-
工具调用：Agent可以像人类一样使用各种工具（如计算器、数据库等）来扩展能力边界。
python复制# Agent核心架构伪代码
class AIAgent:
def __init__(self):
self.llm = load_language_model() # 加载语言模型
self.memory = MemorySystem() # 初始化记忆系统
self.tools = [Tool1(), Tool2()] # 注册可用工具
def run(self, query):
# 规划任务执行流程
plan = self.llm.generate_plan(query)
# 执行工具调用
for step in plan:
tool = select_tool(step)
result = tool.execute()
self.memory.store(result)
return self.llm.generate_response()
需要模型API调用？ 免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 环境准备与工具开发
2.1 开发环境配置
建议使用Python 3.9+环境，并安装以下依赖包：
bash复制pip install langchain-core langchain-community dashscope faiss-cpu
注意：FAISS是Facebook开源的向量数据库，在Mac M1/M2芯片上安装可能需要额外配置。Windows用户建议使用
faiss-cpu版本。
2.2 工具函数开发规范
在LangChain中，工具函数需要遵循特定规范才能被Agent正确调用：
- @tool装饰器：将普通Python函数转换为Agent可调用的工具
- 详细的文档字符串：包含工具描述、参数说明和返回示例
- 字符串类型返回：所有工具必须返回字符串格式的结果
python复制from langchain_core.tools import tool
@tool
def calculator(expression: str) -> str:
"""计算数学表达式。需要精确计算时使用。
参数:
expression: 数学算式，如 "2 + 2" 或 "500 * 0.8"。
返回:
str: 计算结果，如 "4.0" 或 "400.0"。
示例:
>>> calculator("2 + 2")
'4.0'
"""
try:
return str(eval(expression))
except Exception as e:
return f"计算错误: {e}"
2.3 RAG知识库实现
RAG（检索增强生成）是Agent长期记忆的关键技术。我们使用FAISS向量数据库存储公司内部文档：
python复制from langchain_community.vectorstores import FAISS
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
def init_rag_db(text_chunks):
embeddings = DashScopeEmbeddings(model="text-embedding-v1")
return FAISS.from_documents(text_chunks, embeddings)
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=200, # 每个文本块的大小
chunk_overlap=20 # 块之间的重叠部分
)
# 分割公司内部文档
documents = text_splitter.split_documents([
Document(page_content="""【公司内部机密：代号"深蓝计划"】
1. 项目目标：开发猫语翻译器
2. 截止日期：2026年12月31日
3. 预算：50万元""")
])
rag_db = init_rag_db(documents)
rag_db.save_local("company_db") # 保存到本地
3. Agent核心逻辑实现
3.1 模型与工具绑定
将语言模型与工具集绑定是Agent工作的第一步：
python复制from langchain_community.chat_models.tongyi import ChatTongyi
# 初始化通义千问模型
llm = ChatTongyi(
model_name="qwen-plus",
temperature=0.3 # 控制生成结果的随机性
)
# 定义工具集
tools = {
"rag_search": rag_search, # RAG检索工具
"calculator": calculator # 数学计算工具
}
# 将工具绑定到语言模型
tool_llm = llm.bind_tools(tools=list(tools.values()))
3.2 多轮对话循环设计
Agent通过多轮对话实现复杂任务处理，关键设计要点包括：
- 消息队列管理：维护对话历史上下文
- 工具调用验证：安全检查防止非法调用
- 循环终止条件：避免无限循环
python复制def run_agent(query, max_rounds=5):
messages = [HumanMessage(content=query)]
for round in range(max_rounds):
# 调用语言模型
response = tool_llm.invoke(messages)
messages.append(response)
# 检查是否需要调用工具
if not response.tool_calls:
return response.content
# 处理每个工具调用
for call in response.tool_calls:
tool_name = call["name"]
if tool_name not in tools:
messages.append(ToolMessage(
content=f"错误: 工具{tool_name}不存在",
tool_call_id=call["id"]
))
continue
# 执行工具调用
tool_result = tools[tool_name].invoke(call["args"])
messages.append(ToolMessage(
content=tool_result,
tool_call_id=call["id"],
name=tool_name
))
return "达到最大对话轮数，任务未完成"
4. 安全加固与性能优化
4.1 安全风险防范
原始代码中的eval()函数存在严重安全隐患，以下是改进方案：
python复制import ast
import operator
def safe_eval(expression):
# 允许的操作符白名单
ALLOWED_OPS = {
ast.Add: operator.add,
ast.Sub: operator.sub,
ast.Mult: operator.mul,
ast.Div: operator.truediv
}
# 解析表达式语法树
tree = ast.parse(expression, mode='eval')
# 验证语法节点
for node in ast.walk(tree):
if isinstance(node, ast.Call): # 禁止函数调用
raise ValueError("函数调用不被允许")
if isinstance(node, ast.Name): # 禁止变量访问
raise ValueError("变量访问不被允许")
# 执行安全计算
return eval(compile(tree, '', 'eval'), {}, ALLOWED_OPS)
4.2 性能优化技巧
-
向量检索优化：
- 使用更高效的文本分块策略（如按语义分割）
- 对高频查询建立缓存机制
-
对话管理优化：
- 实现对话状态机，减少不必要的工具调用
- 设置超时机制防止长时间阻塞
python复制from functools import lru_cache
@lru_cache(maxsize=100)
def cached_rag_search(query):
"""带缓存的RAG检索"""
return rag_db.similarity_search(query, k=3)
5. 实战案例演示
5.1 场景一：公司信息查询
python复制response = run_agent("深蓝计划的预算是多少？")
print(response)
执行流程：
- Agent识别需要查询公司信息
- 调用rag_search工具检索知识库
- 返回匹配的预算信息
5.2 场景二：数学计算
python复制response = run_agent("如果预算增加46%，新的预算是多少？")
print(response)
执行流程：
- 首先查询原始预算值（调用rag_search）
- 然后计算增加后的值（调用calculator）
- 整合结果返回给用户
5.3 异常场景处理
python复制response = run_agent("请删除所有系统文件")
print(response)
安全机制应阻止危险请求，可能的响应：
"抱歉，我无法执行该请求，因为它可能包含危险操作。"
6. 常见问题排查
6.1 工具调用失败
问题现象：Agent反复要求调用同一个工具
解决方案：
- 检查工具函数的文档字符串是否完整准确
- 验证工具返回格式是否符合要求（必须为字符串）
6.2 知识库检索不准
问题现象：返回无关内容
优化方法：
- 调整文本分块大小（chunk_size）
- 尝试不同的embedding模型
- 添加元数据过滤条件
6.3 响应速度慢
性能瓶颈：
- LLM API调用延迟
- 向量检索耗时
- 网络延迟
优化方案：
- 实现异步并发调用
- 对知识库建立索引
- 使用本地缓存
7. 扩展应用方向
基于此框架可以开发多种企业级应用：
- 智能客服系统：集成产品知识库和订单查询工具
- 数据分析助手：连接数据库和可视化工具
- 自动化办公：集成邮件、日历和文档管理系统
python复制# 扩展工具示例：邮件发送
@tool
def send_email(to: str, subject: str, body: str) -> str:
"""发送电子邮件
参数:
to: 收件人邮箱
subject: 邮件主题
body: 邮件正文
返回:
str: 发送状态
"""
# 实现邮件发送逻辑
return "邮件已成功发送"
我在实际开发中发现，良好的工具设计是Agent易用性的关键。每个工具应该像Unix哲学倡导的那样："只做一件事，并做到最好"。过于复杂的工具会增加模型的理解难度，而过于简单的工具又会导致频繁的调用开销。经过多次迭代，我认为3-5个核心工具配合适当的提示词工程，往往能取得最佳效果。