3步搞定DeepSeek多轮对话：Ollama+LangChain+Flask实战（附完整代码）

最近在折腾本地AI应用，发现很多开发者都想搭建一个能记住对话历史的聊天系统，但面对Ollama、LangChain、Flask这一堆技术栈，常常觉得无从下手。其实，只要抓住核心逻辑，用最少的代码就能实现一个可用的多轮对话系统。今天我就分享一个极简实现方案，适合那些想快速上手、不想被复杂架构困扰的实践派程序员。

这个方案的核心思路很简单：用Ollama在本地运行DeepSeek模型，用LangChain管理对话记忆，用Flask提供Web接口。整个过程只需要三个关键步骤，每步都有明确的代码示例，你可以直接复制粘贴到自己的项目中。我测试过，在普通的开发机器上（甚至没有独立显卡）也能流畅运行。

1. 环境准备与模型部署

在开始写代码之前，我们需要先把基础环境搭建好。这一步看似简单，但很多问题都出在这里，所以我详细说明一下每个环节的注意事项。

1.1 安装Ollama并拉取模型

Ollama是目前最方便的本地大模型运行工具，它把复杂的模型部署过程简化成了几条命令。首先，根据你的操作系统安装Ollama：

# Linux/macOS
curl -fsSL https://ollama.com/install.sh | sh

# Windows (PowerShell管理员模式)
irm https://ollama.com/install.ps1 | iex

安装完成后，启动Ollama服务：

# 启动服务（默认端口11434）
ollama serve

提示：如果你看到“Listening on 0.0.0.0:11434”的输出，说明服务启动成功。可以另开一个终端窗口继续下面的操作。

接下来拉取DeepSeek模型。考虑到不同设备的性能差异，我建议从较小的模型开始：

# 拉取7B参数版本（约4.5GB）
ollama pull deepseek-r1:7b

# 如果想更轻量，可以试试3B版本
# ollama pull deepseek-r1:3b

模型拉取需要一些时间，取决于你的网络速度。完成后，可以用简单命令测试模型是否正常工作：

ollama run deepseek-r1:7b "你好，请介绍一下自己"

如果模型能正常回复，说明Ollama环境已经就绪。这里有个小技巧：如果你发现模型响应很慢，可以检查一下是否启用了GPU加速。Ollama默认会尝试使用GPU，但有些环境下可能需要手动配置。

1.2 Python依赖安装

我们的代码主要用Python编写，需要安装几个关键库。建议先创建一个虚拟环境：

# 创建虚拟环境
python -m venv deepseek_env

# 激活环境
# Linux/macOS:
source deepseek_env/bin/activate
# Windows:
deepseek_env\Scripts\activate

然后安装必要的包：

pip install langchain langchain-community flask ollama

这里解释一下每个包的作用：

• langchain：核心框架，提供对话链、记忆管理等组件
• langchain-community：包含社区维护的各种集成，包括Ollama
• flask：轻量级Web框架，用于创建API接口
• ollama：Python客户端，方便调用本地Ollama服务

注意：LangChain的版本更新比较快，如果遇到API变化，可以尝试指定版本安装：pip install langchain==0.1.0 langchain-community==0.0.10

1.3 验证环境连通性

在写正式代码前，先写个简单的测试脚本确认所有组件都能正常工作：

# test_env.py
import ollama
from langchain_community.llms import OllamaLLM

# 测试Ollama直接调用
response = ollama.chat(model='deepseek-r1:7b', messages=[
    {'role': 'user', 'content': '简单说下Python的特点'}
])
print("Ollama直接调用测试：")
print(response['message']['content'][:100] + "...")

# 测试LangChain集成
llm = OllamaLLM(model="deepseek-r1:7b")
test_response = llm.invoke("1+1等于几？")
print("\nLangChain集成测试：")
print(test_response)

运行这个脚本，如果两处都能正常输出，说明环境配置成功。我遇到过一些常见问题，这里列出来供你参考：

问题现象	可能原因	解决方案
连接被拒绝	Ollama服务未启动	运行 ollama serve
模型找不到	模型名称拼写错误	用 ollama list 查看可用模型
内存不足	模型太大或系统内存不足	换用更小的模型，如 deepseek-r1:3b
响应超时	第一次加载模型较慢	等待几分钟再试，或检查网络

