从零搭建多模态聊天机器人

目标：构建一个支持语音、图片和文字输入的聊天机器人，具备简易Web前端和上下文记忆功能。

一、环境准备

1. 模型选择与部署

我们选择Qwen-2.5-Omni-3B模型，这是一个支持多模态输入的开源模型。可以选择：

  • 私有化部署：只需在一张4090显卡（24G显存以上）上部署
  • API调用（附录提供方法）

注意：如果您已熟悉模型部署，可跳过下面一段

2. 服务器连接与配置

方法一：使用PyCharm SSH连接

  1. 打开PyCharm：设置 → Python解释器 → 添加解释器 → 基于SSH
  2. 输入服务器IP、用户名和密钥文件

方法二：直接使用服务器终端

3. 安装依赖包

在服务器终端执行以下命令：

1
2
3
4
5
6
7
8
9
10
11
# 核心依赖
pip install langchain langchain_openai langchain_community

# 模型部署与推理
pip install vllm vllm[audio]

# 模型下载
pip install modelscope

# 前端与工具
pip install gradio pillow pydub sqlalchemy

4. 下载模型

创建 download_model.py 文件：

1
2
3
4
from modelscope import snapshot_download

model_dir = snapshot_download('Qwen/Qwen2.5-Omni-3B')
print(f"模型下载完成，保存至: {model_dir}")

运行后等待约10分钟完成下载。

5. 启动模型服务

1
2
3
4
5
6
7
8
9
python -m vllm.entrypoints.openai.api_server \
  --model /path/to/Qwen2.5-Omni-3B \  # 替换为您的实际路径
  --served-model-name qwen2.5-omni-3b \
  --max-model-len 8192 \
  --host 0.0.0.0 \
  --port 6006 \
  --dtype auto \
  --gpu-memory-utilization 0.9 \
  --enable-auto-tool-choice

验证：访问 http://服务器IP:6006/docs 查看API文档。

二、项目构建

1. 项目结构

1
2
3
4
5
6
7
multi-modal-chatbot/
├── .env                    # 环境变量
├── my_llm.py              # 模型配置
├── utils.py               # 工具函数
├── main.py               # 主程序
├── requirements.txt       # 依赖列表
└── chat_history.db       # 数据库（自动生成）

2. 环境配置

为代码的健全性，我们不在main函数中调用大模型，而是放在my_llm文件里方便修改

.env 文件：

1
2
LOCAL_BASE_URL="http://your-server-ip:6006/v1"

my_llm.py - 模型配置：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

# 加载环境变量
load_dotenv()

# 配置多模态模型
multiModal_llm = ChatOpenAI(
    model='qwen2.5-omni-3b',
    api_key=os.getenv('OPENAI_API_KEY', 'xx'),  # 本地部署可为任意值
    base_url=os.getenv('LOCAL_BASE_URL'),
    temperature=0.7,
    max_tokens=2048,
)

3. 工具函数 (utils.py)

这里主要负责将音频图片文件转化为base64格式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
import base64
import io
import os
import tempfile
from PIL import Image
from pydub import AudioSegment

def image_to_base64(image_path: str) -> dict:
    """将图片转换为Base64格式"""
    try:
        with Image.open(image_path) as img:
            # 统一格式处理
            if img.mode in ('RGBA', 'LA'):
                background = Image.new('RGB', img.size, (255, 255, 255))
                background.paste(img, mask=img.split()[-1])
                img = background
            
            buf = io.BytesIO()
            img.save(buf, format='JPEG', quality=85)
            b64 = base64.b64encode(buf.getvalue()).decode('utf-8')
        
        return {
            "type": "image_url",
            "image_url": {
                "url": f"data:image/jpeg;base64,{b64}",
                "detail": "low"  # 可选: low, high, auto
            }
        }
    except Exception as e:
        print(f"[ERROR] 图片处理失败: {e}")
        return None

def audio_to_base64(audio_path: str) -> dict:
    """将音频转换为Base64格式"""
    try:
        # 统一转换为WAV格式
        audio = AudioSegment.from_file(audio_path)
        audio = audio.set_frame_rate(16000).set_channels(1)
        
        with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as tmp:
            audio.export(tmp.name, format='wav')
            tmp_path = tmp.name
        
        # 读取并编码
        with open(tmp_path, 'rb') as f:
            b64 = base64.b64encode(f.read()).decode('utf-8')
        
        # 清理临时文件
        os.unlink(tmp_path)
        
        return {
            "type": "audio_url",
            "audio_url": {
                "url": f"data:audio/wav;base64,{b64}",
                "duration": len(audio) / 1000  # 秒
            }
        }
    except Exception as e:
        print(f"[ERROR] 音频处理失败: {e}")
        return None

4. 主程序 (main.py)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
import uuid
import gradio as gr
from sqlalchemy import create_engine
from langchain_core.messages import HumanMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableWithMessageHistory
from langchain_community.chat_message_histories import SQLChatMessageHistory

from my_llm import multiModal_llm
from utils import image_to_base64, audio_to_base64

# ========== 1. 对话链配置 ==========
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个多模态AI助手，可以处理文本、图片和语音输入。请用中文友好地回答用户的问题。"),
    MessagesPlaceholder(variable_name="messages"),
    ("system", "当前对话历史:"),
    MessagesPlaceholder(variable_name="history"),
])

chain = prompt | multiModal_llm

# ========== 2. 记忆管理 ==========
def get_session_history(session_id: str):
    """获取或创建会话历史"""
    engine = create_engine('sqlite:///chat_history.db')
    return SQLChatMessageHistory(
        session_id=session_id,
        connection=engine,
        table_name="chat_history"
    )

