Python LangChain 组件工程化封装实战：带记忆的天气查询助手
·
目标
- 工程化分层封装四大核心模块：Model、Memory、Tool、Chain
- 自定义天气查询 Tool（模拟 / 真实接口两种方案）
- 接入对话记忆，实现上下文感知：记住历史对话、支持追问
- 完整可运行代码 + 分步搭建流程 + 测试用例
环境：Python 3.10+，LangChain 0.2
一、依赖安装
pip install langchain langchain-openai python-dotenv pydantic
# 如果使用通义千问/DeepSeek等模型，替换对应包 langchain-deepseek
二、整体工程分层设计（工程化思想）
plaintext
llm_assistant/
├── config.py # 配置文件：API Key、模型名称
├── llm_provider.py # 【Model层封装】统一LLM实例构建
├── memory_manager.py # 【Memory层封装】对话记忆管理
├── tools/
│ ├── __init__.py
│ └── weather_tool.py # 【Tool封装】自定义天气查询工具
├── agent_chain.py # 【Chain/Agent组装层】整合所有组件
└── main.py # 入口与测试脚本
分层优势：解耦，随时替换模型、更换记忆类型、新增工具，符合生产工程规范。
Python大模型RAG+AI智能体应用开发实战，视频完整课程地址：https://edu.csdn.net/course/detail/41293
https://edu.csdn.net/course/detail/41293
三、分步实现
步骤 1：配置文件 config.py
from dotenv import load_dotenv
import os
load_dotenv()
# LLM配置，可切换 OpenAI / DeepSeek / 通义千问
LLM_CONFIG = {
"api_key": os.getenv("LLM_API_KEY"),
"base_url": os.getenv("LLM_BASE_URL"),
"model_name": "deepseek-chat",
"temperature": 0.1
}
步骤 2：Model 封装 llm_provider.py
统一封装 LLM 实例，避免代码到处初始化模型，便于切换模型供应商
from langchain_openai import ChatOpenAI
from config import LLM_CONFIG
class LLMProvider:
@staticmethod
def get_chat_model():
"""获取统一封装的对话大模型"""
llm = ChatOpenAI(
api_key=LLM_CONFIG["api_key"],
base_url=LLM_CONFIG["base_url"],
model=LLM_CONFIG["model_name"],
temperature=LLM_CONFIG["temperature"]
)
return llm
步骤 3：Memory 封装 memory_manager.py
LangChain 常见记忆：ConversationBufferMemory（全量存储）、SummaryMemory、TokenBufferMemory 工程封装：提供统一创建记忆实例方法，支持多会话隔离
from langchain.memory import ConversationBufferMemory
class MemoryManager:
@staticmethod
def build_conversation_memory(session_id: str = "default") -> ConversationBufferMemory:
"""
构建对话记忆
return: 记忆实例，自动保存历史问答
"""
memory = ConversationBufferMemory(
memory_key="chat_history", # 传给prompt的变量名
return_messages=True # 返回消息对象列表（适配Agent）
)
return memory
重点：
return_messages=True是新版 Agent / 工具调用必备配置。
步骤 4：自定义天气查询 Tool tools/weather_tool.py
两种方案：
- 模拟天气（无需外网接口，本地测试首选）
- 拓展：接入和风天气 / 高德天气真实 API
from langchain.tools import tool
import json
class WeatherTools:
@staticmethod
@tool("weather_query")
def query_weather(city: str) -> str:
"""
查询指定城市当前天气信息
Args:
city: 需要查询天气的城市名称，例如：北京、上海、廊坊
"""
# ==========模拟数据，真实项目替换http请求调用天气API==========
mock_data = {
"廊坊": {"temp": "27℃", "weather": "多云", "wind": "南风2级"},
"北京": {"temp": "30℃", "weather": "晴", "wind": "东风3级"},
"上海": {"temp": "26℃", "weather": "小雨", "wind": "东南风4级"}
}
if city not in mock_data:
return f"暂无【{city}】的天气数据"
res = mock_data[city]
return json.dumps(res, ensure_ascii=False)
Tool 开发规范（工程要点）
- 使用
@tool装饰器自动生成工具描述、参数 schema - 必须书写清晰的函数文档字符串，LLM 依靠文档理解何时调用工具
- 参数类型显式声明，便于工具调用时 LLM 参数解析
步骤 5：组装 Chain + Agent agent_chain.py
场景说明：需要工具调用 + 记忆，优先使用 LangChain Agent（而非普通 Chain） 普通 Chain 无法自动选择调用 Tool；Agent 具备工具决策能力。
from langchain.agents import create_openai_tools_agent, AgentExecutor
from langchain.prompts import ChatPromptTemplate
from llm_provider import LLMProvider
from memory_manager import MemoryManager
from tools.weather_tool import WeatherTools
class WeatherAssistantAgent:
def __init__(self):
# 1. 加载模型
self.llm = LLMProvider.get_chat_model()
# 2. 加载记忆
self.memory = MemoryManager.build_conversation_memory()
# 3. 加载工具列表
self.tools = [WeatherTools.query_weather]
# 4. 构建Prompt模板
self.prompt = ChatPromptTemplate.from_messages([
("system", "你是专业天气助手，可以查询城市天气。不知道城市天气时调用工具，不要编造天气数据。"),
("placeholder", "{chat_history}"),
("user", "{input}"),
("placeholder", "{agent_scratchpad}"),
])
# 5. 创建Agent
self.agent = create_openai_tools_agent(self.llm, self.tools, self.prompt)
# 6. Agent执行器（整合记忆、工具、agent逻辑）
self.agent_executor = AgentExecutor(
agent=self.agent,
tools=self.tools,
memory=self.memory,
verbose=True, # 开启日志，调试工具调用流程
handle_parsing_errors=True
)
def chat(self, user_input: str):
"""对外暴露对话接口"""
response = self.agent_executor.invoke({"input": user_input})
return response["output"]
步骤 6：入口测试 main.py
from agent_chain import WeatherAssistantAgent
if __name__ == "__main__":
assistant = WeatherAssistantAgent()
print("=====天气助手启动=====")
# 测试用例
print("【用户】廊坊现在天气怎么样？")
reply1 = assistant.chat("廊坊现在天气怎么样？")
print(f"【助手】{reply1}\n")
print("【用户】刚才问的哪个城市？气温多少？")
reply2 = assistant.chat("刚才问的哪个城市？气温多少？")
print(f"【助手】{reply2}\n")
print("【用户】上海下雨吗？")
reply3 = assistant.chat("上海下雨吗？")
print(f"【助手】{reply3}\n")
print("【用户】对比廊坊和上海的温度")
reply4 = assistant.chat("对比廊坊和上海的温度")
print(f"【助手】{reply4}")
四、环境配置 .env 文件
LLM_API_KEY=sk-xxx
LLM_BASE_URL=https://api.deepseek.com
五、运行预期结果（记忆生效验证）
- 第一轮：查询廊坊天气 → Agent 自动调用
weather_query工具 - 第二轮追问：“刚才问的哪个城市？” ✅ 记忆生效：模型读取
chat_history，不需要再次调用工具，直接从历史回答 - 后续多轮对话持续保留上下文
六、完整测试方法（3 类测试）
1. 功能正向测试
表格
| 用户提问 | 预期行为 |
|---|---|
| 廊坊天气 | 自动调用 weather_query 工具，返回温度天气 |
| 刚才查询的城市气温？ | 使用对话记忆，不调用工具，读取历史上下文 |
| 廊坊和上海天气对比 | 连续调用两次天气工具，汇总结果 |
2. 边界异常测试
- 查询不存在城市：
查询成都天气→ 返回「暂无数据」 - 无关问题：
1+1等于几→ 不调用天气工具，直接回答 - 模糊提问：
我所在城市天气→ 模型会追问 “请告诉我城市名称”
3. 组件单元测试（工程化必备）
单独测试各个模块，便于定位 bug：
# 单元测试1：单独测试Tool
from tools.weather_tool import WeatherTools
print(WeatherTools.query_weather.invoke({"city":"廊坊"}))
# 单元测试2：测试Memory读写
from memory_manager import MemoryManager
mem = MemoryManager.build_conversation_memory()
mem.save_context({"input":"廊坊天气"}, {"output":"廊坊27℃多云"})
print(mem.load_memory_variables({}))
七、核心组件工程化拓展优化（生产可用）
1. Memory 优化
- 当前
ConversationBufferMemory无限存储长对话，长会话容易 token 溢出 - 生产替换：
TokenBufferMemory、SummaryConversationMemory - 多用户场景：持久化记忆（Redis / PostgreSQL），替换内存记忆，重启不丢失历史
2. Tool 层优化
- 统一工具基类，增加工具超时、异常捕获、日志
- 增加工具权限控制、限流
try:
# http请求天气API
except Exception as e:
return f"天气查询失败：{str(e)}"
3. Model 层优化
- 增加重试机制（tenacity）
- 支持模型动态切换、负载均衡
- 增加 token 消耗统计
4. Chain/Agent 选型说明
- 简单问答、无工具：直接使用
RunnablePassthrough构建 LCEL Chain - 需要调用外部工具、具备思考规划：必须使用 Agent（本案例场景）
八、常见踩坑点
- Memory 不生效：忘记在
AgentExecutor传入 memory；prompt 缺少{chat_history}占位符 - Agent 不肯调用 Tool：工具文档字符串描述不清晰，LLM 不知道何时使用
- 参数解析失败：tool 函数参数类型缺失、文档 Args 书写错误。
更多推荐
32
0- 0
已为社区贡献4条内容
所有评论(0)