2. 核心代码实现

环境准备好后，我们进入最关键的代码实现部分。我会把代码拆解成几个独立模块，方便你理解和修改。

2.1 对话记忆管理

多轮对话的核心是“记忆”——模型需要记住之前的对话内容。LangChain提供了多种记忆管理方案，我们选择最实用的ConversationBufferMemory。

# memory_manager.py
from langchain.memory import ConversationBufferMemory
from langchain.schema import HumanMessage, AIMessage
import json
import os

class SimpleMemoryManager:
    """简化的对话记忆管理器"""
    
    def __init__(self, max_history=5):
        """
        初始化记忆管理器
        :param max_history: 最大对话轮数，避免上下文过长
        """
        self.memory_store = {}
        self.max_history = max_history
        
    def get_memory(self, session_id):
        """获取指定会话的记忆"""
        if session_id not in self.memory_store:
            # 创建新的记忆对象
            self.memory_store[session_id] = ConversationBufferMemory(
                memory_key="chat_history",
                return_messages=True,
                max_history_length=self.max_history
            )
        return self.memory_store[session_id]
    
    def save_context(self, session_id, user_input, ai_response):
        """保存一轮对话"""
        memory = self.get_memory(session_id)
        memory.save_context(
            {"input": user_input},
            {"output": ai_response}
        )
        
    def get_history_text(self, session_id):
        """获取对话历史文本（用于调试）"""
        memory = self.get_memory(session_id)
        history = memory.load_memory_variables({})["chat_history"]
        
        history_text = []
        for msg in history:
            if isinstance(msg, HumanMessage):
                history_text.append(f"用户: {msg.content}")
            elif isinstance(msg, AIMessage):
                history_text.append(f"AI: {msg.content}")
        
        return "\n".join(history_text)
    
    def clear_memory(self, session_id):
        """清空指定会话的记忆"""
        if session_id in self.memory_store:
            del self.memory_store[session_id]

这个记忆管理器有几个设计考虑：

1. 按会话隔离：不同用户（通过session_id区分）有独立的对话历史
2. 长度限制：通过max_history防止上下文过长导致性能下降
3. 简单持久化：虽然这里用了内存存储，但很容易扩展为数据库存储

在实际项目中，你可能需要更复杂的记忆管理。比如，当对话轮数超过限制时，可以自动总结之前的对话内容，而不是直接丢弃。不过对于入门项目，这个简单版本已经足够。

2.2 LangChain对话链构建

有了记忆管理，接下来构建对话链。这是LangChain的核心概念——把多个处理步骤连接成一个“链”。

# chat_chain.py
from langchain_community.llms import OllamaLLM
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain
from memory_manager import SimpleMemoryManager

class DeepSeekChatChain:
    """DeepSeek对话链"""
    
    def __init__(self, model_name="deepseek-r1:7b"):
        self.model_name = model_name
        self.memory_manager = SimpleMemoryManager()
        
        # 初始化Ollama模型
        self.llm = OllamaLLM(
            model=model_name,
            temperature=0.7,  # 控制创造性，0-1之间
            num_predict=512,   # 最大生成token数
        )
        
        # 构建提示词模板
        self.prompt_template = PromptTemplate(
            input_variables=["chat_history", "input"],
            template="""你是一个专业的AI助手，请根据对话历史回答用户的问题。

对话历史：
{chat_history}

当前问题：{input}

请给出详细且专业的回答："""
        )
        
        # 创建对话链（先不绑定记忆，动态绑定）
        self.base_chain = LLMChain(
            llm=self.llm,
            prompt=self.prompt_template,
            verbose=False  # 设为True可以看到详细执行过程
        )
    
    def chat(self, session_id, user_input):
        """处理用户输入，返回AI回复"""
        # 获取该会话的记忆
        memory = self.memory_manager.get_memory(session_id)
        
        # 加载历史对话
        history_vars = memory.load_memory_variables({})
        chat_history = history_vars["chat_history"]
        
        # 构建历史文本
        history_text = ""
        if chat_history:
            # 将消息对象转换为文本
            for msg in chat_history:
                if hasattr(msg, 'type'):
                    role = "用户" if msg.type == "human" else "AI"
                    history_text += f"{role}: {msg.content}\n"
                else:
                    # 兼容不同版本
                    history_text += f"{msg}\n"
        
        # 调用模型
        response = self.base_chain.run(
            chat_history=history_text,
            input=user_input
        )
        
        # 保存对话到记忆
        self.memory_manager.save_context(session_id, user_input, response)
        
        return response
    
    def stream_chat(self, session_id, user_input):
        """流式对话（逐字输出）"""
        # 获取历史
        memory = self.memory_manager.get_memory(session_id)
        history_vars = memory.load_memory_variables({})
        chat_history = history_vars["chat_history"]
        
        # 构建历史文本
        history_text = ""
        if chat_history:
            for msg in chat_history:
                if hasattr(msg, 'type'):
                    role = "用户" if msg.type == "human" else "AI"
                    history_text += f"{role}: {msg.content}\n"
        
        # 使用流式接口
        full_response = ""
        for chunk in self.llm.stream(
            self.prompt_template.format(
                chat_history=history_text,
                input=user_input
            )
        ):
            yield chunk
            full_response += chunk
        
        # 保存完整回复到记忆
        self.memory_manager.save_context(session_id, user_input, full_response)

