基于仓颉 + DeepSeek + MCP 的智能膳食分析助手一、项目背景及需求分析1.1 项目背景随着人们对健康饮食关注度的提升，越来越多用户希望借助 AI 助手实现个性化的饮食分析与管理。然而，目前市面上的饮食类应用普遍存在如下痛点：1️⃣. 缺乏智能分析：大多数仅记录卡路里，无法提供专业点评与优化建议；2️⃣. 知识更新不及时：难以结合最新营养研究进行推荐与判断；3️⃣. 缺乏可扩展性：难以适配特定人群（如高血压、糖尿病、健身人群等）的差异需求。本项目旨在构建一个基于大语言模型（DeepSeek）和结构化协议（MCP）的智能饮食健康助手，通过自然语言交互，帮助用户实现饮食数据结构化、健康风险识别与个性化建议生成。1.2 核心需求1️⃣ 食品实体识别能力系统应具备从自然语言中准确识别出饮食相关的食品实体与关键词的能力。例如，用户输入：“我今天吃了煎饼果子、卤牛肉、奶茶”，系统应能够自动识别出食品列表：["煎饼果子", "卤牛肉", "奶茶"] 该能力是后续营养分析与健康判断的基础，应支持中文短语、多食物组合、模糊词等多种表达方式。2️⃣ 饮食结构化分析能力（MCP 中间层处理）系统应通过 MCP 模块完成对用户输入饮食内容的语义解析与结构化处理，主要包括：应提取食品清单；应识别饮食行为时段（如早餐、晚餐、夜宵等）；应对识别出的食物打上健康标签（如高糖、高油脂、高钠、低纤维等）；应构造符合 LLM 输入规范的 Prompt，以支撑后续模型理解与生成。结构化分析结果应作为 DeepSeek 模型调用的核心中间数据。3️⃣ 合理饮食建议生成能力（基于 DeepSeek）系统应利用 DeepSeek 大模型，根据 MCP 构造的 Prompt 生成具有科学性、专业性与可操作性的饮食建议，具体应包括：对当前饮食存在的健康风险进行分析（如糖分摄入超标、缺乏蔬菜等）；提供可替代食材建议（如炸鸡 → 烤鸡，奶茶 → 绿茶）；给出搭配优化方案（如餐前饮水、搭配高纤维蔬菜等）；提出个性化提示（如适合三高人群、健身人群、孕期饮食等）。模型输出应以自然语言形式呈现，逻辑清晰、结构明确，风格应贴近真实营养师表达方式。二、项目概述本项目旨在构建一套基于多模块协作的 智能膳食分析助手，通过自然语言输入，实现对个人饮食行为的结构化分析与健康建议生成，服务于日常健康管理、饮食优化与疾病预防等场景。系统采用模块化设计，主要由三大核心组成部分构成：2.1 技术选型模块技术栈职责🧑‍💻 客户交互仓颉语言提供终端或界面交互入口，接收用户自然语言输入并展示模型生成结果🔧 中间处理MCP 模块（Python + FastAPI）实现饮食语义解析、食物识别、健康标签打标、Prompt 生成与请求路由🤖 模型服务DeepSeek 大语言模型接收经过标准化构造的请求 Prompt，返回结构化营养建议、健康分析与饮食优化建议2.2 适用对象企业健康服务平台产品经理个人开发者高校学生营养师/健身爱好者病患群体2.3 案例时间环境配置：约30 分钟构建项目：约25分钟测试阶段：约10 分钟本案例总时长预计 65 分钟。2.4 项目流程整个系统的数据处理与响应流程如下所示：2.5 资源总览本案例预计花费 0 元。资源名称规格免费额度/价格开发者空间–云主机2CPU4g X86 Ubuntu 22.04 Server定制版0元CodeArts IDE for CangjieIDE 仓颉 云主机预装0元CodeArts IDE for PythonIDE Python 云主机预装0元Maas DeepSeek 模型服务试用200W Token0元2.6 架构亮点✅ 使用 MCP 解耦输入处理与模型调用，提升可扩展性与灵活性；✅ DeepSeek 专注自然语言生成任务，MCP 预处理提高 prompt 精度；✅ 仓颉终端可后续拓展为图形界面、Web 应用或移动端入口；✅ 本地配置与食物字典可按需调整，适配多种用户类型（如减脂、控糖、健身等）；✅ 架构支持多模型/多来源接入，可拓展至 Claude、SparkDesk、ChatGPT 等后端模型。三、环境准备3.1 申请云主机面向广大开发者群体，华为开发者空间提供一个随时访问的“开发桌面云主机”、丰富的“预配置工具集合”和灵活使用的“场景化资源池”，开发者开箱即用，快速体验华为根技术和资源。如果还没有领取开发者空间云主机，可以参考免费领取云主机文档领取。领取云主机后可以直接进入华为开发者空间工作台界面，点击进入桌面连接云主机。3.2 免费领取DeepSeek R1满血版华为云提供了单模型200万免费Tokens，包含DeepSeek-R1&V3满血版，我们可以登录华为云ModelArts Studio(MaaS)控制台领取免费额度，这里我们选择DeepSeek-R1满血版来搭建我们的专属AI聊天机器人。在云主机桌面底部菜单栏，点击打开火狐浏览器。用火狐浏览器访问ModelArts Studio首页：cid:link_2，点击ModelArts Studio控制台跳转到登录界面，按照登录界面提示登录，即可进入ModelArts Studio控制台。签署免责声明。进入ModelArts Studio控制台首页，区域选择西南-贵阳一，在左侧菜单栏，选择在线推理 > 预置服务 > 免费服务，选择DeepSeek-R1-32K模型，点击领取额度，领取200万免费token。领取后点击调用说明，可以获取到对应的API地址、模型名称。红色方块为API 地址 红色箭头为 API KEY四、项目构建本案例采用 本地MCP 服务+ 本地仓颉AI机器人对话 (参考来自于 (https://gitcode.com/CaseDeveloper/Cangjie-Examples)4.1 MCP 服务构建4.1.1 打开 CodeArts IDE for Python4.1.2 新建项目项目名称 构建为： food_mcp 然后点击 创建4.1.3 代码实施mcp_server.py# -*- coding: utf-8 -*- # mcp_server.py """基于仓颉 + DeepSeek + MCP 的智能膳食分析助手""" from fastapi import FastAPI from pydantic import BaseModel import json from typing import List, Dict, Any from fastapi.responses import StreamingResponse from deepseek_client import stream_deepseek, async_stream_deepseek from text_match import extract_food_simple, extract_food from loguru import logger from cachetools import TTLCache import sys # 加载食物字典 with open("food_tags.json", encoding="utf-8") as f: FOOD_TAGS: Dict[str, Dict[str, Any]] = json.load(f) app = FastAPI() class Input(BaseModel): input: str # 全局缓存：prompt -> full_reply_text CACHE = TTLCache(maxsize=500, ttl=1800) # 30分钟自动过期 logger.remove() logger.add(sys.stdout, level="INFO", enqueue=False, backtrace=False) # ---- 基础角色词 ---- BASE_SYSTEM_PROMPT = ( "你是一名资深注册营养师，擅长以简洁的 Markdown 格式给出科学、可执行的饮食建议。" "所有回答需包含以下小节：\n" "1. 总体评价\n" "2. 推荐摄入（量/食材）\n" "3. 过量风险\n" "4. 不宜同食\n" "5. 行动建议\n" "回答请使用中文，并尽量在 300 字以内。" ) # ---- 场景化模板 ---- PROMPTS = { "nutrition_review": ( BASE_SYSTEM_PROMPT + "\n\n【场景】饮食点评。请先总体评价，再按上表 1~5 小节输出：{input}" ), "diet_plan": ( BASE_SYSTEM_PROMPT + "\n\n【场景】减脂期餐单。请输出 3 日食谱 (表格形式)，并在每餐注明热量估计：{input}" ), "effects_inquiry": ( BASE_SYSTEM_PROMPT + "\n\n【场景】过量影响。请列出已知副作用及参考文献：{input}" ), "avoid_inquiry": ( BASE_SYSTEM_PROMPT + "\n\n【场景】相克查询。请说明不宜同食原因及替代方案：{input}" ), } def _process_input(user_input: str): """内部共用逻辑，返回分析结果和 prompt""" # === 1. 食物关键字匹配 === food_list, exact_hits, fuzzy_hits = extract_food(user_input, FOOD_TAGS.keys()) # 记录匹配详情到日志 logger.info(f"[匹配] 精确={exact_hits} 模糊={fuzzy_hits}") # === 2. 聚合静态信息 === tags = list({tag for food in food_list for tag in FOOD_TAGS[food].get("tags", [])}) effects = {food: FOOD_TAGS[food].get("effects") for food in food_list if FOOD_TAGS[food].get("effects")} avoid_with = list({aw for food in food_list for aw in FOOD_TAGS[food].get("avoid_with", [])}) diet_type = list({dt for food in food_list for dt in FOOD_TAGS[food].get("diet_type", [])}) # === 3. 根据关键词判定 mode === mode = "nutrition_review" if any(k in user_input for k in ["减肥", "减脂", "低卡", "少油", "瘦身", "个人食谱"]): mode = "diet_plan" elif any(k in user_input for k in ["吃多了", "过量", "上火", "副作用", "影响"]): mode = "effects_inquiry" elif any(k in user_input for k in ["不能一起", "相克", "不宜同食", "一起吃"]): mode = "avoid_inquiry" new_prompt_base = PROMPTS[mode].format(input=user_input) # === 4. 拼接静态附加信息 === notes = [] if effects: notes.append("【过量风险】" + "；".join(f"{food}:{desc}" for food, desc in effects.items())) if avoid_with: notes.append("【不宜同食】" + "、".join(avoid_with)) extra_info = "；".join(notes) if extra_info: new_prompt_base += ( "\n\n以下为已知静态信息（请务必先列出【过量风险】与【不宜同食】两个小节，并完整引用下列内容，否则视为回答不完整）：" f"{extra_info}" ) # 要求模型以 JSON 输出，配合 response_format new_prompt = new_prompt_base + "\n请使用 markdown bullet list 输出建议。" return { "food_list": food_list, "tags": tags, "effects": effects, "avoid_with": avoid_with, "diet_type": diet_type, "mode": mode, "routed_input": new_prompt, } @app.post("/mcp") async def route_prompt(data: Input): user_input = data.input result = _process_input(user_input) return result @app.post("/mcp_stream") async def route_prompt_stream(data: Input): user_input = data.input result = _process_input(user_input) async def event_generator(): meta_json = json.dumps({k: v for k, v in result.items() if k != 'routed_input'}, ensure_ascii=False) # 发送 Meta 事件 yield f"event: meta\ndata: {meta_json}\n\n" prompt_key = result["routed_input"] # 若缓存命中，直接按20字符切片发送缓存内容 if prompt_key in CACHE: logger.info("[缓存] 命中") cached_text = CACHE[prompt_key] # SSE data 行不能包含裸换行，需逐行加前缀 payload_lines = [f"data: {line}" for line in cached_text.split("\n")] payload_block = "\n".join(payload_lines) yield f"event: token\n{payload_block}\n\n" yield "event: done\ndata: [DONE]\n\n" return logger.info("[缓存] 未命中") collected_chunks = [] async for chunk in async_stream_deepseek(prompt_key): collected_chunks.append(chunk) yield f"event: token\ndata: {chunk}\n\n" # 缓存完整内容 full_text = "".join(collected_chunks) CACHE[prompt_key] = full_text # 结束事件 yield "event: done\ndata: [DONE]\n\n" return StreamingResponse(event_generator(), media_type="text/event-stream") deepseek_client.py# deepseek_client.py # 基于 DeepSeek API 的客户端 import requests import json from typing import Generator import httpx # 从 config.json 读取配置 with open("config.json", encoding="utf-8") as f: config = json.load(f) DEEPSEEK_API_KEY = config["DEEPSEEK_API_KEY"] DEEPSEEK_URL = config["DEEPSEEK_URL"] MODEL_NAME = config["MODEL_NAME"] def call_deepseek(prompt): headers = { "Content-Type": "application/json", "Authorization": f"Bearer {DEEPSEEK_API_KEY}" } payload = { "model": MODEL_NAME, "messages": [ {"role": "user", "content": prompt} ] } response = requests.post(DEEPSEEK_URL, headers=headers, json=payload) result = response.json() return result["choices"][0]["message"]["content"] # 新增：流式推理生成器 def stream_deepseek(prompt) -> Generator[str, None, None]: """Yield generated content chunks from DeepSeek API using SSE.""" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {DEEPSEEK_API_KEY}" } payload = { "model": MODEL_NAME, "messages": [ {"role": "user", "content": prompt} ], "stream": True, "max_tokens": 2048, "stream_options": {"include_usage": True}, } # 使用 stream=True 触发增量输出 response = requests.post(DEEPSEEK_URL, headers=headers, json=payload, stream=True) # iter_lines 将保持连接并逐行读取 for line in response.iter_lines(decode_unicode=True): if not line: continue # DeepSeek/SSE 行以 "data: " 开头 if line.startswith("data: "): data_str = line[6:] # 结束标志 if data_str.strip() == "[DONE]": break # 解析 JSON，获取增量内容 try: data_json = json.loads(data_str) # usage 块 choices 为空，跳过 choices = data_json.get("choices", []) if not choices: continue delta = choices[0]["delta"].get("content", "") if delta: yield delta except json.JSONDecodeError: # 忽略无法解析的片段 continue # 异步流式生成器 async def async_stream_deepseek(prompt): """异步生成器，从 DeepSeek API 生成内容（SSE）。""" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {DEEPSEEK_API_KEY}" } payload = { "model": MODEL_NAME, "messages": [ {"role": "user", "content": prompt} ], "stream": True, "max_tokens": 2048, "stream_options": {"include_usage": True}, } async with httpx.AsyncClient(timeout=None) as client: async with client.stream("POST", DEEPSEEK_URL, headers=headers, json=payload) as resp: async for line in resp.aiter_lines(): if not line: continue if line.startswith("data: "): data_str = line[6:] if data_str.strip() == "[DONE]": break try: data_json = json.loads(data_str) choices = data_json.get("choices", []) if not choices: continue delta = choices[0]["delta"].get("content", "") if delta: yield delta except json.JSONDecodeError: continue test_pipeline.py# -*- coding: utf-8 -*- # test_pipeline.py import requests from deepseek_client import call_deepseek import json from requests.exceptions import ChunkedEncodingError def test_diet_analysis(): user_input = "中午吃了炸鸡和奶茶，晚上吃了牛肉火锅" # 调用 MCP mcp_resp = requests.post("http://localhost:8001/mcp", json={"input": user_input}) mcp_data = mcp_resp.json() print("MCP 返回：", mcp_data) # 调用 DeepSeek reply = call_deepseek(mcp_data["routed_input"]) print("DeepSeek 回复：", reply) def test_diet_plan(): user_input = "我想要一个一周减肥计划，主要食物包括香菇、鸡胸肉和燕麦" mcp_resp = requests.post("http://localhost:8001/mcp", json={"input": user_input}) mcp_data = mcp_resp.json() print("MCP 返回：", mcp_data) reply = call_deepseek(mcp_data["routed_input"]) print("DeepSeek 回复：", reply) def test_effects_inquiry(): user_input = "香菇吃多了会怎样？" mcp_resp = requests.post("http://localhost:8001/mcp", json={"input": user_input}) mcp_data = mcp_resp.json() print("MCP 返回：", mcp_data) reply = call_deepseek(mcp_data["routed_input"]) print("DeepSeek 回复：", reply) def test_avoid_inquiry(): user_input = "牛奶和虾能一起吃吗？" mcp_resp = requests.post("http://localhost:8001/mcp", json={"input": user_input}) mcp_data = mcp_resp.json() print("MCP 返回：", mcp_data) reply = call_deepseek(mcp_data["routed_input"]) print("DeepSeek 回复：", reply) def test_stream(): user_input = "我晚上吃什么比较好呢？ 中午吃了炸鸡和奶茶" try: resp = requests.post( "http://localhost:8001/mcp_stream", json={"input": user_input}, stream=True, timeout=None, ) print("Stream start →", resp.status_code) meta_printed = False for line in resp.iter_lines(decode_unicode=True): if not line: continue # SSE: 可能含 event: xxx if line.startswith("event: "): current_event = line[7:] # 下一行应该是 data: continue if not line.startswith("data: "): continue data = line[6:] # current_event 默认为 token current_event = locals().get("current_event", "token") if current_event == "meta": try: meta_obj = json.loads(data) pretty_meta = json.dumps(meta_obj, ensure_ascii=False) print("Meta:", pretty_meta) except json.JSONDecodeError: print("Meta:", data) meta_printed = True print("\n--- AI 回复 ---\n", end="") continue if current_event == "done": print("\n✅ Stream done") break # 其他 token 输出 print(data, end="", flush=True) except ChunkedEncodingError: # 服务器已发送完毕但提前关闭连接，可忽略 print("\n⚠️ 连接提前关闭，已接收全部内容。") finally: try: resp.close() except Exception: pass if __name__ == "__main__": # test_diet_analysis() # test_diet_plan() # test_effects_inquiry() # test_avoid_inquiry() test_stream() text_match.py# -*- coding: utf-8 -*- # text_match.py import jieba import difflib from typing import Iterable, List, Set, Tuple def extract_food( user_text: str, food_vocab: Iterable[str], fuzzy_threshold: float = 0.8, ) -> Tuple[List[str], List[str], List[str]]: """根据用户输入提取食物关键词。 1. 先使用 jieba 分词命中精确词。 2. 对分词结果做模糊匹配，解决同义词/花式写法。 Args: user_text: 用户原始输入字符串。 food_vocab: 食物静态数据库的键集合。 fuzzy_threshold: SequenceMatcher 相似度阈值 (0-1)。 Returns: 去重后的食物列表，按出现顺序返回。 """ vocab_set: Set[str] = set(food_vocab) tokens = jieba.lcut(user_text, cut_all=False) hits: List[str] = [] exact_hits: List[str] = [] fuzzy_hits: List[str] = [] added: Set[str] = set() for token in tokens: # 精确匹配 if token in vocab_set and token not in added: hits.append(token) exact_hits.append(token) added.add(token) continue # 模糊匹配 for food in vocab_set: if food in added: continue if difflib.SequenceMatcher(None, token, food).ratio() >= fuzzy_threshold: hits.append(food) fuzzy_hits.append(food) added.add(food) break return hits, exact_hits, fuzzy_hits # 向后兼容的简化接口 def extract_food_simple(user_text: str, food_vocab: Iterable[str], fuzzy_threshold: float = 0.8) -> List[str]: """返回仅 food list 的旧版接口，供其他模块调用。""" return extract_food(user_text, food_vocab, fuzzy_threshold)[0] food_tags.json{ "炸鸡": { "tags": ["高油脂", "高热量"], "effects": "经常食用可能导致能量过剩、血脂升高", "avoid_with": ["啤酒"], "diet_type": ["增重期", "周末放纵"], "recipe_hint": "可使用空气炸锅，减少50%油脂摄入" }, "奶茶": { "tags": ["高糖", "高热量"], "effects": "过量摄入易导致胰岛素抵抗、体重增加", "avoid_with": [], "diet_type": ["偶尔犒劳"], "recipe_hint": "选择低糖或无糖版本，减少珍珠" }, "可乐": { "tags": ["高糖"], "effects": "长期高糖饮料摄入可增加蛀牙及肥胖风险", "avoid_with": ["咖啡"], "diet_type": ["偶尔犒劳"], "recipe_hint": "选择零度可乐或气泡水替代" }, "卤牛肉": { "tags": ["高钠", "高蛋白"], "effects": "高钠摄入可能升高血压", "avoid_with": [], "diet_type": ["增肌期"], "recipe_hint": "配合蔬菜食用平衡营养" }, "牛肉火锅": { "tags": ["高油脂", "高钠"], "effects": "高盐高油易导致水肿", "avoid_with": ["冰啤酒"], "diet_type": ["增重期"], "recipe_hint": "控制汤底油脂和盐分，搭配蔬菜" }, "蔬菜汤": { "tags": ["低热量", "健康推荐"], "effects": "一般安全", "avoid_with": [], "diet_type": ["减脂期", "日常维稳"], "recipe_hint": "少盐少油" }, "白灼虾": { "tags": ["高蛋白", "健康推荐"], "effects": "嘌呤偏高，痛风患者需控制", "avoid_with": ["维生素C高的水果"], "diet_type": ["增肌期", "减脂期"], "recipe_hint": "控制蘸料用量" }, "香菇": { "tags": ["高纤维", "低脂"], "effects": "过量可能导致胀气", "avoid_with": ["寒性食物"], "diet_type": ["减脂期", "日常维稳"], "recipe_hint": "烹饪前泡发充分" }, "燕麦": { "tags": ["高纤维", "低GI"], "effects": "摄入过多可能引起腹胀", "avoid_with": [], "diet_type": ["减脂期", "增肌期"], "recipe_hint": "配合蛋白质食物提高饱腹" }, "糙米": { "tags": ["高纤维", "低GI"], "effects": "富含植酸，影响矿物质吸收，建议与其他谷物搭配", "avoid_with": [], "diet_type": ["减脂期", "日常维稳"], "recipe_hint": "提前浸泡12小时再煮" }, "炸薯条": { "tags": ["高油脂", "高热量"], "effects": "反式脂肪酸摄入风险", "avoid_with": ["汽水"], "diet_type": ["偶尔犒劳"], "recipe_hint": "可使用空气炸锅替代油炸" } , "牛油果": { "tags": ["高脂肪", "高纤维", "健康推荐"], "effects": "脂肪含量高，减脂期注意总热量", "avoid_with": [], "diet_type": ["增肌期", "日常维稳"], "recipe_hint": "搭配全麦面包或沙拉" }, "酸奶": { "tags": ["高蛋白", "益生菌"], "effects": "乳糖不耐人群可能腹胀", "avoid_with": ["高糖谷物"], "diet_type": ["减脂期", "增肌期", "日常维稳"], "recipe_hint": "选择无糖或低糖版本" }, "披萨": { "tags": ["高油脂", "高热量"], "effects": "高钠高脂，注意摄入频次", "avoid_with": ["可乐"], "diet_type": ["周末放纵"], "recipe_hint": "选择薄底、加多蔬菜版本" }, "红薯": { "tags": ["低GI", "高纤维", "健康推荐"], "effects": "过量可致腹胀", "avoid_with": [], "diet_type": ["减脂期", "日常维稳"], "recipe_hint": "蒸煮可保留营养" }, "鲑鱼": { "tags": ["高蛋白", "Omega-3"], "effects": "富含EPA/DHA，有助心血管健康", "avoid_with": ["富含草酸蔬菜"], "diet_type": ["增肌期", "日常维稳"], "recipe_hint": "推荐清蒸或空气炸锅" }, "绿茶": { "tags": ["低热量", "抗氧化"], "effects": "空腹或过量可致胃刺激", "avoid_with": ["牛奶"], "diet_type": ["减脂期", "日常维稳"], "recipe_hint": "餐后一小时饮用更佳" }, "巧克力": { "tags": ["高糖", "高脂"], "effects": "摄入过多易长痘和增重", "avoid_with": [], "diet_type": ["偶尔犒劳"], "recipe_hint": "优选 70% 以上黑巧" }, "能量饮料": { "tags": ["高糖", "咖啡因"], "effects": "过量可能导致心悸、睡眠障碍", "avoid_with": ["咖啡"], "diet_type": ["比赛备战", "偶尔提神"], "recipe_hint": "控制每日总咖啡因 < 400mg" } } config.json{ "DEEPSEEK_API_KEY": "刚刚申请的Key", "DEEPSEEK_URL": "申请得到的Url", "MCP_PORT": 8001, "MODEL_NAME": "DeepSeek-R1" } requirements.txtfastapi uvicorn requests httpx jieba cachetools loguru4.1.4 安装依赖点击 终端pip install -r requirements.txt -i https://mirrors.huaweicloud.com/repository/pypi/simple安装成功 示例图 如果显示仍然缺少依赖 请关闭ide 重新打开即可4.2 调试 MCP终端上执行命令uvicorn mcp_server:app --port 8001 运行成功示例图下一步 新建终端 点加号执行 测试程序python test_pipeline.py执行测试成功示例图 这里我们就可以看到 流式响应的一个效果然后我们最小化 这个 CodeArts IDE for Python4.3 仓颉交互AIChat(魔改MCP)打开 仓颉编辑器4.3.1 新建项目我们这里名称就叫 food_bot 然后点击 创建4.3.2 代码实施config.json{ "model": "", "api_key": "", "base_url": "http://localhost:8001/mcp_stream", "system_prompt": "" } cjpm.toml[dependencies] [package] cjc-version = "0.53.13" compile-option = "" description = "nothing here" link-option = "" name = "food_bot" output-type = "executable" src-dir = "" target-dir = "" version = "1.0.0" package-configuration = {} src/main.cjpackage food_bot import std.console.Console import std.collection.ArrayList func cli_chat(env_info: EnvInfo, stream!: Bool) { println("\n欢迎使用智能膳食分析助手，输入exit退出") var history: ArrayList<(String, String)> = ArrayList<(String, String)>() while (true) { print("Input: ") var prompt: String = "" match(Console.stdIn.readln()) { case Some(str1: String) => prompt=str1 case None => continue } if (prompt == "exit" || prompt == "exit()") { break } if (prompt == "clear") { history.clear() println("Output: 已清理历史对话信息。") continue } // 根据配置或命令行参数判断是否启用流式模式 var use_stream: Bool = stream if (!use_stream && env_info.base_url.endsWith("/mcp_stream")) { use_stream = true } print("ChatBox: ") let response_option: Option<String> = if (use_stream) { stream_chat(prompt, env_info, history) } else { chat(prompt, env_info, history) } match (response_option) { case Some(response: String) => println("${response}") history.append((prompt, response)) case None => println("发生错误，正在重试一次...") let retry_option = if (use_stream) { stream_chat(prompt, env_info, history) } else { chat(prompt, env_info, history) } match(retry_option) { case Some(resp2: String) => println("${resp2}") history.append((prompt, resp2)) case None => println("重试失败，即将退出") break } } } } main(args: Array<String>): Int64 { let env_info = load_env_info() var stream: Bool = false if (args.size == 1) { if (args[0] == "--stream") { stream = true } else { println("参数无效，仅支持 --stream 以开启流式输出") } } cli_chat(env_info, stream: stream) return 0 } env_info.cjpackage food_bot import encoding.json.stream.* import std.fs.File import std.fs.Path import std.io.ByteArrayStream public class EnvInfo <: JsonDeserializable<EnvInfo> & JsonSerializable { public let model: String // 模型名称 public let api_key: String // api密钥 public let base_url: String // 调用接口路径 public let system_prompt: String // 预置系统提示词 public init(model: String, api_key: String, base_url: String, system_prompt: String) { this.model = model this.api_key = api_key this.base_url = base_url this.system_prompt = system_prompt } public static func fromJson(r: JsonReader): EnvInfo { var temp_model: String = "" var temp_api_key: String = "sk-xxx" var temp_base_url: String = "http://xxx.xxx.xxx/v1" var temp_system_prompt: String = "You are a helpful assistant." while (let Some(v) <- r.peek()) { match(v) { case BeginObject => r.startObject() while(r.peek() != EndObject) { let n = r.readName() match (n) { case "model" => temp_model = r.readValue<String>() case "api_key" => temp_api_key = r.readValue<String>() case "base_url" => temp_base_url = r.readValue<String>() case "system_prompt" => temp_system_prompt = r.readValue<String>() case _ => () } } r.endObject() break case _ => throw Exception() } } return EnvInfo(temp_model, temp_api_key, temp_base_url, temp_system_prompt) } public func toJson(w: JsonWriter): Unit { w.startObject() w.writeName("model").writeValue(this.model) w.writeName("api_key").writeValue(this.api_key) w.writeName("base_url").writeValue(this.base_url) w.writeName("system_prompt").writeValue(this.system_prompt) w.endObject() w.flush() } } public func save_env_info(): Unit { // 该函数用于测试EnvInfo类的序列化为json的能力，顺便生成一个env_sample.json样本做为参考 let env_path = Path("env_sample.json") if (File.exists(env_path)) { File.delete(env_path) } let file = File.create(env_path) let env_info = EnvInfo( "xxxx", "sk-xxxxxx", "http://xxx.xxx.xxx/v1/chat/completions", "You are a helpful assistant." ) var byte_stream = ByteArrayStream() var json_writer = JsonWriter(byte_stream) let write_config = WriteConfig.pretty json_writer.writeConfig = write_config env_info.toJson(json_writer) file.write(byte_stream.readToEnd()) println("`env_sample.json` save ok") file.close() } public func load_env_info(): EnvInfo { // 用于加载配置文件 let env_path = Path("config.json") if (!File.exists(env_path)) { throw Exception("The config file not exists, please check again") } let file = File.openRead(env_path) let file_str: Array<UInt8> = file.readToEnd() var byte_stream = ByteArrayStream() byte_stream.write(file_str) let json_reader = JsonReader(byte_stream) let env_info: EnvInfo = EnvInfo.fromJson(json_reader) file.close() // println("model: ${env_info.model}") // println("api_key: ${env_info.api_key}") // println("base_url: ${env_info.base_url}") // println("system_prompt: ${env_info.system_prompt}") return env_info } chat.cjpackage food_bot import encoding.json.stream.* import net.http.ClientBuilder import net.http.HttpHeaders import net.http.HttpRequestBuilder import net.tls.TlsClientConfig import net.tls.CertificateVerifyMode import std.collection.ArrayList import std.io.ByteArrayStream import std.time.Duration import std.unicode.UnicodeExtension // for String.trim() // ===== 可配置常量 ===== public let READ_TIMEOUT_SECONDS: Int64 = 300 // 长轮询 SSE 建议 300 秒 // =========================== public enum RoleType { User | Assistant | System } public func role_type_to_str(role: RoleType): Option<String> { return match(role) { case RoleType.User => Some("user") case RoleType.Assistant => Some("assistant") case RoleType.System => Some("system") } } public func str_to_role_type(role_option_str: Option<String>): RoleType { return match(role_option_str) { case Some(str) => match(str) { case "user" => RoleType.User case "assistant" => RoleType.Assistant case "system" => RoleType.System case _ => RoleType.Assistant } case None => RoleType.Assistant } } public struct Message<: JsonDeserializable<Message> & JsonSerializable { public let role: RoleType public var content: String public init(role: RoleType, content: String) { this.role = role this.content = content } public static func fromJson(r: JsonReader): Message { var temp_role: Option<String> = None var temp_content: String = "" while (let Some(v) <- r.peek()) { match(v) { case BeginObject => r.startObject() while(r.peek() != EndObject) { let n = r.readName() match(n) { case "role" => temp_role = r.readValue<Option<String>>() case "content" => temp_content = r.readValue<String>() case _ => r.skip() } } r.endObject() break case _ => throw Exception("can't deserialize for Message") } } let role_type: RoleType = str_to_role_type(temp_role) return Message(role_type, temp_content) } public func toJson(w: JsonWriter) { w.startObject() w.writeName("role").writeValue<Option<String>>(role_type_to_str(this.role)) w.writeName("content").writeValue<String>(this.content) w.endObject() w.flush() } } public struct ChatRequest <: JsonSerializable { private let model: String private let messages: ArrayList<Message> private let max_tokens: Int64 private let stream: Bool public init( model: String, messages: ArrayList<Message>, max_tokens: Int64, stream: Bool ) { // construction function with messages this.model = model this.messages = messages this.max_tokens = max_tokens this.stream = stream } public init( model: String, prompt: String, history: ArrayList<(String, String)>, system_prompt: String, max_tokens: Int64, stream: Bool ){ // construction function with prompt and system_prompt this.model = model this.messages = ArrayList<Message>([ Message(RoleType.System, system_prompt) ]) for ((use_msg, bot_msg) in history) { this.messages.append(Message(RoleType.User, use_msg)) this.messages.append(Message(RoleType.Assistant, bot_msg)) } this.messages.append(Message(RoleType.User, prompt)) this.max_tokens = max_tokens this.stream = stream } public init( model: String, prompt: String, history: ArrayList<(String, String)>, system_prompt: String, stream: Bool ){ // construction function with prompt and default arguments this.model = model this.messages = ArrayList<Message>([ Message(RoleType.System, system_prompt) ]) for ((use_msg, bot_msg) in history) { this.messages.append(Message(RoleType.User, use_msg)) this.messages.append(Message(RoleType.Assistant, bot_msg)) } this.messages.append(Message(RoleType.User, prompt)) this.max_tokens = 2000 this.stream = stream } public func toJson(w: JsonWriter) { w.startObject() w.writeName("model").writeValue<String>(this.model) w.writeName("messages").writeValue<ArrayList<Message>>(this.messages) w.writeName("max_tokens").writeValue<Int64>(this.max_tokens) w.writeName("stream").writeValue<Bool>(this.stream) w.endObject() w.flush() } } public struct Choice <: JsonDeserializable<Choice> & JsonSerializable { public let index: Int32 public let message: Option<Message> public let delta: Option<Message> public let finish_reason: Option<String> public let logprobs: Option<Float64> // dashscope for qwen need public init( index: Int32, message: Option<Message>, delta: Option<Message>, finish_reason: Option<String>, logprobs: Option<Float64> ) { this.index = index this.message = message this.delta = delta this.finish_reason = finish_reason this.logprobs = logprobs } public static func fromJson(r: JsonReader): Choice { var temp_index: Int32 = -1 var temp_message: Option<Message> = None var temp_delta: Option<Message> = None var temp_finish_reason: Option<String> = None var temp_logprobs: Option<Float64> = None while (let Some(v) <- r.peek()) { match(v) { case BeginObject => r.startObject() while(r.peek() != EndObject) { let n = r.readName() match (n) { case "index" => temp_index = r.readValue<Int32>() case "message" => temp_message = r.readValue<Option<Message>>() case "delta" => temp_delta = r.readValue<Option<Message>>() case "finish_reason" => temp_finish_reason = r.readValue<Option<String>>() case "logprobs" => temp_logprobs = r.readValue<Option<Float64>>() case _ => r.skip() } } r.endObject() break case _ => throw Exception("can't deserialize for Choice") } } return Choice(temp_index, temp_message, temp_delta, temp_finish_reason, temp_logprobs) } public func toJson(w: JsonWriter) { w.startObject() w.writeName("index").writeValue<Int32>(this.index) w.writeName("message").writeValue<Option<Message>>(this.message) w.writeName("delta").writeValue<Option<Message>>(this.delta) w.writeName("finish_reason").writeValue<Option<String>>(this.finish_reason) w.writeName("logprobs").writeValue<Option<Float64>>(this.logprobs) w.endObject() w.flush() } } public struct Usage <: JsonDeserializable<Usage> & JsonSerializable { public let prompt_tokens: UInt64 public let completion_tokens: UInt64 public let total_tokens: UInt64 public init(prompt_tokens: UInt64, completion_tokens: UInt64, total_tokens: UInt64) { this.prompt_tokens = prompt_tokens this.completion_tokens = completion_tokens this.total_tokens = total_tokens } public static func fromJson(r: JsonReader): Usage { var temp_prompt_tokens: UInt64 = 0 var temp_completion_tokens: UInt64 = 0 var temp_total_tokens: UInt64 = 0 while (let Some(v) <- r.peek()) { match(v) { case BeginObject => r.startObject() while(r.peek() != EndObject) { let n = r.readName() match (n) { case "prompt_tokens" => temp_prompt_tokens = r.readValue<UInt64>() case "completion_tokens" => temp_completion_tokens = r.readValue<UInt64>() case "total_tokens" => temp_total_tokens = r.readValue<UInt64>() case _ => r.skip() } } r.endObject() break case _ => throw Exception("can't deserialize for Usage") } } return Usage(temp_prompt_tokens, temp_completion_tokens, temp_total_tokens) } public func toJson(w: JsonWriter) { w.startObject() w.writeName("prompt_tokens").writeValue<UInt64>(this.prompt_tokens) w.writeName("completion_tokens").writeValue<UInt64>(this.completion_tokens) w.writeName("total_tokens").writeValue<UInt64>(this.total_tokens) w.endObject() w.flush() } } public struct ChatResponse <: JsonDeserializable<ChatResponse> { // some api names `id`, and some names `request_id` public let id: Option<String> public let request_id: Option<String> public let system_fingerprint: Option<String> public let model: String public let object: String public let created: UInt64 public let choices: ArrayList<Choice> public let usage: Option<Usage> public init( id: Option<String>, request_id: Option<String>, system_fingerprint: Option<String>, model: String, object: String, created: UInt64, choices: ArrayList<Choice>, usage: Option<Usage> ) { this.id = id this.request_id = request_id this.system_fingerprint = system_fingerprint this.model = model this.object = object this.created = created this.choices = choices this.usage = usage } public static func fromJson(r: JsonReader): ChatResponse { var temp_id: Option<String> = None var temp_request_id: Option<String> = None var temp_system_fingerprint: Option<String> = None var temp_model: String = "" var temp_object: String = "" var temp_created: UInt64 = 0 var temp_choices: ArrayList<Choice> = ArrayList<Choice>([]) var temp_usage: Option<Usage> = None while (let Some(v) <- r.peek()) { match(v) { case BeginObject => r.startObject() while(r.peek() != EndObject) { let n = r.readName() match (n) { case "id" => temp_id = r.readValue<Option<String>>() case "request_id" => temp_request_id = r.readValue<Option<String>>() case "system_fingerprint" => temp_system_fingerprint = r.readValue<Option<String>>() case "model" => temp_model = r.readValue<String>() case "object" => temp_object = r.readValue<String>() case "created" => temp_created = r.readValue<UInt64>() case "choices" => temp_choices = r.readValue<ArrayList<Choice>>() case "usage" => temp_usage = r.readValue<Option<Usage>>() case _ => r.skip() } } r.endObject() break case _ => throw Exception("can't deserialize for ChatResponse") } } return ChatResponse( temp_id, temp_request_id, temp_system_fingerprint, temp_model, temp_object, temp_created, temp_choices, temp_usage ) } } public func get_domain( url: String ): String { var temp_url = url if (temp_url.startsWith("https://")) { temp_url = temp_url["https://".size..] } else if (temp_url.startsWith("http://")) { temp_url = temp_url["http://".size..] } let domain: String = temp_url.split("?")[0].split("/")[0] return domain } public func build_http_client( prompt: String, env_info: EnvInfo, history: ArrayList<(String, String)>, stream!: Bool ){ // prepare input data // If we are targeting the custom `/mcp_stream` backend we must send // a very simple JSON body `{ "input": "<prompt>" }` instead of the // OpenAI-style `ChatRequest`. Detect this via URL suffix. let is_mcp_stream = env_info.base_url.endsWith("/mcp_stream") var post_data: Array<UInt8> if (is_mcp_stream) { var local_stream = ByteArrayStream() let local_writer = JsonWriter(local_stream) // { "input": "..." } local_writer.startObject() local_writer.writeName("input").writeValue(prompt) local_writer.endObject() local_writer.flush() post_data = local_stream.readToEnd() } else { var array_stream = ByteArrayStream() let json_writer = JsonWriter(array_stream) let chat_res = ChatRequest( env_info.model, prompt, history, env_info.system_prompt, stream ) chat_res.toJson(json_writer) post_data = array_stream.readToEnd() } // build headers var headers: HttpHeaders = HttpHeaders() if (!is_mcp_stream) { // local backend doesn't require auth header headers.add("Authorization", "Bearer ${env_info.api_key}") } headers.add("Content-Type", "application/json") if (stream) { headers.add("Accept", "text/event-stream") } let request = HttpRequestBuilder() .url(env_info.base_url) .method("POST") .body(post_data) .readTimeout(Duration.second * READ_TIMEOUT_SECONDS) .addHeaders(headers) .build() let client = if (env_info.base_url.startsWith("https")) { var tls_client_config = TlsClientConfig() tls_client_config.verifyMode = CertificateVerifyMode.TrustAll tls_client_config.domain = get_domain(env_info.base_url) ClientBuilder() .tlsConfig(tls_client_config) .build() } else { ClientBuilder().build() } return (request, client) } public func chat( prompt: String, env_info: EnvInfo, history: ArrayList<(String, String)> ): Option<String> { let (request, client) = build_http_client( prompt, env_info, history, stream: false ) var result_message: Option<String> = None var res_text = "" try { // call api let response = client.send( request ) // read result (support max revice 100k data) let buffer = Array<Byte>(102400, item: 0) let length = response.body.read(buffer) res_text = String.fromUtf8(buffer[..length]) // println("res_text: ${res_text}") var input_stream = ByteArrayStream() input_stream.write(res_text.toArray()) // convert text to ChatResponse object let json_reader = JsonReader(input_stream) let res_object = ChatResponse.fromJson(json_reader) let choices: ArrayList<Choice> = res_object.choices if (choices.size > 0) { let message = choices[0].message.getOrThrow() // println("message: ${message.content}") result_message = Some(message.content) } else { println("can't found any response") } } catch (e: Exception) { println("ERROR: ${e.message}, reviced text is ${res_text}") } client.close() return result_message } public func stream_chat( prompt: String, env_info: EnvInfo, history: ArrayList<(String, String)> ): Option<String> { let (request, client) = build_http_client( prompt, env_info, history, stream: true ) let is_mcp_stream = env_info.base_url.endsWith("/mcp_stream") var result_response: String = "" var temp_text2 = "" try { // call api let response = client.send( request ) // read result let buffer = Array<Byte>(10240, item: 0) if (is_mcp_stream) { var done = false var current_event = "" while(!done) { let length = response.body.read(buffer) if (length == 0) { break } let res_text = String.fromUtf8(buffer[..length]) for (line in res_text.split("\n")) { let trimmed = line.trim() if (trimmed.size == 0) { continue } if (trimmed.startsWith("event: ")) { current_event = trimmed["event: ".size..] continue } if (trimmed.startsWith("data: ")) { let data = trimmed["data: ".size..] if (current_event == "token") { // 累积 token，暂不直接打印 result_response = result_response + data } else if (current_event == "ping") { // 服务器心跳，忽略即可 () } else if (current_event == "error") { println("服务器错误: ${data}") done = true result_response = "" // force empty so caller sees None break } else if (current_event == "done") { done = true break } } } if (done) { break } } } else { var finish_reason: Option<String> = None while(finish_reason.isNone() && temp_text2 != "[DONE]") { let length = response.body.read(buffer) let res_text = String.fromUtf8(buffer[..length]) for (temp_text in res_text.split("\n")) { temp_text2 = if (temp_text.startsWith("data: ")) { temp_text["data: ".size..] } else { temp_text } if (temp_text2.size == 0) { continue } if (temp_text2 == "[DONE]") { break } var input_stream = ByteArrayStream() input_stream.write(temp_text2.toArray()) // convert text to ChatResponse object let json_reader = JsonReader(input_stream) let res_object = ChatResponse.fromJson(json_reader) let choices: ArrayList<Choice> = res_object.choices if (choices.size > 0) { finish_reason = choices[0].finish_reason if (finish_reason.isNone()) { let delta = choices[0].delta.getOrThrow() result_response = result_response + delta.content } } else { println("can't found any response") } } } } } catch (e: Exception) { println("ERROR: ${e.message}, reviced text is ${temp_text2}") } client.close() if (result_response.size > 0) { return Some(result_response) } else { return None } } 4.4 调试AIChat点击 终端执行 命令cjpm run运行成功示例图输入测试语句例如：我中午吃了炸鸡和奶茶还有香菇 我晚上吃点什么比较好呢？切回 CodeArts IDE for Python 切换 终端我们就可以看到 MCP 服务端 接受到请求的日志然后我们切回 仓颉编辑器 查看 这是响应正常 那么我们来测试 缓存 的效果 再次发出这个命令我中午吃了炸鸡和奶茶还有香菇 我晚上吃点什么比较好呢？你会看到它 立刻返回了结果然后我们可以 切换到 Python编辑器 看一下 命中缓存的日志💡这是两次请求1️⃣请求未命中则去请求DeepSeek2️⃣请求相同数据 则命中缓存五、项目总结5.1 MCPMCP，全称 Model Context Protocol，中文叫“模型上下文协议”。你可以把它想象成 AI 的“USB 接口” --让不同的 AI 模型、工具和应用程序能用统一的方式交流。那么我的理解是：它更像是一个适配器来调节各种AI不同的接口 达到一致的效果，让AI的交流更加简单，即使没有身份预设，走MCP是完美的让AI成为你的最佳助手。5.2 项目tag本次案例参考：https://devstation.connect.huaweicloud.com/space/devportal/casecenter/9a5b50615ac44a21af30f8699c81efc6/1https://devstation.connect.huaweicloud.com/space/devportal/casecenter/5a7144a66d1b43e088592ae447d518ed/1https://gitcode.com/CaseDeveloper/Cangjie-Examples简单的实现了 MCP并发(案例教程没有提及 uvicorn mcp_server:app --port 8001 --workers 4 )后续可拓展例如：food_bot 🔜 vue等各种UI 前端 各种适配机器人food_mcp🔜 独立开发为自己更加需要 模式更加契合的"USB" 让AI更加懂你关于缓存方面：缓存可以采用本地文件、redis等更加高性能的方案来替换这样就可以让实现由AI回答 到AI理解到了所表达和呈现的每一个需求点六、活动地址​ 如果对该项目有疑问，可以评论区留言交流~如果喜欢可以点点赞~​ 我正在参加【案例共创】第4期 基于华为开发者空间+仓颉/DeepSeek/MCP完成应用构建开发实践 https://bbs.huaweicloud.com/forum/thread-02127182415062274055-1-1.html

随着Ai编码的普及化，大众化，多起代码版权诉讼引发争议。当AI生成的代码片段可能包含受版权保护的代码时，开发者该如何规避法律风险？欢迎大家讨论

前言不久前在仓颉的交流群中看到了关于仓颉案例共创的活动消息，于是想到可以用仓颉结合自身所长开发一个AI桌面宠物供大家娱乐学习。正好笔者封装的仓颉桌面渲染库 RGF_CJ 和控件库 RGF_UI 能满足这方面的需求，那么说干就干！成品效果开发前的准备在正式进行应用的开发之前，需要先准备好相关的依赖。领取华为云DeepSeek免费资源华为云为开发者提供了单模型200万Tokens的免费额度，包含DeepSeek-R1&V3满血版，登录华为云ModelArts Studio(MaaS)控制台即可领取，此处可以选择DeepSeek-R1满血版来搭建AI聊天桌宠。提示：点击此链接领取：https://www.huaweicloud.com/product/modelarts/studio.html进入网站签署服务声明领取免费额度并查看调用说明通过说明中的链接去创建API点击按钮创建API填写API相关信息获取API并保存提示：请保存好此处的API链接，后续代码开发中会用到。获取RGF依赖文件在后续应用的开发中会依赖到RGF_UI库，此库依赖到一些文件，可以从RGF_UI的库中获取。这些依赖文件后续会链接入应用中去。提示：点击此链接下载：https://gitcode.com/raozj/RGF_UI/tree/master/libs进入RGF_UI库的Libs目录，下载依赖文件保存依赖文件，留待后续使用安装仓颉STDX及相关依赖因为是基于仓颉开发通信相关的应用，因此需要依赖到STDX中的stdx.net.http、stdx.encoding.json、stdx.net.tls等包，而这些包自Cangjie v0.60.5版本开始从std标准库中移出，因此需要自行安装这些包以及这些包所依赖的OpenSSL。提示：点击此链接下载：https://gitcode.com/Cangjie/Cangjie-STDX注’ 由于笔者发文时Cangjie v0.60.5版本尚且处于内测阶段，因此若开发者无内测资格则无法下载到此版本的相关文件。关于 STDX 和 OpenSSL 的安装请参考上述链接中官方的文档，此处不做赘述。序列帧合图资源制作作为一个桌面宠物，自然需要拥有对应的动画，对此可以使用DragonBones、Spine或者Adobe Effect等软件制作。制作动画笔者在此使用的是DragonBones，关于DragonBones的安装和使用不在本文的探讨范畴之内，读者可自行选择喜欢的动画制作软件，设计并制作完成动画后输出序列帧以用于应用的开发。设计动画制作出序列帧制作合图资源如果每帧的图片都单独存储，对于应用的初始化加载和渲染都是一个较大的负担，因此建议将一类的序列帧动画按照相等宽高、间距合并为一张图片。由于手动合图比较麻烦，因此笔者写了一个Python脚本，快速的将目录中的序列帧图片按照指定的宽度分布数量合并为一张图片。注’ 由于此部分不属于文章核心部分，因此直接贴出代码以供参考，读者可通过其它工具或代码实现自己个性化的合图方案。import os import cv2 import numpy as np def load_images_from_folder(folder): """ 加载指定文件夹下的所有 .png 图像 """ image_files = [f for f in os.listdir(folder) if f.lower().endswith('.png')] image_files.sort() # 按文件名排序 images = [] for filename in image_files: file_path = os.path.join(folder, filename) # 读取图像（支持中文路径） with open(file_path, 'rb') as f: img_data = np.frombuffer(f.read(), dtype=np.uint8) img = cv2.imdecode(img_data, cv2.IMREAD_UNCHANGED) images.append(img) print(f"已加载: {filename}") return images def stitch_images(images, cols, bg_color=(0, 0, 0, 0)): """ 将图像按指定列数拼接成一张大图 :param images: 图像列表 :param cols: 每行显示的图像数量 :param bg_color: 背景颜色（支持 RGBA） :return: 拼接后的图像 """ if not images: raise ValueError("没有图像可供拼接") h, w = images[0].shape[:2] rows = (len(images) + cols - 1) // cols # 创建空白背景图 channels = images[0].shape[2] if len(images[0].shape) == 3 else 1 if channels == 1: bg_color = (bg_color[0], ) # 灰度图 elif channels == 3: bg_color = bg_color[:3] # RGB elif channels == 4: bg_color = bg_color[:4] # RGBA canvas = np.full((h * rows, w * cols, channels), bg_color, dtype=np.uint8) for idx, img in enumerate(images): row = idx // cols col = idx % cols x = col * w y = row * h canvas[y:y+h, x:x+w] = img return canvas def main(): folder = input("请输入包含PNG图像的文件夹路径：").strip() if not os.path.isdir(folder): print("输入的不是有效文件夹路径！") return try: cols = int(input("请输入每行显示的图像数量：")) except ValueError: print("请输入有效的数字！") return output_path = input("请输入输出图像的完整路径（如 output.png）：").strip() images = load_images_from_folder(folder) if not images: print("未找到任何PNG图像！") return # 检查所有图像是否尺寸一致 first_h, first_w = images[0].shape[:2] for img in images[1:]: h, w = img.shape[:2] if h != first_h or w != first_w: print("警告：图像尺寸不一致，可能影响拼接效果。") # 拼接图像 print("开始拼接图像...") stitched_image = stitch_images(images, cols) # 保存结果 cv2.imwrite(output_path, stitched_image) print(f"拼接完成，已保存为: {output_path}") if __name__ == '__main__': main() 开发AI桌宠基本规划首先对项目目录和文件进行基本的规划，后续的开发就基于规划进行。目录规划Yez/ -------------------------------------------# 项目主目录 ├─libs/ ---------------------------------------# 依赖库 ├─res/ ----------------------------------------# 资源 │ └─SequenceFrame/ ---------------------------# 序列帧资源 │ └─skip_white/ ---------------------------# 椰椰的白色皮肤资源 ├─src/ ----------------------------------------# 源代码目录 └─tool/ ---------------------------------------# 工具目录文件规划Yez/ ├┈cjpm.lock ├┈cjpm.toml ├─libs/ │ ├┈libimm32.a -------------------------------# Windows 输入法库 │ ├┈resources.rc.o ---------------------------# 资源配置清单 内部声明应用依赖的最低系统版本 │ └┈libRgf.dll -------------------------------# RGF渲染库依赖的动态库 ├─res/ │ ├┈config.json ------------------------------# 椰椰的配置文件 │ └─SequenceFrame/ │ └─skip_white/ │ ├┈blink.json -------------------------# 眨眼合图描述文件 │ ├┈blink.png --------------------------# 眨眼动画合图 │ ├┈jump.json --------------------------# 跳跃合图描述文件 │ ├┈jump.png ---------------------------# 跳跃动画合图 │ ├┈speak.json -------------------------# 说话合图描述文件 │ ├┈speak.png --------------------------# 说话动画合图 │ ├┈wait.json --------------------------# 待机合图描述文件 │ └┈wait.png ---------------------------# 待机动画合图 ├─src/ │ ├┈chat.cj ----------------------------------# 大模型网络通信代码 │ ├┈config.cj --------------------------------# 配置管理代码 │ ├┈main.cj ----------------------------------# 程序主入口代码 │ ├┈res.cj -----------------------------------# 资源管理代码 │ ├┈win_anm.cj -------------------------------# 动画绘制窗口代码 │ ├┈win_input.cj -----------------------------# 输入框窗口代码 │ └┈win_text.cj ------------------------------# 鼠标文本渲染代码 └─tool/ └┈picture_puzzle.py ------------------------# python 合图脚本代码链接文件与库配置在开发之前，需要现为当前项目链接依赖文件和添加依赖库。链接文件在之前的小节中，下载了RGF_UI的依赖文件压缩包，在下载的压缩包内应该包含三个文件，即libRgf.dll、resources.rc.o和libimm32.a。这三个文件中，libRgf.dll是RGF_CJ库的依赖文件，必须随应用打包发布；resources.rc.o文件是Windows资源描述文件，读者也可以自行编写resources.rc文件并生成此文件（需设置最低支持Windows 8，即内核NT 6.2）；libimm32.a文件是自绘编辑框控件依赖的文件，主要提供输入法相关的操作能力。在仓颉包管理配置toml文件中，需要加入链接上述文件的代码:[target] [target.x86_64-w64-mingw32] link-option = "-L ./libs -l:libRgf.dll -l user32 -l imm32" 库配置完成链接后，还需要为应用声明依赖RGF_CJ和RGF_UI包，并且为RGF_CJ包配置宏。提示：点击此链接查看依赖库RGF_CJ：https://gitcode.com/Cangjie-SIG/RGF_CJ提示：点击此链接查看依赖库RGF_UI：https://gitcode.com/raozj/RGF_UI[dependencies] [dependencies.rgf] git = "https://gitcode.com/Cangjie-SIG/RGF_CJ.git" [dependencies.rgfui] git = "https://gitcode.com/raozj/RGF_UI.git" [profile] [profile.customized-option] rgfCfg = "--cfg \"RGF_LANG=zh-cn\"" 上述代码中配置的“RGF_LANG=zh-cn”表示：当RGF_CJ出现异常时，捕获输出的异常提示内容为中文。配置加载模块笔者希望桌宠能够具备记忆配置的能力，因此需要为桌宠写一个配置的加载、保存模块，这个模块会记住桌宠所使用的皮肤、最后所在的桌面位置等信息，以确保桌宠重启后依然符合之前的情况。编写一个配置管理相关的恶汉单例类ConfigMgr，单例具备两个方法，一个用于从文件中加载配置，一个用于写出配置到文件中去。（其实也可以直接编写静态方法或直接编写函数而不使用类，看大家自己的选择吧。）代码如下：// Yez/src/config.cj // 配置管理类 class ConfigMgr { private static let _instance:ConfigMgr = ConfigMgr() public static prop i:ConfigMgr{ get(){ _instance } } private init() {} ... /** * 加载配置文件 * @return 是否成功 */ public func readConfig():Bool{ ... } /** 写出配置文件 */ public func writeConfig():Unit{ ... } ... } 编写一个ConfigInfo类，这个类用于存放实际的配置项，并且提供转换为Json的能力，代码如下：// Yez/src/config.cj class ConfigInfo <: Serializable<ConfigInfo> { // 椰椰右下角窗口坐标系的X坐标 var _x:Int32 = Int32.Min // 椰椰右下角窗口坐标系的Y坐标 var _y:Int32 = Int32.Min // 椰椰使用的皮肤 var _skip = "skip_white" // 椰椰画面缩放比例 var _zoom:Float32 = 0.125 // 椰椰的字体大小 var _fontSize:Float32 = 16.1 public func serialize(): DataModel { return DataModelStruct().add(field<Int32>("x", _x)) .add(field<Int32>("y", _y)) .add(field<String>("skip", _skip)) .add(field<Float32>("zoom", _zoom)) .add(field<Float32>("fontSize", _fontSize)) } public static func deserialize(dm: DataModel): ConfigInfo { var dms = match (dm) { case data: DataModelStruct => data case _ => throw Exception("this data is not DataModelStruct") } var result = ConfigInfo() result._x = Int32.deserialize(dms.get("x")) result._y = Int32.deserialize(dms.get("y")) result._skip = String.deserialize(dms.get("skip")) result._zoom = Float32.deserialize(dms.get("zoom")) result._fontSize = Float32.deserialize(dms.get("fontSize")) return result } } 将ConfigInfo以成员变量的形式加入到ConfigMgr中，并且由ConfigMgr对外提供访问属性，代码如下：// Yez/src/config.cj // 配置管理类 class ConfigMgr { // 配置信息 private var configImpl:ConfigInfo = ConfigInfo() ... public mut prop x:Int32{ get(){ return configImpl._x } set(val){ configImpl._x = val } } public mut prop y:Int32{ get(){ return configImpl._y } set(val){ configImpl._y = val } } public mut prop skip:String{ get(){ return configImpl._skip } set(val){ configImpl._skip = val } } public prop zoom:Float32{ get(){ return configImpl._zoom } } public prop fontSize:Float32{ get(){ return configImpl._fontSize } } } 接下来只要完善readConfig和writeConfig的读取和写出逻辑即可，代码如下：// Yez/src/config.cj // 配置文件路径 const CONFIG_FILE_PATH = "./res/config.json" ... public func readConfig():Bool{ // 判断目录是否存在 if(exists(CONFIG_FILE_PATH)){ let seqFramePath = canonicalize(CONFIG_FILE_PATH) let bytes = File.readFrom(seqFramePath) var jv = JsonValue.fromStr(StringReader(ByteBuffer(bytes)).readToEnd()) var dm = DataModel.fromJson(jv) configImpl = ConfigInfo.deserialize(dm) } // 目前始终返回 true // 当配置不存在时，则使用默认配置 return true } public func writeConfig():Unit{ if(!exists(CONFIG_FILE_PATH)){ File.create(CONFIG_FILE_PATH) } let seqFramePath = canonicalize(CONFIG_FILE_PATH) var Anm = configImpl.serialize() var dm = Anm.toJson() File.writeTo(seqFramePath,dm.toString().toArray()) } 最后，在入口函数中启用配置加载与写出即可。// Yez/src/main.cj main(): Int64 { // 加载配置文件数据 if(!ConfigMgr.i.readConfig()){ return 0 } ... // 保存配置到文件中去 ConfigMgr.i.writeConfig() return 0 } 资源加载模块除了配置之外，应用资源也是一个重要的部分，这部分的完整性会影响到应用是否正常渲染。首先采用和Config部分类似的结构准备恶汉单例类ResMgr和动画信息类AnmInfo，代码如下：// Yez/src/res.cj // 资源管理类 class ResMgr { private static let _instance:ResMgr = ResMgr() public static prop i:ResMgr{ get(){ _instance } } private init() {} // 动画 private static const ANM_BLINK ="blink" private static const ANM_JUMP ="jump" private static const ANM_SPEAK ="speak" private static const ANM_WAIT ="wait" private let _skipImg:HashMap<String, Array<(AnmInfo,MemBitmap)>> = HashMap<String, Array<(AnmInfo,MemBitmap)>>() public func loadRes():Bool{ ... } public func unloadRes(){ ... } public prop skip:HashMap<String, Array<(AnmInfo,MemBitmap)>>{ get(){ return _skipImg } } } class AnmInfo <: Serializable<AnmInfo> { // 单帧图片宽度 var width:Int32 = 0 // 单帧图片高度 var height:Int32 = 0 // 合图图片每行图片数量 var hCount:Int32 = 0 // 总帧数 var frames:Int32 = 0 ... } 然后就是实现在loadRes方法中加载合图及合图相关配置文件，代码逻辑还就是：从"./res/SequenceFrame"目录开始确保目录存在，然后遍历SequenceFrame目录下的所有皮肤目录，并且将皮肤目录内的合图资源及对应的配置文件载入到成员中去。如果目录下的资源文件不完整，则不会加载对应的内容到成员中。// Yez/src/res.cj /** * 代码较为复杂且不是核心代码，读者了解功能即可。 * 若需要查看相关代码，可下载项目查看。 */ public func loadRes():Bool{ ... } public func unloadRes(){ //释放所有图像资源（设备无关资源） for(skipPath in _skipImg){ skipPath[1][0][1].release() skipPath[1][1][1].release() skipPath[1][2][1].release() skipPath[1][3][1].release() } } 最后，在入口函数中启用资源加载与释放即可。// Yez/src/main.cj main(): Int64 { ... // 加载依赖资源 if(ResMgr.i.loadRes()){ ... } // 释放一切可能被加载的资源 ResMgr.i.unloadRes() ... return 0 } 大模型模块接下来便是与AI通信的核心部分了，此部分需要完成与大模型的互通。我们需要准备一个LLM类，这个类负责基于API、URL等信息与大模型建立链接，并且解析信息。// Yez/src/chat.cj class LLM { let client: Client let history = StringBuilder() public LLM( // 大模型URL链接 let url!: String, // 大模型 API Key let key!: String, // 大模型类型标识 let model!: String, // 上下文 let context!: Bool = false ) { var config = TlsClientConfig() config.verifyMode = TrustAll client = ClientBuilder() .tlsConfig(config) // AI 服务响应有时候比较慢，这里设置为无限等待 .readTimeout(Duration.Max) .build() } /** * 编码函数，将角色与内容编码为字符串 * return 字符串 */ func encode(role: Role, content: String) { ... } /** * 发送函数，将对话数据发送到服务器 * return HttpResponse */ func send(input: String, stream!: Bool = false) { ... } /** * 分析函数，解析服务器返回的数据并且提取其中有效的文本内容 * return 有效内容 */ func parse(text: String, stream!: Bool = false) { ... } /** * 流式对话 * @param input 输入的文本 * @param task 流式返回内容的回调函数 * @param taskEnd 返回内容结束的回调函数 */ public func chat(input: String, task!: (String) -> Unit = {o => print(o)},taskEnd!:()->Unit = {=>}) { ... } /** * 非流式对话 */ public func chat(input!: String) { ... } /** * 历史记忆预置 */ public func preset(context: Array<(Role, String)>) { ... } /** * 历史记忆清除 */ public func reset() { ... } } 根据华为云提供的调用参考，完善 send 方法代码如下：// Yez/src/chat.cj func send(input: String, stream!: Bool = false) { let message = encode(I, input) let content = '{"model":"${model}","messages":[${history}${message}],"stream":${stream}}' if (context) { history.append(message) } let request = HttpRequestBuilder() .url(url) .header('Authorization', 'Bearer ${key}') .header('Content-Type', 'application/json') .header('Accept', if (stream) { 'text/event-stream' } else { 'application/json' }) .body(content) .post() .build() client.send(request) } 根据华为云返回的数据格式，完善 parse 方法代码如下：// Yez/src/chat.cj func parse(text: String, stream!: Bool = false) { let json = JsonValue.fromStr(text).asObject() let choices = json.getFields()['choices'].asArray() if(choices.size() == 0){ return "" } // 流式和非流式情况下，这个字段名称不同 let key = if (stream) { 'delta' } else { 'message' } let message = choices[0].asObject().getFields()[key].asObject() if(message.containsKey("content")){ let content = message.getFields()['content'].asString().getValue() // 移除开头的两个 \n if(content == "\n\n"){ return "" } return content }else{ return "" } } 最后实现核心的流式对话部分，代码如下：// Yez/src/chat.cj public func chat(input: String, task!: (String) -> Unit = {o => print(o)},taskEnd!:()->Unit = {=>}) { // 根据传入的字符串构建上发服务器的 HttpResponse let response = send(input, stream: true) // 准备一个字符串缓冲区 let output = StringBuilder() // 准备一个字节缓冲区 let buffer = Array<Byte>(1024 * 8, {i=>0}) // 获取返回的内容 var length = response.body.read(buffer) while (length != 0) { // 将字节缓冲区内的内容按照UTF-8编码解析为字符串 let text = String.fromUtf8(buffer[..length]) const INDEX = 6 // 按行分割 for (line in text.split('\n', removeEmpty: true)) { // 确保获取的是完整的：data: {...} if (line.size > INDEX && line[INDEX] == b'{' && line[line.size - 1] == b'}') { let json = line[INDEX..line.size] // 分析出有效信息 let slice = parse(json, stream: true) if (context) { output.append(slice) } // 若有效信息不为空，则通过回调函数返回 if(slice != ""){ task(slice) } } } length = response.body.read(buffer) } if (context) { history.append(',${encode(AI, output.toString())},') } // 所有处理结束后，执行结束回调 taskEnd() } LLM类是一个较为通用的类，但是使用较为繁琐，因此对于外部调用而言，只需要一个简化的实例即可。因此，可以再编写一个独立的懒汉单例LLMImpl类，并且仅提供一个对外实现流式对话的chat方法，代码如下：// Yez/src/chat.cj class LLMImpl { private static var _i: Option<LLMImpl> = None public static prop i:LLMImpl { get(){ if(_i.isNone()){ _i = LLMImpl() } return _i.getOrThrow() } } private let robot:LLM // 初始化一个对话模型 private init(){ robot = LLM( // 华为云大模型调用URL url: 'https://maas-cn-southwest-2.modelarts-maas.com/v1/infers/8a062fd4-7367-4ab4-a936-5eeb8fb821c4/v1/chat/completions', // 如果示例自带的密钥失效，请自行注册，https://www.huaweicloud.com/product/modelarts/studio.html key: '/* !!! 这里填写读者自己的 API Key !!! */', // 要使用的 大模型 model: 'DeepSeek-R1', context: true ); // 设定基本信息 robot.preset([(System, #"我会自称“椰椰”并亲昵的在三句话之内回复所有问题。对于限定长度内无法阐述清晰的问题，我会表现出呆傻可爱的效果并说出“宕机”。如果对我说“拍照”或“截图”，我就会说“咔嚓~”。如果问我是谁开发的，我会说出“尧佥&椰子”。"#)]) } /** * 流式对话 * @param content 输入的文本 * @param task 流式返回内容的回调函数 * @param end 返回内容结束的回调函数 */ public func chat(content:String, task!: (String) -> Unit = {o => print(o)},end!:()->Unit = {=>}){ robot.chat(content,task:task,taskEnd:end) } } 注’ 完整的代码请参考提供的代码文件桌宠渲染接下来，需要实现可视化的窗口渲染，赋予“椰椰”一副“身体”。动画播放窗口为了实现AI桌宠的渲染，首先需要有一个窗口。对此，笔者选择基于RGF_CJ中的WinBase类，实现一个自己的窗口类WinMain。// Yez/src/win_anm.cj // 使用 RGF_CJ 的宏实现代码简化 @LifeCycle // 自动管理 设备相关资源 的释放 @HookProc // 自动挂载窗口类的消息事件方法 class WinMain <: WinBase{ ... // 窗口类初始化时，传入 ResMgr 载入的资源（设备无关资源） init(anm:Array<(AnmInfo, MemBitmap)>){ _anmRes = anm } /** * 加载设备无关资源 -> 设备相关资源 * @note 本部分将资源ResMgr类中的图片资源，载入到窗口类中 */ private func loadAnm(anm:Array<(AnmInfo, MemBitmap)>){ surface .createBitmapFromMemory(_bmps[0],anm[0][1]) .createBitmapFromMemory(_bmps[1],anm[1][1]) .createBitmapFromMemory(_bmps[2],anm[2][1]) .createBitmapFromMemory(_bmps[3],anm[3][1]) } /** * 创建设备资源事件，窗口渲染表面创建完成后，开始载入资源 */ public func createDeviceResources():Bool{ // 这里调用 loadAnm 载入图片资源 loadAnm(_anmRes) return true; } /** * 窗口创建完成事件，当窗口创建完成时，开始渲染动画 */ public func created(hwnd:RgfHwnd,x:Int32, y:Int32, nWidth:Int32, nHeight:Int32):Unit{ ... } /** * 绘制窗口事件，此处根据当前的动画状态渲染窗口 */ public override func onPaint(wRect:Rect):Unit{ surface.clear(0.0,0.0,0.0,0.0) match(_nowState){ case Wait => {=> // 等待状态的动画渲染比较特殊，因为需要实现眼睛跟随鼠标移动 if(_bmps[3].isValid() && _bmps[0].isValid()){ // 这一部分计算当前渲染帧的静态部分，即：身体， // 并且将这部分内容渲染到窗口 .... // 然后下面计算出鼠标对眼睛造成的偏移量 // 并且将眼睛渲染到身体上 if(_anmStep < _anmRes[0][0].frames){ .... }else{ .... } } }() case Jump => {=> // 这里渲染椰椰的跳跃动画 if(_bmps[1].isValid()){ .... } }() case Speak => {=> // 这里渲染椰椰正在说话的动画 if(_bmps[2].isValid()){ .... } }() } } } 在入口函数中，启用UIMain并且创建窗口，代码如下：// Yez/src/win_anm.cj main(): Int64 { ... if(ResMgr.i.loadRes()){ // 开始创建UI UIMain{=> // 确保皮肤配置有效 if(!ResMgr.i.skip.contains(ConfigMgr.i.skip)){ ConfigMgr.i.skip = ResMgr.i.skip.keys().toArray()[0] } // 创建窗口类 let anmWin:WinMain = WinMain(ResMgr.i.skip.get(ConfigMgr.i.skip).getOrThrow()) // 先随便在桌面 (0,0) 的位置创建一个 10像素×10像素 的窗口 // 窗口类内部会在显示前根据配置中的信息自动布局到正确位置和尺寸 anmWin.createWin(uiGetContext(), "Yez", 0, 0, 10, 10, 0, WinStyle.e2n(WinStyle.Popup), // 使用的是 TopMost 分层窗口（支持透明背景） dwExStyle:WinStyleEx.Default | WinStyleEx.TopMost, // 渲染使用的是 Direct2D 计数 layered:true,substrate:Substrate.Direct2D ) } } ... } 完成上述代码后，窗口已经可以正确的渲染动画了注’ 完整的代码请参考提供的代码文件编辑框窗口对于编辑框的开发是比较繁琐的，因此选择直接继承RGF_UI的UIInput类，然后将此控件修改为窗口，首先需要创建一个编辑框窗口类InputWin，代码如下：// Yez/src/win_input.cj public open class InputWin <: UIInput{ // 为编辑框设置独立的UI样式，而不是继承全局的样式 let css:UITheme = UITheme() init(){ super(theme:css) // 从配置中获取字体大小 css.fontSizeM = ConfigMgr.i.fontSize // 更新样式配置 css.update() } /** * 获取输入框窗口预期出现的位置 * @return 编辑框窗口坐标 * @note 本部分计算编辑框在当前椰椰所在桌面中心时的位置 */ private func getTargetLocation():Point{ // 方案二、桌面中心 let monitor = rsGetMonitorFromPoint(rsGetCursorPos()) let pos = Point( ((monitor.right - monitor.left) - INPUT_WIN_WIDTH) / 2, ((monitor.bottom - monitor.top) - INPUT_WIN_HEIGHT) / 2 ) return pos } /** * 创建窗口 */ public func createWin(hwndParent:RgfHwnd):RgfHwnd{ // 获取出现位置坐标 let pos = getTargetLocation() // 创建一个置顶层的无边框编辑框窗口 let retv = super.createWin( uiGetContext(),"",pos.x,pos.y,INPUT_WIN_WIDTH,INPUT_WIN_HEIGHT,hwndParent, WinStyle.PopupWindow | WinStyle.ClipChildren, dwExStyle:WinStyleEx.Default | WinStyleEx.TopMost, substrate:Substrate.Direct2D,layered:false,bgMod:BackGroundMod.CoverNotDraw,drawInterval:0 ) // 设置默认的内容为：“想对椰椰说什么：” content = INPUT_PROMPT // 创建好就显示窗口 show(true) return retv } } 完成编辑框窗口的创建后，还需要能响应相关的输入事件。对此，笔者希望：a) 当编辑框被按下或输入内容时，如果内容为默认内容，则清空提示内容；b) 当编辑框失去焦点时，窗口主动销毁自身；c) 当编辑框按下回车键时，将编辑框内容上发给大模型，并开始处理返回消息。实现自动清空内容很简单，代码如下：// Yez/src/win_input.cj /** 鼠标左键按下时，判断内容并清空 */ public override open func onLButtonDown(e:EvMouse):Unit{ if(content == INPUT_PROMPT){ content = "" } super.onLButtonDown(e) } /** 键盘按下某键时，判断内容并清空 */ public override open func onKeyDown(e:EvKey):Unit{ if(content == INPUT_PROMPT){ content = "" } super.onKeyDown(e) } 得益于RGF_CJ库的安全性，窗口销毁会自动释放资源，避免了繁琐的资源释放，和释放时机的管理。实现窗口失去焦点就是否窗口的代码如下：// Yez/src/win_input.cj /** 窗口失去焦点时，销毁窗口并释放窗口资源 */ public override open func onKillFocus(e:EvKillFocus):Unit{ destroyWin() } 希望实现按下回车键触发对话，则需要在对应的事件中调用LLMImpl类的流式对话方法，代码如下：// Yez/src/win_input.cj /** 字符输入事件 */ public override open func onChar(e:EvChar):Unit{ // 判断输入的是否为回车键 if(e.char == 0x0D/*VK_RETURN*/){ // 是回车键就销毁本窗口 destroyWin() // 并且调用 开始回答函数 start() // 然后调用 LLMImpl 类的流式对话方法 // 并且传入 流的回调函数 以及 回答结束的函数 LLMImpl.i.chat(content,task:task,end:end) }else{ super.onChar(e) } } // 回复流 public var task:(String) -> Unit = {str=>} // 开始回答 public var start:() -> Unit = {=>} // 开始回答 public var end:() -> Unit = {=>} 编辑框控件的效果如下图所示：注’ 完整的代码请参考提供的代码文件反馈窗口完成上述内容后，输入与处理已经基本完备，接下来需要实现文本的显示，这一部分涉及到内容绘制，同样还是基于WinBase类进行窗口开发。创建一个反馈文本的窗口类 WinText，并且准备基本的资源代码如下：// Yez/src/win_text.cj @Abbr // 文本色彩值转Color对象宏 @LifeCycle @HookProc open class WinText <: WinBase{ // 内容纯色画刷 private let _brhCnt:SolidColorBrush // 背景纯色画刷 private let _brhbg:SolidColorBrush // 文本渲染核心 private let _textCore:TextCore // 文本格式 private let _textformat:TextFormat // 文本布局 private let _textLayout:TextLayout // 文本内容 private var _text:String = "" /** 创建窗口 */ public func createWin( x:Int32, y:Int32, nWidth:Int32, nHeight:Int32, hWndParent:RgfHwnd, substrate!:Substrate = Substrate.Direct2D ):RgfHwnd{ ... } /** 文本对象管理方法，根据配置和窗口尺寸生成文本渲染依赖的对象 */ private func createFontObj(){ let rect:Rect = surface.getRect() surface.createTextCore(_textCore) .createTextFormat(_textformat,ConfigMgr.i.fontSize) .createTextLayout(_textLayout,_text,_textformat,Float32(rect.right),Float32(rect.bottom)) } /** 创建设备/渲染相关资源事件 */ public override open func createDeviceResources():Bool{ // 创建 纯黑内容画刷 和 白色半透明背景画刷 surface.createSolidColorBrush(_brhCnt,"#000",1.0) .createSolidColorBrush(_brhbg,"#FFFA",1.0) // 创建文本相关对象 createFontObj() return true; } } 完成渲染事件，确保内容正确渲染// Yez/src/win_text.cj // 绘制窗口 public override open func onPaint(wRect:Rect):Unit{ surface.clear(0.0,0.0,0.0,0.0) // 透明模式 // 渲染背景圆角矩形 surface.fillRoundedRectangle(surface.getRect(),4.0,4.0,_brhbg) // 渲染大模型返回的文本内容 surface.drawTexts(_textCore,_brhCnt,_textLayout,originX:0.0,originY:0.0) } 完成显示文本方法。当设置文本时，若窗口类已经存在，就更新内部文本；若窗口不存在，就根据当前内容创建窗口并显示。// Yez/src/win_text.cj public func showText(str:String,p:RgfHwnd):Unit{ // 获取光标位置 let pos = rsGetCursorPos() // 获取光标所在屏幕矩形范围 let monitor = rsGetMonitorFromPoint(pos) // 如果窗口未创建，就创建窗口 if(!isCreated()){ createWin(pos.x,pos.y,100,100,p) } if(_textLayout.isValid()){ if(str != _text){ // 更新文本内容 _text = str; // 创建文本布局信息 surface.createTextLayout(_textLayout,_text,_textformat,Float32(monitor.right - monitor.left) / 2.0,Float32(monitor.bottom - monitor.top) / 2.0) } // 获取文本自动布局后的测量数据 let m = _textLayout.getMetrics() let w = Int32(m.getOrThrow().width) let h = Int32(m.getOrThrow().height) 按照测量数据移动窗口 if(pos.x > (monitor.right - monitor.left) / 2){ move(pos.x - w,pos.y - h,w,h) }else{ move(pos.x,pos.y - h,w,h) } } // 窗口显示 show(true) } 文本控件的效果如下图所示：注’ 完整的代码请参考提供的代码文件联结各个窗口逻辑完成上述各个控件后，需要实现各个组件之间的联合。实现流式信息与动画的联动// Yez/src/win_anm.cj /** 文本变更事件 */ func onTextChange(str:String):Unit{ if(!_firstText){ _firstText = true // 当文本输出开始，切换“跳跃”动画状态到“说话”动画状态 play(AnmState.Speak) } _text += str _textBox.showText(_text,hWnd) } /** 大模型开始输出文本 */ func onTextStart():Unit{ killTimer(0) _text = "" // 给大模型上发请求后，播放“跳跃”动画 play(AnmState.Jump) _firstText = false } /** 大模型结束输出文本 */ func onTextEnd():Unit{ // 大模型输出结束后，延迟5s，以让用户完整阅读完内容 const cj_bug_val:Option<CFunc<(HWND, UInt32, UInt64, UInt32) -> Unit>> = None setTimer(0,5000,cj_bug_val) } /** 时钟事件 */ func onTimer(e:EvTimer):Unit{ killTimer(0) // 用户阅读完内容后，销毁文本框窗口 _textBox.destroyWin() // 播放“等待”动画 play(AnmState.Wait) } 双击“椰椰”时打开输入框窗口// Yez/src/win_anm.cj // 客户区双击鼠标左键 func onLButtonDblClk(e:EvMouse):Unit{ // 在“等待”状态下双击“椰椰” match(_nowState){ case AnmState.Wait => {=> // 销毁文本框（虽然已无可能还显示，不过保险起见） _textBox.destroyWin() if(!_inputBox.isCreated()){ // 创建文本输入框窗口 _inputBox.createWin(hWnd) }else{ // 如果窗口存在就移动输入框窗口 // 这部分适用于希望输入框跟随鼠标位置的情况 // 对于方案二的桌面居中而言，没有必要 _inputBox.move() } () }() case _ => () } } 结语至此，仓颉AI桌宠“椰椰”就基本开发完成了，其实还有很多规划的内容可以增加，比如当和椰椰说“咔嚓”或者“拍照”时激活截图功能。这些内容就留给读者继续探索啦！本文相关链接本文项目代码库：https://gitcode.com/raozj/YezRGF_CJ库：https://gitcode.com/Cangjie-SIG/RGF_CJRGF_UI库：https://gitcode.com/raozj/RGF_UIRGF案例库：https://gitcode.com/raozj/RGF_CJ_Example华为云ModelArts：https://www.huaweicloud.com/product/modelarts/studio.html

挑战杯是“挑战杯”全国大学生系列科技学术竞赛的简称，是由共青团中央、中国科协、教育部和全国学联共同主办的全国性的大学生课外学术实践竞赛。今年，本届大赛调整为中国青年科技创新“揭榜挂帅”擂台赛，旨在进一步围绕“硬科技”创新、“卡脖子”关键核心技术攻关。华为深度参与本届“揭榜挂帅”，助力科技创新，设置22个奖项。总奖金池30万元，其中擂主团队奖金10万元。获奖者更有机会进入华为人才储备池，并优先获得实习及就业的推荐机会。 详情请点击：cid:link_001 赛题介绍赛题名称：推理大模型的训练调优与性能加速助力全栈自主AI赛题背景在大语言模型（LLMs）迅速发展的浪潮中，推理能力和运行效率的显著进步正推动着AI技术走向更广泛的应用领域。2025年，以OpenAI的o1系列和DeepSeek R1为代表的模型，不仅在复杂任务如数学解题、编程上展现出卓越能力，还通过技术创新如量化、模型并行（MOE）、知识蒸馏等，大幅提升了模型在轻量化设备上的部署可行性。 华为命题旨在应用全栈自主AI开发工具链，基于昇腾AI云服务、Ascend-snt9b AI芯片、CANN异构计算架构和ModelArts开发平台，提升轻量级模型的推理能力和性能。参赛者将利用强化学习、微调、知识蒸馏、量化等方法优化指定模型，同时探索模型部署于端侧设备的潜力，促进模型在实际场景中的应用。 赛题说明本次比赛分为初赛和总决赛，具体要求如下：初赛：组委会提供Qwen轻量级模型和示例工程，参赛团队选择强化学习（如PPO）或知识蒸馏技术，在ModelArts环境中对模型进行微调，同时优化推理性能。初赛阶段分为A榜和B榜，其中B榜的性能部分要求选手必须有算子融合、优化方面的工作。初赛最终排名以B榜成绩和代码核查结果为准，筛选出一定数量的队伍入围。总决赛（终审擂台赛）：参赛者需准备PPT进行答辩，介绍技术方案和应用demo，阐述推理能力和性能优化的方法及其实际应用场景。酌情加分项：使用华为开发者空间提供的相关资源和服务、端侧应用鸿蒙适配、及其他华为开放能力调用。 * 具体的指导文档点击文末「 阅读原文 」 获取赛题详情哦！ 02 技术能力华为昇腾AI云服务技术能力解读全栈协同优化，构建AI云底座：昇腾AI云服务重构了云基础设施，利用多样化算力池、高性能缓存池和分级存储池等多层池化技术，减少由上到下的计算代价，大大提升了资源利用率，充分释放了昇腾算力规模。 全套工具链，让AI落地更简单：AI开发生产线ModelArts，提供了包括昇腾工具链、SDK、CLI、IDE插件等端到端生产工具链，支持数据管理、模型开发、训练、推理等全流程MLOps开发，提供了统一资源调度能力，提升AI开发效率，降低AI开发门槛和成本。 开箱即用，一站式大模型开发服务：ModelArts Studio大模型即服务平台（简称MaaS），提供了简单易用的模型开发工具链，支持大模型定制开发，让模型应用与业务系统无缝衔接，降低企业AI落地的成本与难度。 * ModelArts Studio大模型即服务平台：cid:link_1华为昇腾计算技术能力解读极致性能，极简应用：CANN是昇腾针对AI场景推出的异构计算架构，向上支持多种AI框架，包括MindSpore、PyTorch等，向下服务AI处理器与编程，发挥承上启下的关键作用，是提升昇腾AI处理器计算效率的关键平台。 * CANN 助力构筑昇腾AI算力平台：cid:link_3可视化，插件化：MindStudio是华为面向昇腾AI开发者提供的全流程工具链，致力于提供端到端的昇腾AI应用开发解决方案，使能开发者高效完成模型迁移与调优、模型转换与压缩、算子开发与编译等。 * MindStudio 全流程开发工具链：cid:link_2基于上述能力，参赛开发者可以提升开发效率：使用昇腾AI云服务，开发者可以在完善的工具链基础上，实现从数据处理、算法训练到模型部署的全流程开发，更快地进行模型迭代。 提升模型性能：借助高效的模型迁移调优工具链、硬件亲和的高性能算子，以及开放的自定义算子开发能力，开发者可快速提升模型推理性能。 提升应用效果：借助优质稳定的基础大模型，使用便捷的微调和测评工具，开发者可以快速构建自己的大模型应用，提高任务精度。 03 应用场景智能终端交互手机/平板AI助手：通过轻量化模型优化（如量化、知识蒸馏），可在端侧实现低延迟的语音交互、实时翻译、日程管理等，避免云端依赖。智能家居控制：优化后的模型可部署在家庭网关或边缘设备中，实现本地化自然语言指令解析（如“调节空调温度”），提升隐私性与响应速度。金融与客服智能客服系统：通过知识蒸馏压缩模型，在银行APP或呼叫中心本地部署，实现低延迟的合规性问答与交易指导。风险实时监控：轻量化模型分析交易流水，快速识别欺诈行为，提升边缘计算场景下的响应效率。医疗与生物技术电子病历分析：在医疗终端部署优化模型，辅助医生快速生成诊断建议（需结合联邦学习保障数据安全）。基因数据推理：通过模型并行技术加速基因组序列分析，缩短科研周期。 04 技术案例美图随着美图AIGC应用版图扩大，算力供应十分紧张，华为云为美图提供从昇腾AI云服务、大模型到上层应用的全方位支持，与美图共建全栈AI解决方案。通过华为云ModelArts一站式AI开发平台，为美图提供数据处理、算法开发、模型训练、模型管理、模型部署等全流程AI开发技术能力，满足复杂场景需求。千卡集群线性度95%、昇腾云脑1000+故障模式、故障发现<1min等领先的AI基础设施能力，保障美图的大规模训练快速迭代。

挑战杯是“挑战杯”全国大学生系列科技学术竞赛的简称，是由共青团中央、中国科协、教育部和全国学联共同主办的全国性的大学生课外学术实践竞赛。团中央于 2021 年首次在第十七届“挑战杯”全国大学生课外学术科技作品竞赛中设置了“揭榜挂帅”专项赛。三届大赛来，累计吸引来自全国上千所高校、数万名学子踊跃参与。为汇聚更多青年科创人才的智慧力量，本届大赛调整为中国青年科技创新“揭榜挂帅”擂台赛，旨在进一步围绕“硬科技”创新、“卡脖子”关键核心技术攻关。2025年度中国青年“揭榜挂帅”擂台赛·华为赛道火热来袭！本次比赛聚焦大模型的推理能力提升和性能优化，参赛团队需基于昇腾全栈AI技术，使用强化学习、知识蒸馏等技术，提升选定的轻量级模型在数学计算、逻辑推理、代码生成等任务上的能力，并通过算子融合与优化、模型结构优化、量化等技术，保证精度的同时提升模型推理性能。诚邀各大高校精英云端竞技，打擂揭榜。一、即刻解锁赛事全攻略关于本次大赛，我们已为你整理总结了赛程安排、详细的参赛指导等选手关注的信息，点击海报查看。二、组队闯关！届时开放精英社群��想Get技术文档？寻找最强队友？与华为专家实时互动？扫码添加小助手，回复暗号“挑战杯”秒入群！在这里，与全国顶尖开发者共研大模型推理优化方案，一起竞技打擂，开发创新应用，助力全栈自主AI技术的新篇章！详情请点击：cid:link_0直达报名通道，攀登技术巅峰！

4月25日，由中国移动通信集团主办的“云擎未来 智信天下”2023移动云大会在苏州金鸡湖国际会议中心顺利召开。华为技术有限公司高级副总裁曹既斌受邀出席此次大会，并参加了移动云信息技术融合应用创新产业生态联合体发布仪式。移动云信息技术融合应用创新产业生态联合体的发布，标志着中国移动将凝聚合力共同构建开放包容、合作共赢的产业生态。当前，中国移动与华为正迈入新的合作周期，华为也将秉持"一切皆服务"的战略，与移动云强强联合，持续做强技术底座，为产业创造更多价值，为建设数字中国注入澎湃动能。        在本届移动云大会上，华为云计算技术有限公司副总裁黄瑾做了主题为“一切皆服务，共促云产业，深耕数智化”的演讲，阐述了华为云对于产业的思考，业务战略，以及与移动云合作的成果及未来展望。“在数字转型的关键时刻，华为云通过一切皆服务，共创行业新价值，助力企业数智化转型，共建繁荣生态共赢。中国移动和华为在云计算领域的战略合作，取得较好进展，践行“一切皆服务”，把研发投入所产生的技术和产品、数字化转型经验，伙伴的能力，都开放在云上作为云服务，提升竞争力，同时更好的共同服务客户。”

4月25日，由中国移动通信集团主办的“云擎未来 智信天下”2023移动云大会在苏州金鸡湖国际会议中心顺利召开。华为技术有限公司高级副总裁曹既斌受邀出席此次大会，并参加了移动云信息技术融合应用创新产业生态联合体发布仪式。移动云信息技术融合应用创新产业生态联合体的发布，标志着中国移动将凝聚合力共同构建开放包容、合作共赢的产业生态。当前，中国移动与华为正迈入新的合作周期，华为也将秉持"一切皆服务"的战略，与移动云强强联合，持续做强技术底座，为产业创造更多价值，为建设数字中国注入澎湃动能。        在本届移动云大会上，华为云计算技术有限公司副总裁黄瑾做了主题为“一切皆服务，共促云产业，深耕数智化”的演讲，阐述了华为云对于产业的思考，业务战略，以及与移动云合作的成果及未来展望。“在数字转型的关键时刻，华为云通过一切皆服务，共创行业新价值，助力企业数智化转型，共建繁荣生态共赢。中国移动和华为在云计算领域的战略合作，取得较好进展，践行“一切皆服务”，把研发投入所产生的技术和产品、数字化转型经验，伙伴的能力，都开放在云上作为云服务，提升竞争力，同时更好的共同服务客户。”

        4月25日，由中国移动通信集团主办的“云擎未来 智信天下”2023移动云大会在苏州金鸡湖国际会议中心顺利召开。华为技术有限公司高级副总裁曹既斌受邀出席此次大会，并参加了移动云信息技术融合应用创新产业生态联合体发布仪式。移动云信息技术融合应用创新产业生态联合体的发布，标志着中国移动将凝聚合力共同构建开放包容、合作共赢的产业生态。当前，中国移动与华为正迈入新的合作周期，华为也将秉持"一切皆服务"的战略，与移动云强强联合，持续做强技术底座，为产业创造更多价值，为建设数字中国注入澎湃动能。        在本届移动云大会上，华为云计算技术有限公司副总裁黄瑾做了主题为“一切皆服务，共促云产业，深耕数智化”的演讲，阐述了华为云对于产业的思考，业务战略，以及与移动云合作的成果及未来展望。“在数字转型的关键时刻，华为云通过一切皆服务，共创行业新价值，助力企业数智化转型，共建繁荣生态共赢。中国移动和华为在云计算领域的战略合作，取得较好进展，践行“一切皆服务”，把研发投入所产生的技术和产品、数字化转型经验，伙伴的能力，都开放在云上作为云服务，提升竞争力，同时更好的共同服务客户。”

dify版本支持：≥v1.0（我环境1.4）1.停止dify运行（在dify/docker目录下）docker-compose down2.导入公钥（将公钥上传至服务器）mkdir docker/volumes/plugin_daemon/public_keyscp dify_key_pair.public.pem docker/volumes/plugin_daemon/public_keys3.修改环境变量配置（/dify/docker/docker-compose.yaml）vi /dify/docker/docker-compose.yaml具体添加修改内容如下：THIRD_PARTY_SIGNATURE_VERIFICATION_ENABLED: trueTHIRD_PARTY_SIGNATURE_VERIFICATION_PUBLIC_KEYS: /app/storage/public_keys/your_key_pair.public.pem4.plugin daemon时间同步同上修改环境变量配置- /etc/localtime:/etc/localtime:ro # 同步时间保存退出5.启动difydocker-compose up -d6.登录dify准备导入工具登陆后点击“插件”点击“安装插件”→“本地插件”选择“baidu_search.signed.difypkg”等待安装成功可查看docker日志观察安装是否存在问题（一般问题会出现在pip拉取依赖失败-自行解决）查看docker实例“dify-plugin-daemon”的ID或名称docker psdocker lofgs -f [容器名称或 ID]备注：如若安装不成功联系我获取插件代码。可能存在问题：导入提示不良签名（需要重新生成签名，通过作者获取插件源文件重新打包）三方签名生成：（自行参考第三方签名验证 - Dify Docs），发不了url检测不过。依赖安装不成功，自行进入实例“dify-plugin-daemon”，虚拟环境中手动安装依赖或者采用国内pip源进行安装。

DeepSeek-R1于2025年5月29日在huggingface、modelscope开源了最新的DeepSeek-R1-0528这次根据官方说法，是小范围升级，目前实测来看，知识库更新到2025年3月，而且在某些任务下DeepSeek-R1的思考时间可能有30-60分钟livecodebench排行榜分数也不错华为云也即将上线新版本DeepSeek-R1-0528模型，大家可以在升级后去体验最新的模型！

华为云 BMS 部署 Qwen3-32b 实战指南在人工智能蓬勃发展的当下，大模型的应用愈发广泛。Qwen3-32b 作为一款备受关注的大模型，在华为云 BMS 上的部署过程是很多开发者关心的焦点。本文将基于详细的部署文档，为大家呈现华为云 BMS 部署 Qwen3-32b 的完整流程，助力大家顺利搭建起属于自己的大模型服务环境。一、华为云 BMS 配置概览在开始部署之前，先来了解一下本次部署所使用的华为云 BMS 配置。具体配置如下：机型：physical.kat2e.48xlarge.8.280t.ei.c002CPU：Kunpeng 920 (4*48Core@2.6GHz)，强大的多核心处理器为模型运行提供了稳定的计算基础。内存：24*64GB DDR4 RAM，充足的内存空间确保模型在处理大量数据时能够流畅运行。本地磁盘：3*7.68TB NVMe SSD，高速的存储设备可以快速读写模型数据和相关文件。扩展配置：2100GE + 8200GE，高速网络配置为数据传输提供了有力保障。镜像：HCE2.0-Arm-64bit-for-Snt9A2-BareMetal-with-24.1.0-7.5.0.3.220-CANN7.6.0.1.220，该镜像为本次部署提供了特定的运行环境。二、环境准备工作1.1 检查 NPU 设备SSH 登录到机器后，首要任务是检查 NPU 设备的状态。通过运行以下命令来获取 NPU 设备信息：查看 NPU 卡状态：npu-smi info，此命令可以在每个实例节点上运行，查看 NPU 卡的具体运行情况。确认总卡数：npu-smi info -l | grep Total，通过该命令确认对应卡数是否已经成功挂载。查看驱动和固件版本：npu-smi info -t board -i 1 | egrep -i "software|firmware"，如果在执行这些命令时出现错误，可能是 NPU 设备没有正常安装，或者 NPU 镜像被其他容器挂载。此时需要先正常安装固件和驱动，或者释放被挂载的 NPU，确保 NPU 设备能够正常工作。1.2 检查与安装 dockerdocker 是部署过程中不可或缺的工具，接下来检查 docker 是否已经安装。可以通过docker -v命令查看 docker 的版本信息，如果尚未安装，则需要运行以下命令进行安装：yum install -y docker-engine.aarch64 docker-engine-selinux.noarch docker-runc.aarch64安装完成后，还需要配置 IP 转发，以实现容器内的网络访问。执行sysctl -p | grep net.ipv4.ip_forward命令查看net.ipv4.ip_forward配置项的值，如果为 1，则说明已经配置好，可跳过此步骤；若不为 1，则执行以下命令进行配置：sed -i's/net\\.ipv4\\.ip\_forward=0/net\\.ipv4\\.ip\_forward=1/g' /etc/sysctl.conf sysctl -p | grep net.ipv4.ip\_forward1.3 获取基础镜像为了确保部署的稳定性和兼容性，建议使用官方提供的镜像来部署推理服务。镜像地址可以通过特定方式获取，获取后使用docker pull命令进行下载，例如：官方镜像目前为对外公布敬请期待！！！docker pull quay.io/ascend/vllm-ascend:v0.8.5rc1（这个镜像理论上可行）1.4 获取权重文件权重文件是 Qwen3-32b 模型运行的关键数据，需要通过解决方案实践 —OBS 极速同步模型权重文件，将其导入到自己的 OBS 中，然后再下载到服务器中。具体下载步骤如下：获取权重文件方式：https://www.huaweicloud.com/solution/implementations/replicating-model-weight-files-across-obs-buckets.html1.4.1 权重文件下载到服务器使用wget命令下载obsutil工具：wget https://obs-community.obs.cn-north-1.myhuaweicloud.com/obsutil/current/obsutil_linux_arm64.tar.gz解压下载的文件：tar -xzvf obsutil_linux_arm64.tar.gz进入解压后的目录：cd obsutil_linux_arm64_5.7.3/赋予obsutil执行权限：chmod 755 obsutil配置obsutil：./obsutil config -i=ak -k=sk -e=obs.cn-north-9.myhuaweicloud.com下载权重文件：./obsutil cp obs:/bucket-test/Qwen3-32B /home/Qwen3-32B -r -f三、启动部署流程2.1 启动容器在完成环境准备后，就可以启动容器了。启动容器需要使用一系列命令参数，具体如下：docker run -itd \ --device=/dev/davinci0 \ --device=/dev/davinci1 \ --device=/dev/davinci2 \ --device=/dev/davinci3 \ --device=/dev/davinci4 \ --device=/dev/davinci5 \ --device=/dev/davinci6 \ --device=/dev/davinci7 \ -v /etc/localtime:/etc/localtime \ -v /usr/local/Ascend/driver:/usr/local/Ascend/driver \ -v /etc/ascend_install.info:/etc/ascend_install.info \ --device=/dev/davinci_manager \ --device=/dev/devmm_svm \ --device=/dev/hisi_hdc \ -v /var/log/npu/:/usr/slog \ -v /usr/local/sbin/npu-smi:/usr/local/sbin/npu-smi \ -v /sys/fs/cgroup:/sys/fs/cgroup:ro \ -v ${dir}:${container_work_dir} \ --net=host \ --name ${container_name} \ ${image_id} \ /bin/bash 这里以一个实际示例来说明：docker run -itd --restart=always --device=/dev/davinci0 --device=/dev/davinci1 --device=/dev/davinci2 --device=/dev/davinci3 --device=/dev/davinci4 --device=/dev/davinci5 --device=/dev/davinci6 --device=/dev/davinci7 -v /etc/localtime:/etc/localtime -v /usr/local/Ascend/driver:/usr/local/Ascend/driver -v /etc/ascend\_install.info:/etc/ascend\_install.info --device=/dev/davinci\_manager --device=/dev/devmm\_svm --device=/dev/hisi\_hdc -v /var/log/npu/:/usr/slog -v /usr/local/sbin/npu-smi:/usr/local/sbin/npu-smi -v /sys/fs/cgroup:/sys/fs/cgroup:ro -v /home/Qwen3-32B:/home/Qwen3-32B --net=host --name qwen3-32b 628e52638d5d /bin/bash2.2 进入容器与资源评估容器启动成功后，使用docker exec -it -u ma-user ${container_name} /bin/bash命令进入容器。进入容器后，首先运行npu-smi info命令评估推理资源，检查 NPU 设备信息可用的卡数，同时确认卡是否被占用、端口是否被占用以及是否有对应运行的进程。然后根据实际情况配置需要使用的 NPU 卡，例如，当有 8 张卡时，可以通过export ASCEND_RT_VISIBLE_DEVICES=0,7命令进行配置。2.3 启动大模型最后一步就是启动大模型了，使用以下命令：nohup python -m vllm.entrypoints.openai.api_server --model /home/Qwen3-32B \ --max-num-seqs=256 \ --max-model-len=131072 \ --max-num-batched-tokens=131072 \ --tensor-parallel-size=8 \ --block-size=128 \ --host=0.0.0.0 \ --port=8080 \ --gpu-memory-utilization=0.9 \ --num-scheduler-steps=8 \ --trust-remote-code \ --enforce-eager \ > /home/Qwen3-32B/logs/vllm_api_server.log 2>&1 & 通过以上步骤，我们就完成了华为云 BMS 部署 Qwen3-32b 的全部过程。在实际部署过程中，可能会遇到各种各样的问题，需要大家根据具体情况进行排查和解决。希望本文能够对大家有所帮助，让大家在大模型部署的道路上少走弯路，顺利搭建起自己的人工智能服务环境。

使用 gradio 框架的时候可能遇到以下错误警告：Could not create share link. Missing file: /usr/local/lib/python3.10/dist-packages/gradio/frpc_linux_amd64_v0.3.  Please check your internet connection. This can happen if your antivirus software blocks the download of this file. You can install manually by following these steps:  1. Download this file: https://cdn-media.huggingface.co/frpc-gradio-0.3/frpc_linux_amd64 2. Rename the downloaded file to: frpc_linux_amd64_v0.3 3. Move the file to this location: /usr/local/lib/python3.10/dist-packages/gradio这个错误信息表明 Gradio 在尝试创建一个共享链接时，缺少了一个必要的文件 frpc_linux_amd64_v0.3。这个文件是 Gradio 用来建立本地服务器和外部网络之间的连接的工具。解决方法根据错误信息，你可以按照以下步骤手动解决这个问题：下载文件：打开浏览器，访问以下链接下载文件： https://cdn-media.huggingface.co/frpc-gradio-0.3/frpc_linux_amd64重命名文件：将下载的文件重命名为 frpc_linux_amd64_v0.3。移动文件：将重命名后的文件移动到以下目录：/usr/local/lib/python3.10/dist-packages/gradio设置文件权限（如果需要）：你可能需要给这个文件设置可执行权限。可以使用以下命令：chmod +x /usr/local/lib/python3.10/dist-packages/gradio/frpc_linux_amd64_v0.3重新运行 Gradio：完成以上步骤后，重新运行你的 Gradio 应用程序，看看问题是否解决。其他注意事项网络连接：确保你的网络连接稳定，特别是在下载文件时。权限问题：如果你在移动文件或设置权限时遇到问题，可能需要使用 sudo 命令来提升权限。完整命令如下：# 下载文件 wget https://cdn-media.huggingface.co/frpc-gradio-0.3/frpc_linux_amd64 # 检查文件是否下载成功 ls -l frpc_linux_amd64 # 重命名文件 mv frpc_linux_amd64 frpc_linux_amd64_v0.3 # 确保目标目录存在 sudo mkdir -p /usr/local/lib/python3.10/dist-packages/gradio # 复制文件到目标目录 sudo cp frpc_linux_amd64_v0.3 /usr/local/lib/python3.10/dist-packages/gradio # 设置文件权限 sudo chmod +x /usr/local/lib/python3.10/dist-packages/gradio/frpc_linux_amd64_v0.3验证是否成功运行以下命令检查文件是否已正确放置并具有可执行权限：ls -l /usr/local/lib/python3.10/dist-packages/gradio/frpc_linux_amd64_v0.3输出应类似于：-rwxr-xr-x 1 root root XXXX Oct 10 12:34 /usr/local/lib/python3.10/dist-packages/gradio/frpc_linux_amd64_v0.3重新启动项目后，就能正常运行了。

胶囊神经网络（Capsule Neural Network，简称 CapsNet）是由深度学习领域的先锋之一 Geoffrey Hinton 等人提出的神经网络结构，旨在改善传统卷积神经网络（CNN）在处理空间层次结构能力上的不足，并提升对图像中物体姿态（pose）、方向、位置变化等感知不变性的建模能力。为什么要提出胶囊网络？传统 CNN 在图像分类等任务中取得了巨大成功，但存在一些局限性：丢失空间信息：经过池化操作（如最大池化）会丢失物体在图像中的空间结构。对小形变不鲁棒：CNN 对图像的旋转、平移、缩放敏感，降维后容易损失重要空间信息。模块性差：由一系列不可解释的神经元组成的黑盒，缺乏结构上的语义模块化。Hinton 提出 胶囊网络 正是为了解决这些限制，特别是对图像中对象的层次结构与姿态感知进行更精确的建模。胶囊网络的基本概念什么是 Capsule？胶囊 是一组神经元的集合。每个胶囊通过其向量输出表示某个特征的存在程度（magnitude） 以及该特征的各种属性（如位置、方向、比例等）。比如，一个胶囊可以识别“眼球”这一特征，并通过输出向量来表示其位置、角度、大小等。胶囊网络的核心创新1. 动态路由（Dynamic Routing）传统神经网络中，信号从一层向下一层做前向传播。胶囊网络采用“动态路由”算法，通过迭代机制选择哪些胶囊应该被上层胶囊激活，从而实现一种“模块化”的信息流。目标：让父胶囊通过子胶囊的信息组合，采用“加权投票”的方式决定上层胶囊的输出。动态路由算法思路如下：初始阶段，设置预测输出向量（子胶囊到父胶囊的预测）。计算“耦合系数”，表示上层胶囊优先选择哪些下层胶囊的信息。使用耦合系数进行加权求和，确定上层胶囊的输出。迭代更新耦合系数，直到收敛或达成稳定。这是一种“自底向上”的注意力机制，类似于注意力机制中的“投票”过程。胶囊网络的结构层次（以 CapsNet 为例）以最经典的图像分类任务中使用的结构为例：卷积层 + 池化层（和 CNN 类似）Primary Capsules（原始胶囊层）：输入为卷积层的高维激活图。输出为胶囊输出的向量表示（如 [8 维向量]），每个胶囊对应某种简单位姿的特征（如边缘、角点等）。Digit Capsules（数字胶囊层）：上一层原始胶囊输出链接到 Digit Capsules。使用动态路由算法进行信息聚合。每个 Digit Capsule 表示一个类别的胶囊（例如 10 个分别表示 0~9）。Margin Loss 作为目标函数：根据 Digit Capsule 输出向量（长度）来计算样本预测的类别得分。示例：CapsNet 的结构流程以 MNIST 手写数字数据集为例，CapsNet 流程如下：Input Layer（图像 28 x 28）Convolutional Layer（卷积 + ReLU）PrimaryCaps（编码低层特征，比如边缘）输出 32 个胶囊，每个胶囊 8 尺度的向量。DigitCaps（使用动态路由算法，编码高层对象）10 个胶囊，每个对应一个数字类。Length of Capsule Output as Class Probability向量的长度代表该类别出现的可能性。Marginal Loss（损失函数）根据标签，最小化目标类别的向量长度，最大化非目标类别的长度。示例代码（简单 CapsNet 模型）# 初始卷积层 conv1 = Conv2D(...)(input) # PrimaryCaps 层 primary_caps = PrimaryCapsulation(...) # 输出形状如 (None, 32, 6, 6, 8) # DigitCaps 采用 Dynamic Routing digit_caps = DigitCapsulation(...) # 输出 10 个胶囊，每个 16 尺度 # 根据 capsule 的输出长度预测分类 output = Length()(digit_caps) # Loss Function: margin loss loss = MarginLoss()([output, labels]) 总结项目说明胶囊网络一种结构感知性更强的神经网络结构核心思想用向量表示特征，用动态路由传递信息优势更强的姿态建模、空间不变性、类人层次结构感知不足训练复杂、工业落地较少适用领域小样本图像识别、姿态估计、可解释AI等胶囊网络相关论文《Dynamic Routing Between Capsules》作者：Sara Sabour, Nicholas Frosst, Geoffrey Hinton年份：2017提出了经典的 CapsNet 架构论文链接 (arXiv)

“元学习”（Meta Learning，也称为“学会学习”）是一种机器学习的高级方法，它的核心思想是让模型学会如何学习，而不是仅仅学会某个具体任务。与传统的机器学习方法不同，元学习不仅关注在具体问题上的性能，更关注模型的泛化能力与学习能力，使其能够更高效地学习新任务。元学习的定义元学习（Meta-Learning） 是一种机制或框架，允许模型从多个任务中提取共同的学习策略或结构信息，进而帮助模型在**新任务（未见过的任务）**上快速而有效地进行学习。通俗地说，元学习是“教模型如何学习”，而非“教模型怎么做某个具体任务”。元学习的关键概念概念说明任务（Task）一个具体的训练目标，例如分类一个图像、预测商品销量等。元任务（Meta-Task）在多个任务中学习跨任务的模式，例如图像分类的元任务就是让模型学会识别图像中物体的一般特征。样本任务（Training Tasks）用于训练模型学会如何学习的任务集合，是元学习的训练数据。目标任务（Target Task）学习完元任务后，部署或测试模型的目标任务，即新出现的、数据量少或未知的任务。元学习的应用场景小样本学习（Few-shot Learning）：构建一个模型，可以在只有少量样本的情况下，快速掌握新类别。举例：MAML、ProtoNet、Reptile 等。超参数学习（Hyperparameter Optimization）：让模型学会找出最优秀的超参数组合，而非手动调参。神经网络架构搜索（Neural Architecture Search, NAS）：让模型学会什么样的网络结构最有助于某个任务。强化学习中的策略迁移：在多个环境中训练一个元策略，用于快速适应新环境。典型元学习算法算法说明MAML（Model-Agnostic Meta-Learning）一种模型无关的元学习方法，通过为模型参数寻找一个“好起点”，使其对新任务只需少量学习即可取得良好效果。ProtoNet基于原型网络的元学习，通过学习类别原型来进行小样本分类。Reptile采用参数多任务更新的方式，避免显式优化元目标函数。Meta-SGD元学习优化器，让模型不仅学习参数，还学习学习率等元参数。元学习与传统机器学习的区别项目传统机器学习元学习目标解决单一任务从多个任务中学习一种通用学习方式数据量每个任务需大量数据每个任务可小样本训练泛化性泛化能力依赖任务训练数据通过元学习提升对新任务的泛化能力模型更新每次任务模型重新训练一次训练学习如何有效学习元学习的优势显著提升小样本任务的性能减少训练时间和数据量依赖适用于多任务、多领域学习促进迁移学习与强化学习的结合元学习的挑战需要大量训练任务（元训练集）计算成本高，训练复杂度高难以设计合适的元目标函数泛化能力仍存在不确定性总结元学习是一种“学习如何学习”的方法，通过模型在多个任务中的训练，提取和泛化出一种学习能力。它非常适合那些数据稀缺、任务多样、训练效率要求高的场景。元学习是连接传统机器学习和人类学习能力的桥梁，是AI向更智能方向发展的重要一步。

当使用 ollama run xxxx --verbose 命令运行模型时，Ollama 会在每次回答完成后输出一组详细的性能统计信息，这些信息可以帮助理解模型在处理当前任务时的计算效率、资源消耗、瓶颈分析等关键性能指标。示例输出：total duration: 53.816488258s load duration: 17.007831ms prompt eval count: 2048 token(s) prompt eval duration: 1.129753292s prompt eval rate: 1812.79 tokens/s eval count: 1480 token(s) eval duration: 52.632442857s eval rate: 28.12 tokens/s 各项指标详解：total duration: 53.816488258s总处理时间从模型初始化、加载到输入处理、响应生成的整个流程所花费的总时间（单位为秒）用于评估整体延迟，适合对比不同输入或模型之间的性能表现load duration: 17.007831ms模型加载时间Ollama 加载模型（包含计算图、权重、初始化）所需的时间（单位为毫秒）如果之前已加载过该模型，该值可能为 0 或极短prompt eval count: 2048 token(s)Prompt 环境的 tokens 数量表示输入提示（Prompt）在模型上下文窗口中的 token 总数token 数量越多，处理时间、内存占用等通常也会增加prompt eval duration: 1.129753292sPrompt 相关计算耗时模型对 prompt 的 token 进行处理和编码所消耗的时间用于解析用户输入的上下文，预计算状态prompt eval rate: 1812.79 tokens/sPrompt 处理速度prompt eval count / prompt eval duration 的结果该值越高，说明模型处理输入文本越快eval count: 1480 token(s)生成内容的 tokens 数量表示模型在此次响应中生成的内容 token 数量（即输出长度）通常对应回答的字数eval duration: 52.632442857s生成内容的计算时间模型生成每个 token 所花费的总时间评估响应生成效率，判断是否被“卡顿”、“延迟”等性能问题影响eval rate: 28.12 tokens/s生成 token 的速度eval count / eval duration 的结果反映模型生成内容时的速度。该值越低，表示 response 生成越慢总结：指标含义相关性total duration整个流程耗时延迟表现load duration模型加载耗时启动性能prompt eval count输入 tokens 数量上下文大小prompt eval duration输入处理耗时输入解析prompt eval rate输入解析速度Input 处理效率eval count输出 tokens 数量回答长度eval duration生成答案耗时响应生成延迟eval rate生成速度实时性 / 响应效率这些参数可以帮助分析：模型在不同输入下的处理能力优化输入提示的长度或结构判断性能瓶颈（如响应生成慢）比较不同模型或不同版本的性能差异