7月22日-7月24日，为期三天的华为云鸿蒙应用开发师资培训·湖北站在湖北轻工职业技术学院圆满落下帷幕。本次培训聚焦鸿蒙应用开发技术与华为云认证，吸引了20多所学校的40多位教师参与，得到了参培老师的高度评价。       培训伊始，湖北轻工职业技术学院信息工程学院院长赵欣发表了致辞，他强调，随着万物互联时代的加速到来，鸿蒙操作系统正迅速成为推动各行业数字化转型的关键力量。他表示希望参培教师能够以本次培训为契机，不断学习新技术、新技能，提升自身的综合素质和能力水平，为培养更多优秀人才贡献力量。       随后，湖北云解决方案部部长苑磊首先对到场的老师们表示热烈的欢迎，并表示HarmonyOS 已走进百所高校课堂，成为热门课程，参加培训的各位老师也是将鸿蒙与开发者认证推向全国课堂的 “火种”。他希望能以此次培训为纽带，与各位老师持续深化交流合作，培养出更多适应未来社会需求的高素质人才。       接下来，华为云高校生态运营经理陈银豪为大家带来了“推进产教融合走向深水区，助力高校创新实践人才培养”的议题分享。他分享了华为云开发者人才培养在鸿蒙、人工智能等产品领域的人才解决方案，围绕人才培养的关键节点，构建深度完善的运营体系，通过“岗、课、赛、证、师、会”等全链条的生态运营服务，深化产教融合，助力高校人才培养，培养高质量实践人才。       培训期间，讲师从HarmonyOS简介、设计理念、技术架构、技术特性等多个角度对HarmonyOS进行介绍，使老师们对HarmonyOS具备一个较为系统全面的认识。紧接着，讲师阐述了鸿蒙应用开发的编程语言—ArkTS，系统地介绍了ArkTS语言的基础语法、UI描述规范、常用装饰器和渲染控制语法等知识。       通过对HarmonyOS基础知识与开发语言的学习，讲师又详细讲述基于ArkTS的声明式开发范式组件，包括基础组件、容器组件和媒体组件。ArkUI 是一套声明式开发框架，它具备简洁自然的 UI 信息语法、丰富的 UI 组件、多维状态管理，以及实时多维度预览等能力，帮助开发者提升应用开发效率，并能在多种设备实现生动而流畅的用户体验。讲师通过实际的组件应用案例，帮助老师们理解和区分ArkTS中的各类组件，具备在实际应用开发中熟练使用ArkTS组件的能力。       本次师资培训通过课堂学习、实验交流、考试等环节，旨在帮助参培老师掌握开发者认证课程知识体系以及提升鸿蒙开发实践能力，更好地将鸿蒙应用开发技术融入教学体系，为教学活动及人才建设提供有力的支撑。培训尾声，组织参培老师参加结课考试并为他们发放结课证书。通过本次暑期师资培训，老师们表示他们不仅掌握了鸿蒙应用开发的基础知识和最新趋势，还提升了鸿蒙开发的职业技能。       本次暑期师资培训的成功举办，离不开湖北轻工职业技术学院的鼎力支持与高度重视，也离不开各位教师的积极参与和辛勤付出。未来，我们期待更多的教师能够参与到华为云根技术与鸿蒙应用开发的培训和学习中来，促进高素质人才的培养。

亲爱的用户：您好！为改善用户体验，华为云开发者学堂《GaussDB工作级开发者认证》于2025年7月31日至8月1日下线升级，升级期间暂停在线学习、实验练习及考试，预计8月2日重新上线，届时请您关注华为云开发者学堂资讯专栏-业务动态通知，给您带来的不便，敬请谅解。如有其他问题，可拨打客服热线4000-955-988或950808反馈和咨询。