这里有几个关键点需要注意：

1. 提示词设计：好的提示词能显著提升模型表现。我在这里明确告诉模型要参考对话历史，并给出具体的回答要求。

2. 流式输出：stream_chat方法实现了逐字输出，用户体验更好。这在Web界面中特别有用。

3. 参数调优：

   • temperature=0.7：平衡创造性和一致性，值越高回答越多样
   • num_predict=512：限制单次回复长度，避免生成过长的内容

如果你发现模型回答不够准确，可以调整提示词。比如，增加更具体的指令：

template="""你是一个专业的AI助手，请严格遵守以下要求：
1. 仔细阅读对话历史，确保回答与之前内容一致
2. 如果用户的问题基于之前的对话，请明确引用
3. 回答要简洁明了，避免冗长
4. 如果不知道答案，直接说"我不确定"，不要编造

对话历史：
{chat_history}

当前问题：{input}

请开始回答："""

2.3 Flask Web服务封装

最后一步，用Flask把对话功能包装成Web API。这样前端应用或其他系统就可以通过HTTP请求来使用我们的聊天系统了。

# app.py
from flask import Flask, request, jsonify, Response
from chat_chain import DeepSeekChatChain
import json

app = Flask(__name__)
chat_chain = DeepSeekChatChain()

@app.route('/api/chat', methods=['POST'])
def chat_endpoint():
    """聊天接口"""
    try:
        data = request.json
        if not data or 'message' not in data:
            return jsonify({
                'error': '缺少message参数'
            }), 400
        
        session_id = data.get('session_id', 'default')
        user_message = data['message']
        stream = data.get('stream', False)
        
        if stream:
            # 流式响应
            def generate():
                for chunk in chat_chain.stream_chat(session_id, user_message):
                    yield f"data: {json.dumps({'chunk': chunk})}\n\n"
            
            return Response(
                generate(),
                mimetype='text/event-stream',
                headers={
                    'Cache-Control': 'no-cache',
                    'X-Accel-Buffering': 'no'
                }
            )
        else:
            # 普通响应
            response = chat_chain.chat(session_id, user_message)
            return jsonify({
                'response': response,
                'session_id': session_id
            })
            
    except Exception as e:
        app.logger.error(f"聊天接口错误: {str(e)}")
        return jsonify({
            'error': '内部服务器错误',
            'detail': str(e)
        }), 500

@app.route('/api/history/<session_id>', methods=['GET'])
def get_history(session_id):
    """获取对话历史"""
    try:
        history_text = chat_chain.memory_manager.get_history_text(session_id)
        return jsonify({
            'session_id': session_id,
            'history': history_text
        })
    except Exception as e:
        return jsonify({'error': str(e)}), 500

@app.route('/api/clear/<session_id>', methods=['POST'])
def clear_history(session_id):
    """清空对话历史"""
    try:
        chat_chain.memory_manager.clear_memory(session_id)
        return jsonify({
            'success': True,
            'message': f'已清空会话 {session_id} 的历史记录'
        })
    except Exception as e:
        return jsonify({'error': str(e)}), 500

