1.作者介绍韦昊男西安工程大学电子信息学院2025级研究生研究方向机器视觉与人工智能电子邮件2154548512qq.com程锡贵男西安工程大学电子信息学院2025级研究生张宏伟人工智能题组研究方向机器视觉与人工智能电子邮件15327178796163.com2关于理论方面的知识介绍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 DiffusionSDXL、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 长文本时依然保持高速和稳定避免了性能下降和显存溢出问题图三 MOEDeepSeek中的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, encodingutf-8) as f: for line in f: line line.strip() if line: data json.loads(line) messages.append(Message( roledata[role], contentdata[content], timestampdata.get(timestamp) )) return messages def append_message(self, message: Message): 追加一条消息到文件 with open(self.history_file, a, encodingutf-8) as f: f.write(json.dumps(message.to_dict(), ensure_asciiFalse) \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: fBearer {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, streamFalse) # 发起同步 POST 请求 response requests.post( f{self.base_url}/chat/completions, headersself.headers, jsonpayload, timeout60 ) # 检查 HTTP 状态码非 200 则抛出异常 if response.status_code ! 200: raise Exception(fAPI 请求失败: {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, streamTrue) # 发起 POST 请求开启 streamTrue 以接收 Server-Sent Events (SSE) 数据流 response requests.post( f{self.base_url}/chat/completions, headersself.headers, jsonpayload, streamTrue, timeout60 ) if response.status_code ! 200: raise Exception(fAPI 请求失败: {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(roleuser, contentuser_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(roleassistant, contentresponse) 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(roleuser, contentuser_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(roleassistant, contentfull_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_pathconfig.ini): config configparser.ConfigParser() if not os.path.exists(config_path): raise FileNotFoundError(f配置文件 {config_path} 不存在) config.read(config_path, encodingutf-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_urlconfig[base_url], api_keyapi_key, modelconfig[model], max_tokensconfig[max_tokens], temperatureconfig[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 使得“助手: ”打印后不换行flushTrue 强制刷新缓冲区使文字立即显示 print(助手: , end, flushTrue) # 遍历服务层产生的每一个增量文本块 (Token) for token in chat_service.send_message_stream(user_input): # 逐个字打印出来。end 保证字与字紧密相连flushTrue 才能形成行云流水的打字机效果 print(token, end, flushTrue) # 当整个流式生成结束后手动打印一个换行符为下一次“你:”的输入留出空间 print(\n) else: # 非流式分支直接调用并同步等待获取完整的文本结果 response chat_service.send_message(user_input) print(f助手: {response}\n) except KeyboardInterrupt: # 捕获用户在终端按下 CtrlC 的行为优雅退出而不是抛出一堆乱七八糟的报错 print(\n退出程序) break except Exception as e: # 捕获异常如网络超时、API 报错等保证当前这轮报错后循环不中断依然能继续下一次对话 print(f发生错误: {e}\n) if __name__ __main__: main() requirements.txt requests2.31.03.2运行结果4.参考连接硅基流动介绍https://blog.csdn.net/fyfugoyfa/article/details/145430651deepseekV3介绍https://blog.csdn.net/weixin_50917576/article/details/145601149