调用深度求索的智能对话
·
1.作者介绍
韦昊，男，西安工程大学电子信息学院，2025级研究生
研究方向：机器视觉与人工智能
电子邮件：2154548512@qq.com
程锡贵，男，西安工程大学电子信息学院，2025级研究生，张宏伟人工智能题组
研究方向：机器视觉与人工智能
电子邮件：15327178796@163.com
2．关于理论方面的知识介绍
2.1 硅基流动平台一站式大模型云服务
硅基流动（SiliconCloud）是硅基流动公司推出的一站式大模型云服务平台 ，致力于为开发者和企业提供高效、低成本且全面的生成式人工智能（GenAI）模型服务，包括 IaaS、PaaS 和 MaaS。其核心目标是通过优化大模型使用体验，帮助用户实现 “Token 自由”，即以更低成本和更高效率使用先进的大语言模型（LLMs）及其他生成式人工智能（AI）模型。
主要特点与功能
丰富的大模型支持：平台集成多种主流开源大模型，涵盖但不限于：文本生成模型（如 DeepSeek R1 & V3、Qwen2.5、GLM-4、Llama-3.X、Gemma-2、InternLM、Yi-1.5 等）和图片生成模型（如 Janus-Pro、Stable Diffusion（SDXL）、FLUX 等）。
多模态模型：支持文本生成语音、文本生成图像、文本生成视频等功能。代码生成模型包括 Qwen2.5-Coder-32B-Instruct 等。
高性价比优势：提供行业内较低的 API 调用价格。对于 9B 及以下的模型，平台提供永久免费 API 服务，适合个人开发者及小型项目使用。
图一 硅基流动首页图
2.2 deepseek-ai/DeepSeek-V3
图二 模型架构图
模型定位: 深度求索（DeepSeek-AI）推出的新一代 MoE（混合专家）架构通用大语言模型，旨在实现高性能与低成本推理的完美平衡。
核心架构: MoE 稀疏激活架构，拥有 671B 的总参数量，但每次推理仅激活 37B 参数，兼顾了强大的模型容量与高效的计算效率。
上下文窗口: 支持高达 128K 的超长上下文，能够轻松处理长文档、代码库、法律合同等复杂任务
长文本优化（MLA）: 采用多头潜在注意力（MLA）技术，大幅压缩 KV 缓存，使得在处理 128K 长文本时依然保持高速和稳定，避免了性能下降和显存溢出问题
图三 MOE
DeepSeek中的MOE
为了在负载均衡 和模型性能之间取得更好的平衡，DeepSeek开创了一种无辅助损失的负载均衡策略：为每个专家引入一个偏差项，并将其添加到相应的亲和力分数中以确定top-K路由，具体来说：如果其对应的专家过载，我们将偏差项减少b ；如果其对应的专家负载不足，我们将偏差项增加b ，其中b 是一个称为偏差更新速度的超参数。
图四 MLA 多头潜在注意力
MLA 多头潜在注意力
本质就是在计算注意力之前进行各种向量降维(用于降低注意力计算和后端推理的计算量)和加入旋转位置编码(RoPE)(目的是：通过旋转矩阵将位置信息融入词向量的内积计算中，从而在注意力机制中隐式编码相对位置关系)。
2.3 硅基流动API接入
3．关于实验过程的介绍，完整实验代码，测试结果
3.1完整代码
目录结构
|
|
config.ini: 存储API配置（base_url, api_key, model等）。 entities: 定义消息实体，用于统一对话格式。 providers/llm_provider.py: 封装API调用逻辑，包括非流式和流式。 providers/chat_repository.py: 管理对话历史存储和加载。 services: 对话服务，整合 LLM 调用和历史管理。 utils: 加载 config.ini 配置。 main.py: 主入口，实现终端交互循环。 requirements.txt: 列出依赖 |
entities下的message.py
from typing import Literal
from dataclasses import dataclass
from datetime import datetime
@dataclass
class Message:
role: Literal["user", "assistant", "system"]
content: str
timestamp: str = None
def __post_init__(self):
if self.timestamp is None:
self.timestamp = datetime.now().isoformat()
def to_dict(self):
return {
"role": self.role,
"content": self.content,
"timestamp": self.timestamp
}
providers下的chat_repository.py
import json
import os
from typing import List
from entities.message import Message
class ChatRepository:
def __init__(self, history_file: str):
self.history_file = history_file
def load_history(self) -> List[Message]:
"""从文件加载历史消息"""
if not os.path.exists(self.history_file):
return []
messages = []
with open(self.history_file, 'r', encoding='utf-8') as f:
for line in f:
line = line.strip()
if line:
data = json.loads(line)
messages.append(Message(
role=data["role"],
content=data["content"],
timestamp=data.get("timestamp")
))
return messages
def append_message(self, message: Message):
"""追加一条消息到文件"""
with open(self.history_file, 'a', encoding='utf-8') as f:
f.write(json.dumps(message.to_dict(), ensure_ascii=False) + '\n')
def clear_history(self):
"""清空历史文件"""
if os.path.exists(self.history_file):
os.remove(self.history_file)
providers下的llm_prodiver.py
import requests
import json
from typing import List, Dict, Generator, Optional
# 假设这是你项目中定义的 Message 实体类
from entities.message import Message
class LLMProvider:
def __init__(self, base_url: str, api_key: str, model: str, max_tokens: int, temperature: float):
# 移除 URL 末尾的斜杠，防止拼接路径时出现双斜杠
self.base_url = base_url.rstrip('/')
self.api_key = api_key
self.model = model
self.max_tokens = max_tokens
self.temperature = temperature
# 构造标准的 API 请求头，通常采用 Bearer Token 鉴权方式
self.headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}"
}
def _build_payload(self, messages: List[Dict], stream: bool) -> Dict:
"""构建发送给大模型 API 的请求体 (遵循 OpenAI 接口规范)"""
return {
"model": self.model,
"messages": messages,
"max_tokens": self.max_tokens,
"temperature": self.temperature,
"stream": stream, # 控制是否开启流式输出
"top_p": 0.9 # 控制生成文本的多样性
}
def chat_completion(self, messages: List[Dict]) -> str:
"""非流式调用，等待模型生成完毕后返回完整回答"""
payload = self._build_payload(messages, stream=False)
# 发起同步 POST 请求
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json=payload,
timeout=60
)
# 检查 HTTP 状态码，非 200 则抛出异常
if response.status_code != 200:
raise Exception(f"API 请求失败: {response.status_code} - {response.text}")
# 解析返回的 JSON 数据
data = response.json()
# 从标准的 OpenAI 响应结构中提取最终回复的文本内容
return data["choices"][0]["message"]["content"]
def chat_completion_stream(self, messages: List[Dict]) -> Generator[str, None, None]:
"""流式调用，逐个产出 (yield) token，用于实现打字机效果"""
payload = self._build_payload(messages, stream=True)
# 发起 POST 请求，开启 stream=True 以接收 Server-Sent Events (SSE) 数据流
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json=payload,
stream=True,
timeout=60
)
if response.status_code != 200:
raise Exception(f"API 请求失败: {response.status_code} - {response.text}")
# 核心逻辑：遍历流式响应的每一行数据
for line in response.iter_lines():
if line:
# 将 bytes 解码为字符串
line = line.decode('utf-8')
# SSE 协议标准：数据通常以 "data: " 开头
if line.startswith("data: "):
# 截取 "data: " 后面的实际 JSON 字符串
data_str = line[6:]
# API 约定：以 "data: [DONE]" 标识流式输出结束
if data_str == "[DONE]":
break
try:
# 解析这一数据块 (chunk) 的 JSON
chunk = json.loads(data_str)
# 获取当前数据块的增量内容 (delta)
delta = chunk["choices"][0]["delta"]
# 只有当 delta 中包含 "content" 字段时，才说明生成了新的文本内容
if "content" in delta:
# 产出 (yield) 这个字/词，供外部循环调用实时显示
yield delta["content"]
except json.JSONDecodeError:
# 如果出现 JSON 解析错误（例如数据截断），直接跳过继续处理下一行
continue
service下的chat_service.py
from typing import List, Generator
from entities.message import Message
from providers.llm_provider import LLMProvider
from providers.chat_repository import ChatRepository
class ChatService:
def __init__(self, llm_provider: LLMProvider, repository: ChatRepository, system_prompt: str = None):
self.llm = llm_provider
self.repo = repository
self.system_prompt = system_prompt
def _build_messages_for_api(self, history: List[Message]) -> List[dict]:
"""将内部的 Message 实体列表转换为大模型 API 所需的标准字典数组格式"""
messages = []
# 核心逻辑：如果设置了 system prompt（系统提示词），必须将其作为第一条消息注入
# 这决定了 AI 的人设、回答格式和核心行为准则
if self.system_prompt:
messages.append({"role": "system", "content": self.system_prompt})
# 遍历历史对话记录，将上下文按顺序加入，让大模型拥有记忆
for msg in history:
messages.append({"role": msg.role, "content": msg.content})
return messages
# --- 拆分 1：非流式方法（只负责 return） ---
def send_message(self, user_input: str) -> str:
"""发送用户消息，获取助手完整回复（非流式）。"""
# 1. 从存储介质（数据库/本地文件）中加载之前的历史对话
history = self.repo.load_history()
user_msg = Message(role="user", content=user_input)
# 2. 构造包含历史记录的 API 消息体，并将当前用户的最新输入追加到末尾
api_messages = self._build_messages_for_api(history)
api_messages.append({"role": "user", "content": user_input})
# 3. 发起网络请求，阻塞等待大模型生成完整的回复
response = self.llm.chat_completion(api_messages)
# 4. 核心持久化逻辑：将“用户的问题”和“AI的完整回答”保存到本地记录中
# 这样下一次调用 load_history() 时就能带上这次的对话
self.repo.append_message(user_msg)
assistant_msg = Message(role="assistant", content=response)
self.repo.append_message(assistant_msg)
return response
# --- 拆分 2：流式方法（只负责 yield） ---
def send_message_stream(self, user_input: str) -> Generator[str, None, None]:
"""发送用户消息，流式获取助手回复。"""
history = self.repo.load_history()
user_msg = Message(role="user", content=user_input)
api_messages = self._build_messages_for_api(history)
api_messages.append({"role": "user", "content": user_input})
# 核心逻辑：定义一个空字符串，用于在流式产出时“拼接”完整的回答
# 因为我们虽然是一段段发给前端，但最终保存到数据库里需要一段完整的文本
full_response = ""
# 遍历底层 LLMProvider yield 出来的每一个 token（字/词）
for token in self.llm.chat_completion_stream(api_messages):
# 将每个 token 拼接到完整回复中
full_response += token
# 实时将 token 抛给上层调用者（通常是 UI 层），实现打字机效果
yield token
# 核心持久化逻辑：当 for 循环结束时，意味着流式输出完毕，full_response 已经拼接完整。
# 此时再将用户消息和拼接完成的 AI 消息一同持久化保存。
self.repo.append_message(user_msg)
assistant_msg = Message(role="assistant", content=full_response)
self.repo.append_message(assistant_msg)
def clear_conversation(self):
"""清空对话历史"""
self.repo.clear_history()
utils下的config_loader.py
import configparser
import os
def load_config(config_path="config.ini"):
config = configparser.ConfigParser()
if not os.path.exists(config_path):
raise FileNotFoundError(f"配置文件 {config_path} 不存在")
config.read(config_path, encoding='utf-8')
api_config = config["API"]
return {
"base_url": api_config.get("base_url"),
"api_key": api_config.get("api_key"),
"model": api_config.get("model"),
"max_tokens": api_config.getint("max_tokens"),
"temperature": api_config.getfloat("temperature"),
"stream": api_config.getboolean("stream"),
"save_history": api_config.getboolean("save_history"),
"history_file": api_config.get("history_file")
}
配置文件config.ini
# 配置组标识，说明接下来的配置项都属于 API 相关的参数
[API]
# 大模型 API 的基础请求路径（这里使用的是硅基流动 SiliconFlow 的 OpenAI 兼容接口）
base_url = https://api.siliconflow.cn/v1
# 你的 API 密钥/鉴权令牌（需要从服务商后台获取并填入，注意不要泄露给他人）
api_key = sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxx
# 指定调用的具体大模型名称（这里使用的是 DeepSeek-V3 模型）
model = deepseek-ai/DeepSeek-V3
# 单次模型生成回复的最大 Token 数量限制（控制单次回答的长度，防止消耗过多额度）
max_tokens = 2000
# 采样温度，控制生成文本的随机性/创造性（值在 0 到 2 之间，0.7 属于日常聊天平衡严谨与创意的推荐值）
temperature = 0.7
# 是否开启流式传输开关（true 表示开启，模型会像打字机一样逐字返回结果，提升用户体验）
stream = true
# 是否保存对话历史记录的开关（true 表示开启持久化存储，下次启动可以加载记忆）
save_history = true
# 历史记录保存的文件路径和名称（这里采用 .jsonl 格式，即每行一条完整 JSON，非常适合按行追加对话数据）
history_file = conversation_history.jsonl
main.py
import sys
# 导入自定义的配置加载器、模型服务商、历史仓储以及核心对话服务
from utils.config_loader import load_config
from providers.llm_provider import LLMProvider
from providers.chat_repository import ChatRepository
from services.chat_service import ChatService
def main():
# ----------------------------------------------------------------
# 1. 初始化与配置校验阶段
# ----------------------------------------------------------------
try:
# 读取 config.ini / config.json 等配置文件的参数
config = load_config()
except Exception as e:
print(f"加载配置失败: {e}")
return
api_key = config["api_key"]
# 安全检查：防止用户忘记修改占位符直接运行程序
if not api_key or api_key == "YOUR_API_KEY_HERE":
print("请先在 config.ini 中填写有效的 API Key")
return
# ----------------------------------------------------------------
# 2. 依赖注入与组件组装
# ----------------------------------------------------------------
# 初始化底层的 API 调用器
llm = LLMProvider(
base_url=config["base_url"],
api_key=api_key,
model=config["model"],
max_tokens=config["max_tokens"],
temperature=config["temperature"]
)
# 初始化历史记录读写仓储（负责读写 JSONL 文件）
repo = ChatRepository(config["history_file"])
# 设定全局的系统提示词（人设）
system_prompt = "你是智能对话机器人"
# 将底层基础组件“注入”到核心业务层 ChatService 中
chat_service = ChatService(llm, repo, system_prompt)
# ----------------------------------------------------------------
# 3. 终端交互循环 (REPL 保持程序不退出)
# ----------------------------------------------------------------
print("===== 智能对话终端 (DeepSeek-V3) =====")
print("输入 'exit' 退出，输入 'clear' 清空对话历史\n")
while True:
try:
# 获取用户输入并去除首尾空格
user_input = input("你: ").strip()
# 命令检查：退出程序
if user_input.lower() == "exit":
print("再见！")
break
# 命令检查：清空上下文缓存和本地历史文件
elif user_input.lower() == "clear":
chat_service.clear_conversation()
print("对话历史已清空。\n")
continue
# 空输入拦截：防止把空字符串发给大模型浪费算力
elif not user_input:
continue
# 从配置文件获取是否开启流式的布尔值（默认不开启）
use_stream = config.get("stream", False)
# ------------------------------------------------------------
# 4. 对话路由分支（流式 VS 非流式）
# ------------------------------------------------------------
if use_stream:
# end="" 使得“助手: ”打印后不换行；flush=True 强制刷新缓冲区使文字立即显示
print("助手: ", end="", flush=True)
# 遍历服务层产生的每一个增量文本块 (Token)
for token in chat_service.send_message_stream(user_input):
# 逐个字打印出来。end="" 保证字与字紧密相连，flush=True 才能形成行云流水的打字机效果
print(token, end="", flush=True)
# 当整个流式生成结束后，手动打印一个换行符，为下一次“你:”的输入留出空间
print("\n")
else:
# 非流式分支：直接调用并同步等待获取完整的文本结果
response = chat_service.send_message(user_input)
print(f"助手: {response}\n")
except KeyboardInterrupt:
# 捕获用户在终端按下 Ctrl+C 的行为，优雅退出而不是抛出一堆乱七八糟的报错
print("\n退出程序")
break
except Exception as e:
# 捕获异常（如网络超时、API 报错等），保证当前这轮报错后，循环不中断，依然能继续下一次对话
print(f"发生错误: {e}\n")
if __name__ == "__main__":
main()
requirements.txt
requests>=2.31.0
3.2运行结果
4.参考连接
硅基流动介绍：
https://blog.csdn.net/fyfugoyfa/article/details/145430651
deepseekV3介绍：
https://blog.csdn.net/weixin_50917576/article/details/145601149
汇聚全球AI编程工具，助力开发者即刻编程。
更多推荐
10
0- 0
已为社区贡献6条内容
所有评论(0)