尊敬的客户：您好！ 为了确保昇腾云工作级开发者认证课程的时效性和技术领先性，我们将对《华为昇腾云工作级开发者认证》内容进行全面升级，预计2025年9月2日正式更新上线！更新内容如下：1.      整体课程模块的内容组织逻辑调整：因后续实验需要依赖平台进行操作，故先介绍昇腾云服务相关平台，再介绍适配昇腾云的模型开发，最后介绍未适配昇腾云的模型的迁移调优，调整后逻辑顺序为：昇腾AI基础-昇腾云服务AI平台-昇腾云服务AI框架-大模型开发套件-昇腾迁移；2.      昇腾AI基础：精简机器学习、深度学习等经典AI理论内容，增加LLM、CV大模型、多模态大模型等基础理论；3.      昇腾云服务AI平台：增加ModelArts Standard、ModelArts Studio（MaaS）平台介绍和相关实验；4.      大模型开发套件：增加大模型训练推理相关的基础和优化方法介绍，增加常用的vLLM、MindSpeed、LLaMA Factory等大模型套件介绍和实验，删除MindFormers介绍和实验；5.      模型适配：刷新昇腾迁移调优过程中的工具链介绍（如精度调优工具改为msprobe，删除不常用的工具如AIT、PTQ等），并刷新配套实验。

系统学习人工智能，掌握硬核AI技术，通关秘籍来啦！！！昇腾AI专区精心打造的「6步进阶式学习路径」正式上线，为学习者提供了一条全面且系统的学习路径，帮助学习者掌握关键技术，推动AI创新与应用。无论你是初学者还是经验丰富的开发者，都能在这条学习路径中找到适合自己的学习方向。助你从入门到精通，清晰路径，步步为营，稳扎稳打攀登AI高峰！一、人工智能：AI世界的万能钥匙「Python」Python作为AI领域的“通用语言”，是AI学习中效率最高的工具之一，它能让你把精力聚焦在AI算法和业务逻辑上，而不是被复杂的编程语法束缚。无论是入门机器学习还是开发复杂的大模型应用，Python都是不可或缺的基础技能。第一步从动手学Python开始，从基础语法到高级编程技巧，再到实际项目实践，掌握Python核心知识与应用能力。① 动手学Python② python数据处理③ 实用AI库④ 动手学机器学习⑤ 动手学深度学习二、昇腾基础入门：走进昇腾AI进入昇腾基础入门阶段，学习者将正式踏入昇腾AI的领域。这一板块主要介绍昇腾AI硬件的全栈技术体系，深度剖析昇腾架构核心模块，解读910/310等AI处理器的性能特性与场景适配策略，并实战演练Atlas全栈产品的部署方案，为昇腾AI解决方案开发提供芯片级调优能力。以及CANN（Compute Architecture for Neural Networks）异构计算架构的基础知识。学习者将掌握基于AscendCL的高性能应用开发技能，深入理解GE图引擎优化技术，并熟练运用ATC、AOE、AMCT等核心…系统培养昇腾AI处理器的高性能算子开发能力。① 昇腾硬件② 异构计算架构CANN③ 昇腾算子开发三、昇腾模型开发：掌握模型构建核心昇腾模型开发板块深入探讨基于昇腾平台的模型开发技术。将系统培养基于昇腾平台的PyTorch全栈开发能力，覆盖环境配置、模型迁移、性能优化及精度调优全链路实战。提供昇腾平台无缝迁移全栈方案，深度融合MindSpore框架与昇腾硬件特性。① 昇腾PyTorch开发② 昇腾模型开发工具链③ 昇腾PyTorch三方库④ 昇腾PyTorch经典任务实践⑤ 昇腾MindSpore开发⑥ 昇腾MindSpore迁移开发四、大模型开发：探索昇腾大模型框架在大模型开发板块，学习者将深入了解大模型的原理、架构和训练方法、高效推理能力。学习昇腾平台大模型训练核心技术，聚焦MindSpeed框架的并行架构革新、计算加速优化与内存极致压缩三大维度。① 昇腾大模型训练框架② 昇腾大模型推理框架五、应用开发：实现AI技术落地应用开发是将AI技术转化为实际价值的关键环节。在这一板块，学习者将学习构建大语言模型的高效交互能力，从提示工程基础到进阶应用，通过精准提示词设计掌握模型行为控制、任务分解与多轮对话编排技术，实现大模型输出质量与可靠性的双重提升。聚焦LangChain框架的核心技术实践，实现复杂任务的自动化分解与执行编排。详细介绍Pangu-Pro-MoE、DeepSeek等经典大模型在提示工程、代码编写、信息抽取以及Agent构建等方面的应用，LlamaIndex框架的智能应用开发能力。① 提示工程② LangChain大模型应用开发③ 经典大模型应用开发④ LlamaIndex大模型应六、AI4S：拓展AI应用新领域AI4S（AI for Science）是AI技术在科学研究领域的应用，为科学研究带来了新的方法和思路。在AI4S板块，学习者将深度融合人工智能与生命科学前沿技术，聚焦生物序列智能建模，为精准医疗与生物工程提供数据驱动的科研新范式。并聚焦AI技术在地球科学领域的应用与前沿发展，培养学生利用机器学习、深度学习等方法解决地球系统问题的能力。通过这一板块的学习，学习者能够拓展AI应用的视野，为未来在科学研究领域的创新提供技术支持。① AI与生命科学② AI与数学/物理求解③ AI与地球科学此外，昇腾AI专区学习路径还涵盖了经典大模型应用开发、使用AI库、动手学机器学习等热门学习路径。通过实际操作和案例分析，学习者能够更加深入的理解和掌握相关技术。昇腾AI专区学习路径为学习者提供了一个全面、系统、循序渐进的学习框架，帮助学习者逐步掌握昇腾关键技术，推动AI创新与应用。无论你是希望在AI领域开启职业生涯，还是寻求技术突破和创新，都能在这条学习路径中找到属于自己的成长之路。快来加入昇腾AI专区的学习之旅，点这里>>cid:link_0一起探索AI世界的无限可能！

