1. 从零开始理解LangChain中的AI Agent
AI Agent这个概念最近在技术圈里越来越火,但很多人对它的理解还停留在"能聊天的机器人"这个层面。实际上,一个真正的AI Agent更像是一个数字世界的智能助手,它不仅能理解你的需求,还能自主调用各种工具完成任务。我在实际项目中发现,很多开发者对LangChain框架下的Agent开发存在不少困惑,今天我就结合一个公司内部信息查询的案例,带大家彻底搞懂Agent的实现原理和开发技巧。
这个案例中的Agent有两个核心能力:一是能查询公司内部知识库(RAG实现),二是能进行精确数学计算。这种组合在实际业务场景中非常实用,比如财务分析、项目进度跟踪等场景。下面我会从基础概念开始,逐步拆解代码实现,最后分享几个我在开发过程中踩过的坑。
2. AI Agent的核心组件解析
2.1 Agent的四大支柱
一个完整的AI Agent通常由四个关键部分组成,就像搭建一座房子需要四根支柱:
-
LLM(大语言模型):这是Agent的"大脑",负责理解和生成语言。在案例中我们使用的是通义千问(qwen-plus)模型。
-
记忆系统:
- 短期记忆:保存当前对话的上下文(案例中的message数组)
- 长期记忆:通过RAG技术接入公司知识库
-
规划能力:Agent能拆解复杂任务为多个步骤。案例中通过多轮对话循环实现这一点。
-
工具调用:Agent可以像人类使用工具一样调用外部函数。案例中实现了计算器和文档搜索两个工具。
提示:在实际开发中,工具的描述文档非常重要。LLM完全依赖这些描述来决定何时以及如何使用工具,所以一定要写得详细准确。
2.2 工具调用的实现细节
工具函数的定义有几个关键点需要注意:
python复制@tool
def calculator(expression: str) -> str:
"""
计算数学表达式。需要精确计算时使用。
参数:
expression: 数学算式,如 "2 + 2" 或 "500 * 0.8"。
返回:
str: 计算结果,如 "4.0" 或 "400.0"。
"""
print(f" [工具调用] 计算器正在计算: {expression}")
try:
return str(eval(expression))
except Exception as e:
return f"计算错误: {e}"
- 必须使用
@tool装饰器标记 - 函数文档字符串要包含三部分:功能描述、参数示例、返回示例
- 返回值必须是字符串类型(因为LLM只能处理文本)
- 工具函数内部可以做任何Python能做的事,但要特别注意安全性(后面会详细讲)
3. 完整案例实现与代码解析
3.1 项目初始化与环境配置
首先需要安装必要的依赖库:
bash复制pip install langchain-core langchain-community dashscope faiss-cpu
然后设置API密钥(以阿里云DashScope为例):
python复制import os
os.environ["DASHSCOPE_API_KEY"] = "your_api_key_here"
3.2 RAG知识库的实现
案例中使用FAISS作为向量数据库,这是Meta开源的轻量级向量搜索引擎,特别适合本地开发环境:
python复制from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
raw_text = """【公司内部机密:代号"深蓝计划"】
1. 项目目标:开发一款能听懂猫语的翻译器。
2. 核心技术:基于Transformer的"喵声波"分析算法。
..."""
def init_rag_db():
docs = [Document(page_content=raw_text)]
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=25,
chunk_overlap=5
)
split_docs = text_splitter.split_documents(docs)
embeddings = DashScopeEmbeddings(model="text-embedding-v1")
if os.path.exists("faiss_index"):
return FAISS.load_local("faiss_index", embeddings, allow_dangerous_deserialization=True)
else:
db = FAISS.from_documents(split_docs, embeddings)
db.save_local("faiss_index")
return db
这里有几个值得注意的技术点:
-
文本分块:使用RecursiveCharacterTextSplitter将长文档切分为小片段,chunk_size控制每个片段的大小,chunk_overlap设置片段间的重叠部分。
-
向量化:使用DashScope的文本嵌入模型将文本转换为向量。
-
本地存储:FAISS索引可以保存到本地,避免每次重启程序都重新构建。
3.3 Agent的核心执行流程
Agent的执行遵循一个标准的循环模式:
python复制def run_agent(query: str):
tool_maps = {
"rag_search": rag_search,
"calculator": calculator
}
llm = ChatTongyi(model_name="qwen-plus")
tool_llm = llm.bind_tools(tools=list(tool_maps.values()))
message = [HumanMessage(content=query)]
for i in range(5): # 最多5轮对话防止死循环
response = tool_llm.invoke(message)
message.append(response)
if not response.tool_calls:
return response.content
for tool_call in response.tool_calls:
# 执行工具调用
tool_output = tool_maps[tool_call["name"]].invoke(tool_call["args"])
message.append(
ToolMessage(
content=tool_output,
tool_call_id=tool_call["id"],
name=tool_call["name"]
)
)
这个流程中有几个关键机制:
- 工具绑定:通过
bind_tools()方法将工具集与LLM关联 - 多轮对话管理:维护message列表保存对话历史
- 工具调用处理:解析LLM返回的tool_calls,执行对应工具并记录结果
- 循环终止条件:当LLM不再返回工具调用或达到最大轮次时结束
4. 安全风险与最佳实践
4.1 eval函数的安全隐患
案例中的计算器工具使用了Python内置的eval函数,这在实际项目中是非常危险的:
python复制@tool
def calculator(expression: str) -> str:
try:
return str(eval(expression)) # 安全风险!
except Exception as e:
return f"计算错误: {e}"
恶意用户可能通过精心构造的输入执行任意代码,比如:
code复制"__import__('os').system('rm -rf /')"
4.2 安全加固方案
方案一:使用ast.literal_eval替代
python复制import ast
def safe_eval(expr: str) -> str:
try:
node = ast.parse(expr, mode='eval')
if not all(isinstance(n, (ast.Constant, ast.UnaryOp, ast.BinOp)) for n in ast.walk(node)):
raise ValueError("不安全的表达式")
return str(eval(compile(node, '<string>', 'eval')))
except Exception as e:
return f"计算错误: {e}"
方案二:白名单过滤
python复制import re
def safe_calculator(expr: str) -> str:
if not re.match(r'^[\d\s+\-*/().]+$', expr):
return "错误:表达式包含非法字符"
try:
return str(eval(expr))
except:
return "计算错误"
方案三:使用专用数学库
python复制import numexpr
def safe_calculator(expr: str) -> str:
try:
return str(numexpr.evaluate(expr))
except:
return "计算错误"
经验分享:在实际项目中,我推荐方案三。numexpr不仅安全,而且性能比原生eval更好,特别适合处理复杂数学表达式。
4.3 其他安全注意事项
- 工具调用权限控制:不同级别的Agent应该有不同的工具调用权限
- 输入输出过滤:对所有用户输入和工具输出进行安全检查
- 访问日志记录:记录所有工具调用详情,便于审计
- 资源访问限制:限制工具对系统资源的访问(如文件、网络等)
5. 性能优化与调试技巧
5.1 多轮对话的优化
案例中设置了最多5轮对话的限制,这是为了防止出现无限循环。但在实际项目中,我们可以更智能地控制对话流程:
python复制max_turns = 5
for i in range(max_turns):
response = tool_llm.invoke(message)
# 检查是否已经得出最终答案
if not response.tool_calls and response.content:
return response.content
# 检查是否陷入工具调用循环
if i > 0 and response.tool_calls == message[-1].tool_calls:
return "错误:检测到工具调用循环"
# 其余处理逻辑...
5.2 工具调用的超时控制
有些工具可能会长时间运行(如网络请求),我们需要添加超时机制:
python复制import signal
from contextlib import contextmanager
class TimeoutException(Exception): pass
@contextmanager
def time_limit(seconds):
def signal_handler(signum, frame):
raise TimeoutException("超时")
signal.signal(signal.SIGALRM, signal_handler)
signal.alarm(seconds)
try:
yield
finally:
signal.alarm(0)
@tool
def safe_rag_search(query: str) -> str:
try:
with time_limit(5): # 5秒超时
return rag_search(query)
except TimeoutException:
return "查询超时,请稍后再试"
5.3 调试与日志记录
开发Agent时,详细的日志非常重要。我通常会记录:
- 每轮对话的完整消息历史
- 所有工具调用的输入输出
- LLM的原始响应
- 执行耗时统计
python复制import logging
from datetime import datetime
logging.basicConfig(filename='agent.log', level=logging.INFO)
def log_interaction(role: str, content: str, metadata=None):
timestamp = datetime.now().isoformat()
log_entry = {
"timestamp": timestamp,
"role": role,
"content": content,
"metadata": metadata or {}
}
logging.info(json.dumps(log_entry))
6. 实际应用中的经验分享
6.1 工具描述的优化技巧
LLM如何决定使用哪个工具?完全依赖工具函数的文档字符串。经过多次实践,我发现好的工具描述应该:
- 第一句话明确说明工具的用途
- 参数部分给出2-3个典型示例
- 说明工具的适用场景和限制
- 避免使用专业术语,用自然语言描述
比如,这是优化后的计算器工具描述:
python复制@tool
def calculator(expression: str) -> str:
"""
执行基础数学运算,适合需要精确计算的场景。
可以处理加减乘除(+ - * /)和括号运算。
示例:
"2 + 3 * 4" → 14.0
"(5 - 2) / 2" → 1.5
注意:
- 不支持高级数学函数如sin/cos
- 最大支持15位数字
- 除以零会返回错误
"""
# 实现代码...
6.2 处理模糊查询的技巧
当用户查询比较模糊时(如"今天天气真好"),Agent需要有合理的响应策略。我通常会在工具调用前添加一层预处理:
python复制def preprocess_query(query: str) -> str:
# 检测是否是闲聊
if is_chitchat(query):
return "这是一个闲聊消息,不涉及工具调用"
# 检测查询是否足够明确
if not is_specific_enough(query):
return "请提供更具体的查询信息"
return query
def is_chitchat(text: str) -> bool:
chitchat_keywords = ["天气", "你好", "吃饭了吗"]
return any(keyword in text for keyword in chitchat_keywords)
6.3 内存管理的注意事项
长时间运行的Agent可能会积累大量对话历史,导致内存占用过高。我的解决方案是:
- 定期清理早期对话历史
- 对长期记忆进行摘要压缩
- 实现对话分页机制
python复制from langchain_core.messages import AIMessage, HumanMessage
def summarize_messages(messages, max_tokens=1000):
"""压缩过长的对话历史"""
total_len = sum(len(msg.content) for msg in messages)
if total_len <= max_tokens:
return messages
# 保留最近的对话,对早期对话生成摘要
summary = f"之前的对话摘要({len(messages)-5}条): ..."
return [AIMessage(content=summary)] + messages[-5:]
7. 扩展思路与进阶方向
7.1 多Agent协作系统
单个Agent能力有限,可以考虑构建多个Agent协作的系统:
- 主管Agent:负责任务分解和分配
- 专家Agent:专注于特定领域(如财务、法律)
- 协调Agent:处理Agent间的通信和冲突
7.2 动态工具加载
目前的工具是静态定义的,可以扩展为动态加载:
python复制def load_tools_from_directory(dir_path):
tools = {}
for filename in os.listdir(dir_path):
if filename.endswith('.py'):
module = import_module(f"tools.{filename[:-3]}")
for name in dir(module):
obj = getattr(module, name)
if hasattr(obj, '_is_tool'):
tools[name] = obj
return tools
7.3 可视化监控界面
对于生产环境,可以开发一个监控面板,实时显示:
- Agent的活动状态
- 工具调用统计
- 系统资源使用情况
- 异常报警
python复制from flask import Flask, jsonify
app = Flask(__name__)
@app.route('/status')
def agent_status():
return jsonify({
"active": True,
"last_query": last_query,
"tool_usage": tool_stats
})
开发LangChain Agent是一个不断迭代的过程,我在实际项目中最大的体会是:一定要从简单开始,逐步增加复杂度。先确保基础工具调用流程跑通,再考虑性能优化和安全加固。每次添加新功能后,都要进行充分的测试,特别是边界条件和异常情况。