# 包装对话链
chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="messages",
    history_messages_key="history"
)

# 生成唯一会话ID
session_id = str(uuid.uuid4())

# ========== 3. 核心处理函数 ==========
def process_inputs(text: str, audio: str, image: str, chat_history: list):
    """处理用户输入并生成响应"""
    
    # 构建多模态内容
    content = []
    user_display = []
    
    # 处理图片
    if image:
        img_data = image_to_base64(image)
        if img_data:
            content.append(img_data)
            user_display.append("[图片]")
    
    # 处理音频
    if audio:
        audio_data = audio_to_base64(audio)
        if audio_data:
            content.append(audio_data)
            user_display.append("[语音]")
    
    # 处理文本
    if text and text.strip():
        content.append({"type": "text", "text": text})
        user_display.append(text)
    
    # 无有效输入
    if not content:
        chat_history.append(("", "请输入文本、上传图片或录制语音！"))
        return "", None, None, chat_history
    
    # 构建用户消息
    user_message = HumanMessage(content=content)
    display_text = " ".join(user_display) if user_display else "[多媒体内容]"
    
    try:
        # 调用模型
        response = chain_with_history.invoke(
            {"messages": [user_message]},
            config={"configurable": {"session_id": session_id}}
        )
        
        # 更新聊天历史
        chat_history.append((display_text, response.content))
        
    except Exception as e:
        error_msg = f"请求失败: {str(e)}"
        print(f"[ERROR] {error_msg}")
        chat_history.append((display_text, error_msg))
    
    # 清空输入
    return "", None, None, chat_history

# ========== 4. 前端界面 ==========
def create_interface():
    """创建Gradio界面"""
    
    with gr.Blocks(
        title="多模态聊天机器人",
        theme=gr.themes.Soft(),
        css="""
        .chatbot { min-height: 500px; }
        .input-row { margin-top: 20px; }
        """
    ) as demo:
        
        gr.Markdown("""
        # 🤖 多模态聊天机器人
        支持**文字**、**图片**、**语音**输入 | 具备上下文记忆
        """)
        
        # 聊天窗口
        chatbot = gr.Chatbot(
            label="对话历史",
            height=500,
            bubble_full_width=False,
            show_copy_button=True
        )
        
        # 输入区域
        with gr.Row(variant="panel", elem_classes="input-row"):
            text_input = gr.Textbox(
                placeholder="输入文字消息...",
                label="文本输入",
                scale=4,
                container=False
            )
            submit_btn = gr.Button("发送", variant="primary", scale=1)
        
        # 多媒体输入
        with gr.Row():
            audio_input = gr.Audio(
                sources=["microphone"],
                type="filepath",
                label="语音输入",
                interactive=True
            )
            image_input = gr.Image(
                type="filepath",
                label="图片上传",
                sources=["upload"],
                interactive=True
            )
        
        # 绑定事件
        submit_btn.click(
            fn=process_inputs,
            inputs=[text_input, audio_input, image_input, chatbot],
            outputs=[text_input, audio_input, image_input, chatbot],
            queue=True
        )
        
        # 回车键提交
        text_input.submit(
            fn=process_inputs,
            inputs=[text_input, audio_input, image_input, chatbot],
            outputs=[text_input, audio_input, image_input, chatbot],
            queue=True
        )
        
        # 清理按钮
        with gr.Row():
            clear_btn = gr.Button("清空对话", variant="secondary")
            clear_btn.click(
                fn=lambda: ([], "", None, None),
                outputs=[chatbot, text_input, audio_input, image_input],
                queue=False
            )
    
    return demo

# ========== 5. 启动应用 ==========
if __name__ == "__main__":
    demo = create_interface()
    
    demo.queue(
        max_size=20,
        default_concurrency_limit=5
    )
    
    demo.launch(
        server_name="0.0.0.0",
        server_port=7860,
        share=False,  # 设为True可生成公网链接
        show_error=True,
        debug=True
    )

三、运行与测试

1. 启动应用

1
python main.py

2. 访问地址

  • 本地: http://localhost:7860
  • 局域网: http://[服务器IP]:7860

3. 功能测试

  • ✅ 文本对话
  • ✅ 图片识别与描述
  • ✅ 语音转文字
  • ✅ 上下文记忆
  • ✅ 对话历史保存

四、常见问题

Q1: 显存不足

  • 降低 max_model_len 参数
  • 设置 --gpu-memory-utilization 0.8
  • 使用量化版本模型

Q2: 音频处理失败

  • 安装 ffmpeg: sudo apt install ffmpeg
  • 检查音频格式支持

Q3: 响应缓慢

  • 调整 max_tokens 限制
  • 使用 stream=True 启用流式输出
  • 升级服务器配置

附录：API调用方案

如果不方便部署，可以使用阿里云、OpenAI等API服务：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 阿里云通义千问
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="qwen-max",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key="your-api-key"
)

# OpenAI GPT-4V
llm = ChatOpenAI(
    model="gpt-4-vision-preview",
    api_key="your-openai-key"
)

总结

本文详细介绍了如何从零构建一个功能完整的多模态聊天机器人。通过私有化部署Qwen-2.5-Omni模型，结合LangChain和Gradio，我们实现了：

  1. 多模态支持：文本、图片、语音输入
  2. 上下文记忆：SQLite持久化存储
  3. Web界面：直观的用户交互
  4. 可扩展架构：模块化设计，易于扩展

项目代码已开源，可根据实际需求进行调整和优化。欢迎您宝贵的建议~

许可协议

本文采用 署名-非商业性使用-相同方式共享 4.0 国际 许可协议，转载请注明出处。

评论区