【获奖公示】一、活动获奖公示信息如下：（1）活动期间账户积分数据见本论坛贴附件1（2）积分兑换奖品公示名单见附件2（含不符合条件用户）（3）完成云实验抽奖公示名单见附件3（含不符合条件用户）（4）邀请有礼人数公示见附件4（5）完成空间案例实操抽奖中奖公示链接：cid:link_21二、公示时间：2025年8月8日—2025年8月17日（含），若有疑问请在该时间段反馈，逾期视为放弃奖励！三、积分数据统计周期：仅统计6月30日—8月7日期间的考试数据（首次考取以上认证可参与积分），如积分数对不上请先检查是否考了重复的认证或微认证证书未生成，微认证要证书生成才核算积分。四、奖品发放：所有奖励将于活动公示期后陆续安排发放，其中本活动中的实物礼品如遇缺货等情况，将替换成类似款或其他等值礼品发放。五、积分兑换好礼（点此填写积分兑换问卷-仅可兑换1次，填写时间截止到8月12日18:00，请及时填写）六、点此填写活动满意度调查问卷 【活动时间】2025年6月30日—2025年8月7日【活动报名】立即报名【活动福利】福利1：邀请报名有礼，最高可领1000元云资源代金券；福利2：积分兑好礼，最高可兑换1000元云资源券代金券、1099元工作级开发者认证代金券福利3：完成云实验实操抽奖，抽华为手环9、定制双肩包、开发者定制冲锋衣等礼品；  福利4：完成开发者空间案例实操抽奖，抽华为智能体脂称3、定制雨伞、定制冲锋衣等礼品。【微认证1元购】序号认证名称（含购买入口）积分数价格1听歌识曲-抖音小视频背景音乐识别2282实现图片压缩及水印添加2283华为云计算服务实践2284华为云存储服务实践2285Web暴力破解漏洞挖掘2286AI智能语音识别计算器2287ModelArts实现智能花卉识别2288ModelArts实现零售商客户分群2289基于昇腾AI处理器的算子开发22810电子相册智慧整理22811微信公众号后台消息回复22812搭建Discuz论坛网站22813基于鲲鹏搭建zabbix分布式监控系统22814华为云数据库服务实践22815基于物联网平台构建智慧路灯应用22816使用低代码平台开发园区访客应用22817华为企业级JAVA编程规范22818华为云上两地三中心实践24819FunctionGraph服务入门及实战248点击前往活动页，可领取考试代金券及查看详细福利。部分奖品限量，先到先得，赶紧来参加吧！