@app.route('/health', methods=['GET'])
def health_check():
    """健康检查接口"""
    return jsonify({
        'status': 'healthy',
        'model': chat_chain.model_name
    })

if __name__ == '__main__':
    # 启动Flask服务
    app.run(
        host='0.0.0.0',
        port=5000,
        debug=True  # 生产环境设为False
    )

这个Flask应用提供了几个关键接口：

1. POST /api/chat：主聊天接口，支持流式和非流式两种模式
2. GET /api/history/<session_id>：查看指定会话的历史记录
3. POST /api/clear/<session_id>：清空指定会话的历史
4. GET /health：健康检查，用于监控服务状态

注意：在生产环境中，你需要添加更多的错误处理、请求验证和安全性措施。比如，可以添加API密钥验证、请求频率限制等。

3. 完整项目集成与测试

现在我们把所有模块整合起来，创建一个完整的可运行项目。

3.1 项目结构

建议按以下结构组织你的项目：

deepseek-chat/
├── app.py              # Flask主程序
├── chat_chain.py       # 对话链实现
├── memory_manager.py   # 记忆管理
├── requirements.txt    # 依赖列表
├── test_chat.py        # 测试脚本
└── README.md          # 项目说明

requirements.txt内容：

flask>=3.0.0
langchain>=0.1.0
langchain-community>=0.0.10
ollama>=0.1.14

3.2 一键启动脚本

为了方便启动，可以创建一个启动脚本：

#!/bin/bash
# run.sh

# 检查Ollama服务是否运行
if ! curl -s http://localhost:11434/api/tags > /dev/null; then
    echo "启动Ollama服务..."
    ollama serve &
    OLLAMA_PID=$!
    sleep 10  # 等待服务启动
fi

# 检查模型是否已下载
if ! ollama list | grep -q "deepseek-r1:7b"; then
    echo "下载DeepSeek模型..."
    ollama pull deepseek-r1:7b
fi

# 启动Flask应用
echo "启动Flask应用..."
python app.py

给脚本添加执行权限：chmod +x run.sh，然后直接运行./run.sh即可启动整个系统。

3.3 测试对话系统

启动服务后，我们可以用多种方式测试系统是否正常工作。

方法1：使用curl测试API

# 测试普通聊天
curl -X POST http://localhost:5000/api/chat \
  -H "Content-Type: application/json" \
  -d '{
    "session_id": "test_user_001",
    "message": "Python有什么特点？"
  }'

# 测试流式聊天
curl -X POST http://localhost:5000/api/chat \
  -H "Content-Type: application/json" \
  -d '{
    "session_id": "test_user_001",
    "message": "继续说说Python在数据分析中的应用",
    "stream": true
  }'

# 查看对话历史
curl http://localhost:5000/api/history/test_user_001

方法2：使用Python测试脚本

# test_chat.py
import requests
import json

BASE_URL = "http://localhost:5000"

def test_chat():
    """测试多轮对话"""
    session_id = "test_session"
    
    # 第一轮
    print("第一轮对话：")
    response = requests.post(f"{BASE_URL}/api/chat", json={
        "session_id": session_id,
        "message": "介绍一下机器学习"
    })
    print(f"AI: {response.json()['response'][:100]}...")
    
    # 第二轮（应该能记住上下文）
    print("\n第二轮对话：")
    response = requests.post(f"{BASE_URL}/api/chat", json={
        "session_id": session_id,
        "message": "刚才说的监督学习具体指什么？"
    })
    print(f"AI: {response.json()['response'][:100]}...")
    
    # 查看历史
    print("\n对话历史：")
    history = requests.get(f"{BASE_URL}/api/history/{session_id}")
    print(history.json()['history'])

def test_stream():
    """测试流式输出"""
    print("测试流式输出：")
    response = requests.post(
        f"{BASE_URL}/api/chat",
        json={
            "session_id": "stream_test",
            "message": "用Python写一个快速排序算法",
            "stream": True
        },
        stream=True
    )
    
    print("收到流式响应：")
    for line in response.iter_lines():
        if line:
            data = json.loads(line.decode('utf-8').replace('data: ', ''))
            print(data['chunk'], end='', flush=True)
    print()

