1. 从零开始构建LangChain智能体：实战企业级AI助手
最近在开发企业知识管理系统时，我发现很多同事都在重复询问相同的问题："项目截止日期是什么时候？"、"预算还剩多少？"。于是我用LangChain框架开发了一个能自动回答这类问题的AI助手，今天就把这个实战案例分享给大家。
这个智能体不仅能查询公司内部文档，还能进行精确计算。比如当问到"预算提高46%后是多少"时，它能自动检索原始预算数据，然后调用计算器完成百分比运算。下面我会详细拆解实现过程，包括工具定义、多轮对话机制和安全防护措施。
2. 智能体核心架构解析
2.1 智能体的四大核心组件
现代AI智能体（Agent）不同于简单的聊天机器人，它更像一个数字员工，具备完整的认知和工作能力。我们的企业助手包含以下核心模块：
- 大脑（LLM）：使用通义千问（qwen-plus）作为推理引擎
- 记忆系统：
- 短期记忆：保存当前对话历史（message数组）
- 长期记忆：基于FAISS构建的RAG知识库
- 规划能力：通过多轮对话机制自主决定工具调用顺序
- 工具集：目前集成计算器和文档检索两个工具
python复制# 典型智能体工作流程示例
def agent_workflow(query):
# 1. LLM理解问题
# 2. 决定是否需要调用工具
# 3. 如有工具调用则执行
# 4. 整合结果生成最终回复
# 5. 更新对话历史
2.2 工具定义规范
在LangChain中，工具（Tool）是智能体与外界交互的接口。定义工具时需要特别注意三点：
- @tool装饰器：将普通函数转换为智能体可调用的工具
- 文档字符串：必须详细说明功能、参数和返回值
- 返回类型：必须返回字符串格式的结果
python复制@tool
def calculator(expression: str) -> str:
"""
计算数学表达式。需要精确计算时使用。
参数:
expression: 数学算式，如 "2 + 2" 或 "500 * 0.8"。
返回:
str: 计算结果，如 "4.0" 或 "400.0"。
"""
try:
return str(eval(expression)) # 注意：实际项目需要替换这个危险的eval
except Exception as e:
return f"计算错误: {e}"
重要提示：示例中使用eval仅用于演示，实际项目必须使用更安全的计算库如numexpr
3. 企业知识库实现细节
3.1 RAG知识库构建
我们的企业文档检索工具采用FAISS向量数据库，处理流程如下：
- 文档预处理：将原始文本分割为适合检索的片段
- 向量化：使用达摩院text-embedding-v1模型生成嵌入
- 索引构建：建立高效的相似度搜索索引
python复制def rag_search(query: str) -> str:
# 文档分割配置
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=25, # 每个片段约25个字符
chunk_overlap=5 # 片段间重叠5个字符
)
# 加载或创建向量库
if os.path.exists("faiss_index"):
db = FAISS.load_local("faiss_index", embeddings)
else:
docs = [Document(page_content=company_data)]
split_docs = text_splitter.split_documents(docs)
db = FAISS.from_documents(split_docs, embeddings)
db.save_local("faiss_index")
# 执行相似度搜索
return "\n\n".join(doc.page_content for doc in db.similarity_search(query, k=2))
3.2 检索优化技巧
在实际使用中发现几个关键点：
- 分块大小：25字符对于FAQ类数据很合适，但技术文档可能需要100-200字符
- 重叠设置：确保关键信息不会在分块时被切断
- 检索数量：k=2平衡了准确性和性能，复杂问题可增加到3-5
4. 多轮对话引擎实现
4.1 核心循环结构
智能体的"大脑"通过以下流程处理用户查询：
python复制def run_agent(query):
# 初始化对话历史
messages = [HumanMessage(content=query)]
# 最多进行5轮工具调用
for _ in range(5):
# 获取LLM响应
response = llm.invoke(messages)
messages.append(response)
# 检查是否需要调用工具
if not response.tool_calls:
return response.content
# 执行工具调用
for call in response.tool_calls:
tool_output = execute_tool(call)
messages.append(create_tool_message(tool_output, call))
return "超过最大迭代次数"
4.2 工具调用安全机制
在工具调用环节有几个关键安全措施：
- 工具存在性检查：防止LLM hallucination导致调用不存在的方法
- 参数类型验证：确保传入参数符合工具要求
- 执行环境隔离：危险操作应在沙箱中运行
python复制# 工具调用安全包装示例
def safe_tool_invoke(tool_name, args):
if tool_name not in registered_tools:
return f"错误：工具{tool_name}未注册"
try:
return registered_tools[tool_name](**args)
except Exception as e:
return f"工具执行错误: {str(e)}"
5. 安全风险与防护方案
5.1 常见安全威胁
在开发过程中发现几个高危风险点：
- 代码注入：通过计算器工具的eval函数可能执行任意代码
- 敏感信息泄露：RAG可能返回不应公开的文档片段
- 无限循环：LLM可能持续要求调用工具
5.2 加固措施实践
针对上述风险，我们实施了以下解决方案：
计算器安全改造：
python复制import numexpr # 替换eval的安全计算库
def safe_calculator(expr: str) -> str:
# 白名单校验
if not re.match(r'^[\d\.\+\-\*/%\(\) ]+$', expr):
return "错误：包含非法字符"
try:
return str(numexpr.evaluate(expr))
except:
return "计算错误"
RAG访问控制：
- 添加用户身份验证层
- 实现文档级权限过滤
- 记录所有检索历史
循环终止条件：
- 设置最大迭代次数（如5次）
- 监控耗时过长的会话
- 实现"取消"指令处理
6. 性能优化与调试技巧
6.1 响应速度提升
通过以下优化将平均响应时间从3.2秒降至1.5秒：
- 向量库预热：服务启动时预加载FAISS索引
- 工具并行化：多个独立工具可并发执行
- 缓存策略：对常见查询结果缓存5分钟
6.2 调试日志规范
完善的日志能快速定位问题，我们的日志格式如下：
code复制[2024-03-20 14:00:00] QUERY: "预算提高46%后是多少"
[TOOL CALL] calculator(500*1.46)
[TOOL RESULT] 730.0
[RESPONSE] "提高46%后预算为730元"
关键日志字段包括：
- 时间戳
- 用户原始查询
- 工具调用详情
- 最终响应内容
7. 扩展应用场景
这个基础框架可以轻松扩展更多企业应用：
-
会议纪要助手：
- 添加日历查询工具
- 集成语音转录功能
- 自动生成待办事项
-
技术支持机器人：
- 连接工单系统
- 知识库增加技术文档
- 支持截图分析
-
数据分析助手：
- 集成SQL查询工具
- 添加可视化生成功能
- 支持自然语言转查询
实现这些扩展只需要定义新工具并更新提示词，核心引擎无需修改。
8. 踩坑记录与经验分享
在实际部署中遇到过几个典型问题：
中文分词异常：
- 现象：RAG对中文文档检索准确率低
- 原因：默认分片按字符切割破坏语义
- 解决：改用中文分词器（如jieba）预处理
工具选择偏差：
- 现象：LLM过度依赖某个工具
- 原因：工具描述不平衡导致
- 解决：统一各工具描述的详细程度
长会话记忆丢失：
- 现象：对话超过10轮后逻辑混乱
- 原因：未做对话历史摘要
- 解决：定期用LLM生成会话摘要
这些经验让我深刻体会到，开发生产级AI应用不仅需要算法知识，更需要系统工程思维和对业务场景的深入理解。