基于仓颉 + 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

尊敬的客户：您好！由于微认证生命周期管理规定，《智能信息过滤和图片鉴别》微认证将于2025年7月30日正式下线停止服务。为此，我们将采取以下措施：1．对于已购买该微认证的客户，如有课程学习需求，仍可通过“个人中心-我的学习”中通过搜索课程进行学习，但课程内容已不再维护刷新，请注意课程内容的时效性。2．对于已购买该微认证并通过考试领取证书的客户，原证书在证书有效期内仍有效；如暂未通过考试且仍有考试机会的客户，可在微认证正式停止服务之前参加考试并领取证书，证书在有效期内仍有效；3．对于已购买该微认证但未使用代金券进行实验操作的客户：1) 在2025年6月30日前，自购买日期起在30天内（代金劵有效）未进行实验操作的客户，可在停止服务之前正常开展实验。2) 在2025年6月30日前，自购买日期起超出30天（代金劵失效）未进行实验操作的客户，不建议继续开展实验。如您有任何问题，可随时通过工单或者服务热线（+86-4000-955-988 ）与我们联系。感谢您对华为云微认证的支持！

尊敬的客户：您好！为保证您获得最佳的学习与实验体验，华为云开发者学堂于2025年6月24日对微认证《黑白棋实时对战游戏云上开发》进行下线优化，预计2025年7月15日重新上线，届时请您关注华为云开发者学堂资讯专栏-业务动态通知。为此，我们将采取以下措施：1．对于已购买该微认证并通过考试领取证书的客户，原证书在证书有效期内仍有效，并与课程优化后的证书拥有同等效力；2．对于已购买该微认证但未通过考试且仍有考试机会的客户，可在重新上线后进行新课程学习、实验，参加考试并领取证书；3．对于已购买该微认证但未完成实验操作的客户，请您关注华为云开发者学堂资讯专栏-业务动态的上线通知，待实验优化重新上线后再前往沙箱开展实验。如您有任何问题，可随时通过工单或者服务热线（+86-4000-955-988 ）与我们联系。感谢您对华为云微认证的支持！发布日期：2025年6月24日

尊敬的客户：您好！为了更好地服务千万开发者，真正让开发者在云上作业，懂开发会开发，华为云会定期对开发者认证进行更新，《云技术精髓入门级开发者认证》已于2025年6月19日正式更新上线！更新升级点：原华为云GaussDB(for MySQL)服务名称更新为TaurusDB，本次更新依据上述服务名称的改变，更新《云数据库的选择与使用》章节中所涉及的相关内容，并同步刷新配套MOOC视频以及试题。届时我们还将开展相关开发者认证上新活动，详情请关注华为云开发者学堂论坛-热门活动 相关通知。发布日期：2025年6月19日