if __name__ == "__main__":
    # 先检查服务是否健康
    health = requests.get(f"{BASE_URL}/health")
    print(f"服务状态: {health.json()}")
    
    test_chat()
    test_stream()

方法3：使用前端界面

如果你想要一个简单的Web界面，可以创建一个HTML文件：

<!-- static/index.html -->
<!DOCTYPE html>
<html>
<head>
    <title>DeepSeek聊天</title>
    <style>
        body { font-family: Arial; max-width: 800px; margin: 0 auto; padding: 20px; }
        #chat-box { border: 1px solid #ccc; height: 400px; overflow-y: auto; padding: 10px; margin-bottom: 10px; }
        .user-msg { text-align: right; color: blue; margin: 5px 0; }
        .ai-msg { text-align: left; color: green; margin: 5px 0; }
        #input-box { width: 100%; padding: 10px; margin-top: 10px; }
        button { padding: 10px 20px; margin-top: 10px; }
    </style>
</head>
<body>
    <h1>DeepSeek多轮对话测试</h1>
    <div id="chat-box"></div>
    <input type="text" id="input-box" placeholder="输入消息...">
    <button onclick="sendMessage()">发送</button>
    <button onclick="clearHistory()">清空历史</button>
    
    <script>
        let sessionId = 'web_user_' + Math.random().toString(36).substr(2, 9);
        
        function addMessage(role, content) {
            const chatBox = document.getElementById('chat-box');
            const msgDiv = document.createElement('div');
            msgDiv.className = role + '-msg';
            msgDiv.textContent = (role === 'user' ? '你: ' : 'AI: ') + content;
            chatBox.appendChild(msgDiv);
            chatBox.scrollTop = chatBox.scrollHeight;
        }
        
        async function sendMessage() {
            const input = document.getElementById('input-box');
            const message = input.value.trim();
            if (!message) return;
            
            addMessage('user', message);
            input.value = '';
            
            // 发送请求
            const response = await fetch('/api/chat', {
                method: 'POST',
                headers: { 'Content-Type': 'application/json' },
                body: JSON.stringify({
                    session_id: sessionId,
                    message: message,
                    stream: false
                })
            });
            
            const data = await response.json();
            if (data.response) {
                addMessage('ai', data.response);
            } else if (data.error) {
                addMessage('ai', '错误: ' + data.error);
            }
        }
        
        async function clearHistory() {
            await fetch(`/api/clear/${sessionId}`, { method: 'POST' });
            document.getElementById('chat-box').innerHTML = '';
            addMessage('ai', '历史记录已清空');
        }
        
        // 回车发送
        document.getElementById('input-box').addEventListener('keypress', function(e) {
            if (e.key === 'Enter') sendMessage();
        });
    </script>
</body>
</html>

然后在Flask应用中添加静态文件服务：

# 在app.py中添加
from flask import send_from_directory

@app.route('/')
def index():
    return send_from_directory('static', 'index.html')

@app.route('/static/<path:path>')
def serve_static(path):
    return send_from_directory('static', path)

3.4 性能优化建议

当你的聊天系统基本功能正常后，可以考虑以下优化：

1. 模型性能优化

# 调整模型参数
llm = OllamaLLM(
    model="deepseek-r1:7b",
    temperature=0.7,
    num_predict=256,  # 减少生成长度
    num_ctx=2048,     # 上下文长度
    top_k=40,         # 采样参数
    top_p=0.9,
    repeat_penalty=1.1  # 减少重复
)

2. 记忆优化策略

当对话历史过长时，可以自动总结而不是直接截断：

def summarize_history(self, session_id):
    """总结过长的对话历史"""
    memory = self.get_memory(session_id)
    history = memory.load_memory_variables({})["chat_history"]
    
    if len(history) > self.max_history * 2:
        # 提取关键对话进行总结
        summary_prompt = f"""请总结以下对话的核心内容：
        
        {history}
        
        总结要点："""
        
        summary = self.llm.invoke(summary_prompt)
        
        # 清空历史，添加总结
        memory.clear()
        memory.save_context(
            {"input": "系统提示"},
            {"output": f"之前的对话已总结：{summary}"}
        )