1、活动中奖名单公布如下，公示期10天（2025年8月12日-2025年8月21日），如有疑问请在公示期间反馈，逾期不予处理！2、请中奖用户于8月21日23:59:59点前填写【中奖用户收货信息收集表】，逾期未填写视为放弃奖励，请知悉！3、公示期结束后，30个工作日内统一邮寄奖品，如有问题请与论坛版主【开发者学堂欢欢】联系。其中本活动中的实物礼品如遇缺货等情况，将替换成类似款或其他等值礼品发放。昵称华为云账户名中奖奖品andyleungand**eung1005云宝盲盒czb_biuhid**etnfngjmhkpfh9开发者空间定制双肩包miyalianhid**f120xjdev_3p9p开发者空间定制双肩包yd_229106051hid**yun开发者空间定制双肩包yd_238637427hid**9mpca1c1cu9wfa云宝盲盒yd_238822659fj8**8w123开发者空间定制冲锋衣yd_248582828Lin**an开发者空间定制冲锋衣yd_267354561hid**2c7dpa7yyb8l77开发者空间定制冲锋衣yd_277766320luo**68华为智能体脂秤 3 yd_286761491mug**etan开发者空间定制冲锋衣yd_291175919hid**72l4dybdv6zg3j开发者空间定制双肩包给无眠点压力GT-**ixin_52568491云宝盲盒敲键盘的诗人gpf**dream开发者空间定制冲锋衣（中奖名单如上展示，如有疑问请及时反馈） 【活动时间】即日起—8月7日【活动流程】——领取奖励的用户必须完成活动报名+领取华为开发者空间一、活动报名二、免费领取华为开发者空间三、完成以下任意一个华为开发者空间案例，在本论坛贴评论区分享案例完成截图（完成时间+案例名称+案例完成截图+实验心得），活动结束后，将在评论区符合条件的用户中抽奖。指定空间案例实操入口云主机轻松部署DeepSeek点击进入云主机调用DeepSeek实现代码自动生成点击进入初识云主机：CodeArts IDE入门点击进入运用昇思MindSpore框架成为垃圾分类小能手点击进入仓颉 – C跨语言编程实现控制台小游戏点击进入基于鲲鹏服务器的打砖块小游戏部署点击进入【活动礼品】【空间案例实操方式】1、 登录个人华为账号，从上述指定开发者空间案例中选取任意1个，点击进入，下载自己感兴趣的案例，根据提示完成案例实操2、 本活动贴评论区：发送完成时间+案例名称+案例完成截图+实验心得【抽奖方式】活动结束后，我们将从参与活动的用户中（华为云新老用户均可参与），通过巨公平台或Excel 函数形式抽取获奖用户，并在本活动帖进行公示获奖用户。 用户限制说明:1、参加本次社区活动的用户必须为华为云注册用户。同时为保证活动公平性，禁止用户以IAM账号身份参与活动，否则将视为无效。2、领取奖品的用户需为华为云实名用户，未完成实名认证的用户将不发放活动奖励。3、本次活动如一个实名认证对应多个账号，只有一个账号可领取奖励。如在同一概率活动中，同一账号重复获奖，只发放首先获奖奖品。4、本次活动一个实名认证账号只能对应一个收件人，如同一账号填写多个不同收件人，不予发放奖励。5、请开发者不要在活动期间随意修改社区昵称和华为云账号，由此产生的统计问题，如过了申诉期，小助手不再处理。（申诉期为活动结果公示3天内。）奖品发放说明:1、本活动结束之后10个工作日内公示获奖信息，获奖开发者用户需在截止时间在获奖信息收集表中填写获奖信息，获奖信息截止收集日过后30个工作日内，将统一发出奖品。华为云遵守《中华人民共和国个人信息保护法》规定，将以上个人信息仅用于礼品发放之目的，不会向任何第三方披露。若由于获奖开发者用户自身原因（包括但不限于联系方式有误、身份不符或超过截止登记日期等）造成奖品无法发送，视为获奖开发者用户放弃领奖。2、为保证活动的公平公正，华为云有权对恶意刷活动资源（“恶意”是指为获取资源而异常注册账号等破坏活动公平性的行为），利用资源从事违法违规行为的开发者用户收回抽奖及奖励资格。3、若发放奖品时，出现库存不足，则优先发放等价值的其他实物奖品；开发者空间定制冲锋衣尺码随机发放，不指定尺码。4、所有参加本活动的开发者用户，均视为认可并同意遵守《华为云开发者用户协议》，包括以援引方式纳入《华为云开发者用户协议》、《可接受的使用政策》、《法律声明》、《隐私政策声明》、相关服务等级协议（SLA），以及华为云服务网站规定的其他协议和政策（统称为“云服务协议”）的约束。5、如果您不同意本活动规则和云服务协议的条款，请勿参加本活动。

尊敬的客户：您好！为保证您获得最佳的学习与实验体验，华为云开发者学堂于2025年6月13日对微认证《基于容器实现一分钟自动化部署》进行下线优化，预计2025年9月30日重新上线，届时请您关注华为云开发者学堂资讯专栏-业务动态通知。为此，我们将采取以下措施：1．对于已购买该微认证并通过考试领取证书的客户，原证书在证书有效期内仍有效，并与课程优化后的证书拥有同等效力；2．对于已购买该微认证但未通过考试且仍有考试机会的客户，可在重新上线后进行新课程学习、实验，参加考试并领取证书；3．对于已购买该微认证但未完成实验操作的客户，请您关注华为云开发者学堂资讯专栏-业务动态的上线通知，待实验优化重新上线后再前往沙箱开展实验。如您有任何问题，可随时通过工单或者服务热线（+86-4000-955-988 ）与我们联系。感谢您对华为云微认证的支持！发布日期：2025年6月13日

尊敬的客户：您好！为保证您获得最佳的学习与实验体验，华为云开发者学堂于2025年6月12日对微认证《基于鲲鹏架构的飞机大战游戏》进行下线优化，预计2025年9月30日重新上线，届时请您关注华为云开发者学堂资讯专栏-业务动态通知。为此，我们将采取以下措施：1．对于已购买该微认证并通过考试领取证书的客户，原证书在证书有效期内仍有效，并与课程优化后的证书拥有同等效力；2．对于已购买该微认证但未通过考试且仍有考试机会的客户，可在重新上线后进行新课程学习、实验，参加考试并领取证书；3．对于已购买该微认证但未完成实验操作的客户，请您关注华为云开发者学堂资讯专栏-业务动态的上线通知，待实验优化重新上线后再前往沙箱开展实验。如您有任何问题，可随时通过工单或者服务热线（+86-4000-955-988 ）与我们联系。感谢您对华为云微认证的支持！发布日期：2025年6月12日

直播主题：华为云GaussDB入门级认证 - 考试辅导直播时间：2025.06.13   16:00-17:30直播老师： Steven丨华为云学堂技术讲师直播链接：cid:link_0直播简介：本次直播为HCCDA-GaussDB认证考试提供全面辅导，旨在帮助学员深入了解数据库技术原理与应用实践。我们将详细解析考试大纲，分享高效备考策略，并讲解关键知识点，包括但不限于SQL操作介绍、GaussDB数据库管理与运维基础等。直播亮点：本期直播亮点：1、GuassDB入门级开发者认证题型分析2、GuassDB理论重点精讲以及模拟考题拆解3、GuassDB核心实操重点步骤梳理

直播主题：华为云鸿蒙应用入门级开发者认证- 考试辅导直播时间：2025.06.11   16:00-17:30直播老师： Skye丨华为云学堂技术讲师直播链接：cid:link_0直播简介：本次直播专为备考华为云鸿蒙应用入门级开发者认证的学员设计，提供系统的考试辅导。深度解析认证核心知识点，涵盖HarmonyOS介绍、应用开发入门、ArkTS语言、声明式开发范式组件、Stage应用模型、玩转服务卡片、鸿蒙应用网络请求开发、鸿蒙应用云函数调用等内容。直播亮点：1、鸿蒙应用开发认证题型分析2、鸿蒙应用开发核心理论概念精讲以及模拟考题拆解3、鸿蒙应用开发实操要点梳理

直播主题：云技术精髓入门级开发者认证- 考试辅导直播时间：2025.06.09   16:00-17:30直播老师： 大树丨华为云学堂技术讲师直播入口：cid:link_0直播简介：本次直播专为备考云技术精髓入门级开发者认证的学员设计，提供系统考试辅导。深度解析认证核心知识点，涵盖云技术基础概念、华为云基础设施与生态政策；详解计算、存储等华为云核心服务功能、场景及操作；解读云原生技术价值与转型要点。本期直播亮点：​1、云技术核心概念精讲与样题剖析​2、华为云基础产品功能解读及考题拆解​3、云原生技术要点梳理与认证题型分析