3. 缓存常用回答

对于常见问题，可以添加缓存减少模型调用：

import hashlib
from functools import lru_cache

class CachedChatChain(DeepSeekChatChain):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.response_cache = {}
    
    def get_cache_key(self, session_id, user_input):
        """生成缓存键"""
        content = f"{session_id}:{user_input}"
        return hashlib.md5(content.encode()).hexdigest()
    
    def chat(self, session_id, user_input):
        # 检查缓存
        cache_key = self.get_cache_key(session_id, user_input)
        if cache_key in self.response_cache:
            return self.response_cache[cache_key]
        
        # 调用父类方法
        response = super().chat(session_id, user_input)
        
        # 缓存结果（只缓存简短回答）
        if len(response) < 100:
            self.response_cache[cache_key] = response
        
        return response

4. 生产环境部署

如果你想把项目部署到生产环境，需要考虑更多因素。这里提供几个关键步骤：

4.1 使用Gunicorn替代开发服务器

Flask自带的开发服务器不适合生产环境。改用Gunicorn：

pip install gunicorn
gunicorn -w 4 -b 0.0.0.0:5000 app:app

4.2 添加Nginx反向代理

使用Nginx处理静态文件和负载均衡：

# /etc/nginx/sites-available/deepseek-chat
server {
    listen 80;
    server_name your-domain.com;
    
    location / {
        proxy_pass http://127.0.0.1:5000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
    }
    
    location /static {
        alias /path/to/your/static/files;
    }
}

4.3 使用Supervisor管理进程

确保服务在崩溃后自动重启：

# /etc/supervisor/conf.d/deepseek-chat.conf
[program:deepseek-chat]
command=/path/to/venv/bin/gunicorn -w 4 -b 127.0.0.1:5000 app:app
directory=/path/to/deepseek-chat
user=www-data
autostart=true
autorestart=true
stderr_logfile=/var/log/deepseek-chat.err.log
stdout_logfile=/var/log/deepseek-chat.out.log

4.4 监控和日志

添加详细的日志记录：

import logging
from logging.handlers import RotatingFileHandler

# 配置日志
handler = RotatingFileHandler(
    'app.log', maxBytes=10000, backupCount=3
)
handler.setLevel(logging.INFO)
app.logger.addHandler(handler)

# 记录每次请求
@app.before_request
def log_request():
    app.logger.info(f"请求: {request.method} {request.path}")

@app.after_request
def log_response(response):
    app.logger.info(f"响应: {response.status}")
    return response

4.5 安全性增强

生产环境必须考虑安全性：

from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

limiter = Limiter(
    app=app,
    key_func=get_remote_address,
    default_limits=["100 per day", "10 per hour"]
)

@app.route('/api/chat', methods=['POST'])
@limiter.limit("5 per minute")  # 限制聊天频率
def chat_endpoint():
    # ... 原有代码

4.6 Docker容器化

最后，你可以将整个应用Docker化，方便部署：

# Dockerfile
FROM python:3.9-slim

WORKDIR /app

# 安装系统依赖
RUN apt-get update && apt-get install -y \
    curl \
    && rm -rf /var/lib/apt/lists/*

# 安装Ollama
RUN curl -fsSL https://ollama.com/install.sh | sh

# 复制Python依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用代码
COPY . .

# 下载模型（可以在构建时或运行时）
RUN ollama pull deepseek-r1:7b

# 暴露端口
EXPOSE 5000

# 启动脚本
COPY entrypoint.sh .
RUN chmod +x entrypoint.sh

ENTRYPOINT ["./entrypoint.sh"]

#!/bin/bash
# entrypoint.sh

# 启动Ollama
ollama serve &

# 等待Ollama启动
sleep 10

# 启动Flask应用
gunicorn -w 4 -b 0.0.0.0:5000 app:app

这个三步实现的DeepSeek多轮对话系统虽然简单，但包含了核心功能。我在实际项目中用类似方案处理过客服对话、技术问答等场景，效果不错。最大的优点是可控性强——所有代码都在自己手里，可以根据需要随时调整。如果你在实现过程中遇到问题，或者有更好的优化建议，欢迎交流讨论。