【获奖公示】一、活动获奖公示信息如下：（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、云原生技术要点梳理与认证题型分析

前言在当今这个数字经济时代，数据无疑成为企业最核心的资产之一。面对激烈的市场竞争，企业若想精准决策，迅速响应市场变化，就必须依赖高效的数据分析能力。数据分析不仅能够帮助企业洞察市场趋势、预测消费者行为，更可以优化运营效率，降低成本，提升企业的整体竞争力。举个简单的例子：以某服装电商为例：双十一前夕，数据分析师发现一个关键趋势——平台上’oversized卫衣’的搜索量在过去一周内激增了300%，但转化率却只有2.1%，远低于平均水平的5.8%。通过深入分析用户行为轨迹，团队发现问题所在：用户在商品详情页的平均停留时间只有15秒，且90%的用户会点击尺码表后直接关闭页面。进一步调研发现，消费者对’oversized’的具体版型理解存在偏差，担心买到的衣服过大或过小。基于这一洞察，企业立即调整策略：在商品页面增加真人试穿视频，标注模特身高体重及穿着尺码；优化尺码表，用’宽松度对比图’替代传统数据表格；针对搜索’oversized卫衣’的用户推送’30天无理由退换’的专属服务。结果在48小时内，该品类的转化率提升至6.2%，销量环比增长了180%。而没有及时响应这一数据信号的竞争对手，则错失了这波流量红利。这个成功案例的背后，反映出高效数据分析工作流的重要性。然而，在实际的数据分析实践中，许多企业和开发者仍面临着诸多挑战：从数据采集、清洗、建模到可视化展示，每个环节都可能成为效率瓶颈。如何构建一套流畅、智能的全链路数据分析工作流，正成为当下亟待解决的关键问题。本文将深入探讨华为开发者空间与DeepSeek的强强联合，如何助力打造全链路高效数据分析工作流，帮助开发者和企业在数据驱动的道路上走得更快、更稳。数据分析链路中存在哪些典型痛点？尽管数据分析的重要性已经得到普遍认可，但现实中企业在数据分析链路中仍面临诸多具体的挑战：首先，数据采集与处理难度大。企业的数据通常散落在各类系统和业务平台中，这些数据格式多样且杂乱无章，数据往往分散在ERP、CRM、电商平台、社交媒体、IoT设备等多个系统中，数据格式千差万别（结构化、半结构化、非结构化），数据标准不统一。获取这些数据需要对接各种API接口，配置复杂的ETL流程，整个过程耗时耗力。企业需要花费大量时间和精力去获取、清洗和整合数据，确保数据的准确性和一致性。这种繁琐而耗时的数据预处理环节严重拖慢了整体数据分析效率。其次，分析过程繁琐复杂，从数据预处理到建模分析，再到结果验证，往往需要使用多种不同的工具和平台（如Python、R、SQL、Tableau、PowerBI等），工具间切换频繁，学习成本高，协作效率低。析过程往往缺乏标准化流程和版本管理，同一个分析任务在不同时间、不同人员操作下可能产生不同结果，影响分析结果的可信度和可复现性。此外，数据分析人才短缺也是企业普遍面临的问题。纯技术背景的分析师往往缺乏对业务场景的深度理解，而业务专家又缺乏数据分析的技术能力，导致分析结果与业务需求之间存在较大鸿沟。专业的数据分析师招聘难度大、成本高，并且培养周期较长，企业难以快速组建起有效的数据分析团队，进一步加剧了数据分析能力的瓶颈。最后，分析结果的可视化程度较低。数据分析结果如果无法直观清晰地呈现给决策者，数据分析往往涉及IT、业务、管理等多个部门，缺乏统一的协作平台和标准化流程，沟通成本高，项目推进缓慢。以上痛点使得企业亟需更高效、更智能、更简单的数据分析解决方案。如何利用云计算、人工智能等新兴技术，打造一套全链路、低门槛、高效能的数据分析工作流，已成为当下数字化转型的关键课题。华为开发者空间与DeepSeek的技术基础华为开发者空间是面向全球开发者打造的专属空间，其技术基础具有多方面特性。它汇聚了鸿蒙、昇腾、鲲鹏、GaussDB、欧拉等各项根技术的开发资源及工具。在硬件资源层面，依托华为强大的技术实力，能为开发者提供性能强劲的云主机，保障开发过程中的计算需求，无论是复杂的模型训练，还是大规模的数据处理，都能高效运行。在开发工具方面，提供了一站式的开发环境。以 CodeArts IDE for Cangjie 编辑器为例，它不仅具备常用的语法高亮、错误诊断、自动补全等功能，可实时反馈并提升开发效率，还支持反向调试，方便开发者查看历史调试信息。并且，该编辑器被预置在云主机环境中，开发者开箱即用，极大降低了开发的前期准备成本。同时，开发者空间还集成了丰富的命令行工具，满足不同开发者多样化的开发习惯和复杂的开发需求。此外，华为开发者空间提供了从开发编码到应用调测的配套案例，结合其云上存储空间，形成了一个完整的开发闭环。开发者可以基于这些资源，在华为根技术生态下高效便捷地进行知识学习、技术体验以及应用创新。在利用华为开发者空间提供的云主机开发 DeepSeek 技术时，可从多维度实现两者的有机融合。华为开发者空间性能强劲的云主机，能为 DeepSeek 技术开发提供坚实的硬件支撑。预置的 CodeArts IDE for Cangjie 编辑器具备语法高亮、错误诊断、自动补全等功能，还支持反向调试，方便开发者对 DeepSeek 模型开发过程进行调试和优化。集成的丰富命令行工具，可满足 DeepSeek 开发中不同场景的需求。从开发闭环角度看，华为开发者空间从开发编码到应用调测的配套案例，结合云上存储空间，为 DeepSeek 技术开发提供了完整的流程支持。开发者可借助这些资源，在华为根技术生态下，更高效地进行 DeepSeek 模型的知识学习、技术体验以及应用创新。同时，DeepSeek 完全开源 R1 等模型、采用 MIT 协议的开源策略，与华为开发者空间的开放生态相契合，开发者可在云主机环境中自由定制和二次开发 DeepSeek 模型，进一步推动技术共享与协作。开放的 API 定价优势，也能让更多中小企业在华为开发者空间云主机上开发和应用 DeepSeek 技术，降低进入 AI 领域的门槛，促进 DeepSeek 在各行业的广泛应用。全链路高效数据分析工作流搭建与应用实践1.配置云主机首先进入到开发者空间，之后点击工作台，开始配置云主机：配置相应的云主机：以上系统模拟我们开发生产环境的主要机器，之后我们来开始一步步搭建，首先进入到云开发空间里面，打开终端界面，输入curl -fsSL https://dtse-mirrors.obs.cn-north-4.myhuaweicloud.com/case/0035/install.sh | sudo bash2.下载ollama先下载ollama下载完毕之后我们可以借助 Ollama 工具来部署 Deepseek 大模型，部署 deepseek-r1:1.5b 版本，如果硬件支撑可以部署更高效的模型，执行命令：ollama run deepseek-r1:1.5b以上我们就部署完了，可以尝试输入prompt来测试效果：通过命令：sudo netstat -tunlp可以查看olloama开放的本地端口：3.CodeArt IDE for Python那么接下来我们可以打开CodeArt IDE for Python，对端口进行通信，完成这一步之后我们可以开始尝试构建智能体：我们知道ollama serve默认监听地址为 http://localhost:11434，首先下载requests库：pip install requests我们可以通过Python代码尝试通信：import requests def chat_with_ollama(prompt, model="deepseek-r1:1.5b"): url = "http://localhost:11434/api/generate" headers = { "Content-Type": "application/json" } data = { "model": model, "prompt": prompt, "stream": False # 关闭流式返回，适合简单测试 } try: response = requests.post(url, json=data, headers=headers) response.raise_for_status() result = response.json() print("模型回复：", result["response"]) except requests.exceptions.RequestException as e: print("❌ 请求出错：", e) except Exception as e: print("❌ 其他错误：", e) if __name__ == "__main__": test_prompt = "你好" chat_with_ollama(test_prompt) 如何能获取到大模型输出，说明我们之前的过程都没有问题：4.工作流搭建.那么我们下一步就可以开始全链路高效数据分析工作流搭建了：4.数据获取/清洗/提取首先我们可以思考数据获取层，这方面可以是企业存储数据和历史数据，也可以爬取互联网上相关信息数据，这方面不需要AI介入，而数据清洗和提取就可以通过大模型完美替代，而且清洗提取特别高效。我们可以将这个 Agent 封装为一个 Python 类，支持如下功能：功能说明clean_text(text)文本清洗：去除空格、标点符号、HTML、无用词等extract_fields(text, instruction)利用大模型提取指定结构信息（如姓名、地址、公司名等）batch_process(data_list)支持批量数据清洗和字段抽取custom_prompt(data, task)通过自定义任务 prompt 构造更复杂的清洗与抽取任务具体代码可以如下编写：import requests import re class DataCleaningAgent: def __init__(self, model_name="deepseek-r1:1.5b", host="http://localhost:11434"): self.model = model_name self.api_url = f"{host}/api/generate" def _call_model(self, prompt, stream=False): payload = { "model": self.model, "prompt": prompt, "stream": stream } try: response = requests.post(self.api_url, json=payload) response.raise_for_status() result = response.json() return result.get("response", "").strip() except Exception as e: print("调用失败：", e) return "" def clean_text(self, text): """ 基础清洗：去除HTML标签、特殊字符、重复空格等 """ text = re.sub(r"<.*?>", "", text) # 去HTML text = re.sub(r"\s+", " ", text) # 合并空格 text = re.sub(r"[^\w\s\u4e00-\u9fff]", "", text) # 去除特殊符号 return text.strip() def extract_fields(self, text, instruction="请从中提取所有公司名称和联系人"): """ 调用LLM进行字段提取 """ prompt = f"以下是原始数据：\n{text}\n\n{instruction}" return self._call_model(prompt) def batch_process(self, data_list, instruction): results = [] for i, text in enumerate(data_list): print(f"处理第{i+1}条数据...") cleaned = self.clean_text(text) extracted = self.extract_fields(cleaned, instruction) results.append({ "original": text, "cleaned": cleaned, "extracted": extracted }) return results # 示例调用 if __name__ == "__main__": agent = DataCleaningAgent() sample_data = [ "联系人：张三，联系电话：123456789，公司：江西省招标有限公司", "地址：南昌市东湖区，北京华为技术有限公司，联系人王五" ] instruction = "请提取所有公司名称和联系人姓名，返回JSON格式" results = agent.batch_process(sample_data, instruction) for res in results: print("\n原始：", res["original"]) print("清洗后：", res["cleaned"]) print("提取信息：", res["extracted"]) 大家可自行验证：4.2 数据库读取SQL清洗完毕之后我们将数据转入库中即可，那么这是建立在我们需要收集外部数据的情况下设置的数据清洗和提取智能体，大部分企业是有存储自己业务数据的，一般直接放置在数据库中。但是获取这部分数据需要比较繁琐的步骤，一般来说数据分析师通过编写SQL获取，但也有不是数据分析师的客户或者是企业管理者想直接拿到数据分析，为了降低数据分析的门槛，我们可以直接设定一个获取数据的智能体，返回给我们基础数据。简而言之就是构建一个自然语言 → 数据库查询 → 自动执行SQL → 返回结果的智能体系统，为企业管理者、运营、产品等非技术用户提供低门槛的数据访问能力。需要注意的是这对我们的Prompt以及知识库要求较高，这里需要我们根据企业业务数据库去制作一份对应的数据字典，也就是说这份数据字典是包含所有业务数据字段的解释和各个表模块的介绍。这里给出一部分的Prompt示例:请根据下方表结构，从表 `sales_data` 中查询 2024 年每个月的销售总额和增长率： 表结构： CREATE TABLE sales_data ( id INT, sale_date DATE, amount DECIMAL, region VARCHAR(50) ) 问题： 2024年每个月的销售额是多少？同比去年增长了多少？ 我们可以先解决需要代码编写的部分，比如执行层Python 实现数据库访问和结果返回：from sqlalchemy import create_engine import pandas as pd def execute_sql(sql: str, db_url: str): engine = create_engine(db_url) with engine.connect() as conn: df = pd.read_sql_query(sql, conn) return df 组件含义sqlalchemy.create_engine创建数据库连接对象，支持多种数据库（如 MySQL、PostgreSQL、SQLite）db_url数据库连接字符串，格式示例： mysql+pymysql://user:password@host:port/dbnamepd.read_sql_query(sql, conn)使用 pandas 执行 SQL 语句并将结果返回为 DataFrameconn数据库连接上下文，自动管理连接释放返回值 df查询结果组成的 DataFrame，可用于打印、导出、图表分析等调用实例如下：sql = "SELECT region, SUM(amount) as total FROM sales_data GROUP BY region" db_url = "mysql+pymysql://root:123456@localhost:3306/mydb" df = execute_sql(sql, db_url) print(df) 之后我们可以封装DeepSeek大模型，封装成一个DataQueryAgent，示例：class DataQueryAgent: def __init__(self, db_url: str, model_host="http://localhost:11434", model="deepseek-r1:1.5b"): ... def _generate_sql(self, question: str, schema_hint: str = ""): ... def query(self, question: str, schema_hint: str = ""): ... 设定面向用户的“自然语言问数据库”的封装类，封装了SQL生成 + 执行 两个流程。__init__ 方法比较简单定义：def __init__(self, db_url: str, model_host="http://localhost:11434", model="deepseek-r1:1.5b"): self.db_url = db_url self.model_host = model_host self.model = model _generate_sql() 方法就是完成功能（自然语言 → SQL）def _generate_sql(self, question: str, schema_hint: str = ""): prompt = f"根据以下数据库结构：\n{schema_hint}\n请生成对应SQL语句来回答问题：{question}" response = requests.post(f"{self.model_host}/api/generate", json={ "model": self.model, "prompt": prompt, "stream": False }) return response.json().get("response", "") 用本地部署的 DeepSeek 模型，将自然语言问题 + 数据库结构作为 Prompt，发送请求后解析返回结果为 SQL。query() 方法完成整个工作流（完整工作流：问题 → SQL → 查询结果）def query(self, question: str, schema_hint: str = ""): sql = self._generate_sql(question, schema_hint) print("生成的SQL：", sql) try: engine = create_engine(self.db_url) with engine.connect() as conn: df = pd.read_sql_query(sql, conn) return df except Exception as e: return f"SQL 执行失败: {e}" 示例调用：agent = DataQueryAgent( db_url="mysql+pymysql://root:password@localhost:3306/mydb" ) schema_hint = """ CREATE TABLE orders ( id INT, customer_name VARCHAR(50), amount DECIMAL(10,2), order_date DATE, status VARCHAR(20) ); """ question = "查询今年每个月的订单总金额" df = agent.query(question, schema_hint) print(df) 这样我们就得到一个完整的dataframe，之后我们就可以很轻松对其进行数据可视化了。4.3 DataFrame 可视化智能体（Visualization Agent）也可以将DataFrame 交给大模型智能体，让它根据数据内容决定可视化方式，并自动生成图表代码（如用 matplotlib / plotly / seaborn）。这种做法本质上就是构建一个 DataFrame 可视化智能体（Visualization Agent），它能够结合数据结构、字段含义与任务需求，为用户自动设计图表并呈现。具体实现逻辑很简单：DataFrame → LLM → 图表自动生成。构建可视化智能体类：import requests import pandas as pd import contextlib import io import traceback class VisualizationAgent: def __init__(self, model_host="http://localhost:11434", model="deepseek-r1:1.5b"): self.model_host = model_host self.model = model def _generate_plot_code(self, df: pd.DataFrame, task_description: str): # 将 df 转为 CSV 方便 LLM 解析 csv_data = df.to_csv(index=False) prompt = f""" 你是一个数据分析专家，以下是CSV格式的数据，以及用户的可视化需求。请根据这些数据生成 Python 绘图代码，使用 matplotlib（优先）或 plotly。 用户需求：{task_description} CSV数据如下：{csv_data} 请返回完整可运行的 Python 绘图代码： """ response = requests.post(f"{self.model_host}/api/generate", json={ "model": self.model, "prompt": prompt, "stream": False }) return response.json().get("response", "") def _safe_exec(self, code: str, local_vars=None): """ 安全执行模型生成的代码 """ local_vars = local_vars or {} global_vars = {"__builtins__": __builtins__, "pd": pd} try: # 重定向输出，避免污染控制台 with contextlib.redirect_stdout(io.StringIO()) as f: exec(code, global_vars, local_vars) except Exception as e: print("⚠️ 代码执行错误：") traceback.print_exc() def visualize(self, df: pd.DataFrame, task_description: str = "绘制销售额柱状图", show_code: bool = True): code = self._generate_plot_code(df, task_description) if show_code: print("模型生成绘图代码：\n") print(code) print("\n正在执行绘图代码...") self._safe_exec(code, local_vars={"df": df}) 使用示例if __name__ == "__main__": df = pd.DataFrame({ "region": ["华东", "华南", "华北", "西南"], "sales": [120000, 95000, 78000, 67000] }) agent = VisualizationAgent() agent.visualize(df, task_description="请生成一个地区销售额柱状图") 当运行这段代码后，它将调用模型生成图表代码，并自动执行、展示图表。5.总结在本章中，我们把“想法”一步步落地成了一条 端到端的智能数据分析流水线：环境就绪——先在云开发空间上快速拉起一台高配云主机，借助 curl install.sh 一键初始化开发环境，再用 Ollama 把 Deepseek-r1:1.5b 模型部署好并通过 netstat 确认端口；模型打通——用最简洁的 requests 脚本验证模型能正常对话，确保后面所有智能体都能调用；数据获取 & 清洗——编写 DataCleaningAgent，让大模型接管繁琐的文本去噪、实体抽取，批量输出结构化 JSON；自然语言查库——通过 DataQueryAgent 把 “一句中文 → SQL → DataFrame” 自动串起来，让非技术同事也能秒查运营数据；智能可视化——把查询得到的 DataFrame 交给 VisualizationAgent，由大模型自动产出 Matplotlib/Plotly 代码并即时执行，图表立现；全链路闭环——至此，数据从 获取 → 清洗 → 入库 → 查询 → 展示 全程自动化完成，真正实现了“低门槛、高效率、可复用”的企业级数据分析工作流。关键收益零 SQL / 零绘图门槛：业务人员只需说人话，背后全部自动完成。高度可插拔：每个 Agent 都是独立模块，清洗、查询、可视化可按需组合或替换更强模型。私有化可控：全部部署在企业自己的云主机与数据库内，安全合规。企业实战案例分享基于以上我们全链路的数据分析工作流的搭建，我们已经有足够的能力来实践一套供应链周报自动生成系统。一句话目标：让供应链经理把“写周报”这件事，彻底交给代码和大模型——数据自己来、图表自己画、结论自己写，人只需要点头确认。1. 背景与痛点典型痛点传统做法带来的问题数据源多、格式乱手动从 ERP / WMS / OMS 导出 Excel粗糙拼表、易漏字段，费时费神维度多、指标杂人肉写 SQL 拉数据运营/采购不懂 SQL，严重依赖数据团队图表重复劳动“复制数据 → 贴进模板 → 调样式”一键回车变成一上午文字结论经理凭经验手写主观、易漏看趋势结果：一个 10 页左右的供应链周报，往往要 3-4 人、费时半天才能端出来。2. 解决思路 ：全链路工作流上场3. 核心模块落地3.1 数据清洗 & 入库接入源：ERP（采购、到货）、WMS（库存）、OMS（销售订单）工具：DataCleaningAgent任务：消歧同品名、规格（“A4纸80g” = “A4-80g”）解析自然语言备注，提取 供应商投诉 字段输出标准化 CSV，按天入 supply_dw 数据仓库3.2 自然语言取数使用者：采购经理、供应链分析师典型问句“近 12 周每周的采购金额与去年同期对比”“本周供应商到货准时率前 5 & 后 5”流程DataQueryAgent 读取 数据字典（自动生成或手写）Deepseek 模型根据问句 + Schema → 生成 SQL执行后返回 DataFrame示例快速验证：question = "列出最近4周库存周转天数趋势" schema_hint = "TABLE inv_summary(id, item, turnover_days, stat_week)" df_turnover = agent.query(question, schema_hint) 3.3 自动可视化 + 结论生成VisualizationAgent：DataFrame → 图表代码 → 即时渲染支持多图联排、双轴折线、漏斗、热力地图结论生成：再次调用 Deepseek：prompt = f"下面是近12周库存周转天数表格，请用200字生成总结性分析：\n{df.to_markdown()}" summary = llm(prompt) 3.4 报告编排 & 交付ReportBot（Python）：收集所有图 PNG + 文字结论用 python-pptx 拼成周报 Deck，或 WeasyPrint 导出 PDF调用企业 IM / 邮件 API：“Hi @Team，本周周报已生成 👉 链接/附件”调度：Crontab / Airflow 每周一早 8 点自动跑4.关键代码骨架一览def weekly_report(): # 1. 拉取核心指标 metrics = { "采购金额": query_agent.query("近4周采购金额趋势", sales_schema), "到货准时率": query_agent.query("近4周到货准时率", inbound_schema), "库存周转": query_agent.query("近12周库存周转天数", inv_schema) } # 2. 生成图表 charts = {} for k, df in metrics.items(): charts[k] = viz_agent.visualize(df, f"绘制{k}折线图", show_code=False) # 3. AI 书写摘要 summary = llm("请基于以上三张表和图，总结供应链本周主要问题及建议") # 4. 组装 PPT ppt = build_ppt(charts, summary) ppt.save("weekly_supply_report.pptx") # 5. 送达 send_mail("weekly_supply_report.pptx", to=["boss@corp.com"]) 5. 效果速览指标传统方式全链路自动化周报产出时长≈ 4 小时 / 人≤ 10 分钟 / 机器参与人数运营 + 数据 + 设计0（无人值守）可追溯性低（手改 Excel）高（SQL + Git + 日志）结论一致性人工主观AI 根据数据重算6. 实施要点 & 踩坑提示数据字典必须够详细字段含义、单位、取值范围，一定要让大模型“看得懂”。SQL 安全在执行层加白名单，只允许 SELECT。图表代码安全执行VisualizationAgent 内部要过滤 os, subprocess 等危险关键字。权限管理建议给每个调用者分配专属 API Key + 查询范围。缓存/增量机制周报通常是“滚动周”，可以把前 N-1 周的数据缓存，速度更快。7. 价值回顾“让系统写报告，人来做决策。”效率：周报自动产出，数据团队有更多时间做深度分析。准确：口径统一、自动核对，避免手抄表格出错。透明：从原始数据到图表代码全链路留痕，可追溯可复盘。可扩展：同样的框架可复制到财务、营销、客服，形成企业级“智能报告工厂”。下一步，如果你想把这套系统推广到其他业务线，只需要：补充新表的 Schema；编写对应的 KPI Prompt；在调度器里新增定时任务。实践建议与关键技术要点总结大模型优化方法众多（如剪枝、量化、算子融合等），但没有 “通用最优解”。选择优化策略时，需结合业务需求、硬件条件、成本等多维度因素，避免盲目追求技术先进性而忽视实际效果。1应用场景导向应用场景导向的核心逻辑是不同场景对模型性能的需求不同，优化目标需与场景匹配。实时交互场景（如聊天机器人、自动驾驶）：需低延迟，优先选择模型量化、算子融合或模型蒸馏。批量处理场景（如离线数据分析、大规模预测）：需高吞吐量，可采用模型并行或动态批处理。边缘部署场景（如手机、物联网设备）：受限于算力和能耗，需模型压缩+ 轻量级架构设计。若部署一个需要实时响应的智能家居语音助手，即使模型精度略低，也应优先选择量化和剪枝，以确保推理速度。2成本与性能平衡成本与性能平衡的核心逻辑是不同硬件对优化方法的支持程度不同，需充分发挥硬件优势。GPU/TPU 等加速硬件适合算子融合、模型并行等需要高并行计算的优化方法。边缘设备则需模型轻量化（剪枝、量化）+ 轻量级推理框架（如 TensorFlow Lite），避免复杂计算。CPU 服务器可选择多线程优化或内存优化（如 onnxruntime 的 CPU 优化）。比如在 GPU 集群上部署大模型时，使用 TensorRT 进行算子融合和图优化，能显著提升推理速度；而在手机端部署时，使用量化后的模型和 MNN 框架更合适。3. 实际应用中的综合决策在实际部署中，需结合多个原则进行权衡。比如自动驾驶公司需在车载设备（边缘硬件）上部署实时目标检测模型：场景导向：选择低延迟优化，如模型剪枝和量化。硬件匹配：针对车载芯片的架构，使用专用优化工具（如 Nvidia 的 TensorRT 针对 GPU 优化）。成本平衡：通过蒸馏轻量化模型，降低对高端硬件的依赖，控制成本。扩展性：设计可动态调整的流水线，未来若更换传感器或算法，可快速适配。这四个原则为大模型优化提供了从场景需求到技术落地的完整思考路径：以场景为起点，结合硬件和成本约束，选择灵活可扩展的方案。通过这种系统性的决策，企业既能在当下实现高效部署，又能为未来的业务增长预留技术空间。智能体编排未来应用趋势随着大模型从“一个聪明的回答机”逐步进化为“面向任务的执行体”，多智能体编排（Agent Orchestration） 正成为新一轮 AI 应用变革的核心引擎。相比传统单模型调用逻辑，智能体编排更像是构建“具身智能系统”或“数字员工团队”，将感知、思考、决策与执行完整串联起来，实现从“回答你”到“替你做”的飞跃。在我们展示的数据分析系统中，其实已经具备“雏形智能体”的要素：如 DataCleaningAgent、DataQueryAgent、VisualizationAgent 等等。这些模块虽然是按功能划分，但本质上已经具备“接收任务 → 理解意图 → 执行操作 → 返回结果”的 Agent 行为模式。而在更复杂的未来业务中，一个任务往往无法由单个智能体完成，这时需要：多个 Agent 分工合作、传递信息、联合完成复杂任务，甚至具备自治能力。比如在一个“企业经营分析”智能系统中：角色智能体职责说明任务协调 Agent（Coordinator）理解高层命令，如“生成月度经营报告”，并拆解成子任务数据智能体（Query Agent）根据需求调用数据库并返回 DataFrame图表智能体（Viz Agent）生成图表代码并渲染结果洞察智能体（Insight Agent）结合数据输出摘要分析与趋势判断报告生成 Agent汇总图表与文字，输出 Markdown/PPT/邮件等格式监控反馈 Agent记录执行日志、收集用户反馈并优化流程每一个 Agent 都是可复用的能力单元，而智能体间的协作流转由Agent Framework 或 编排平台调度控制。要真正落地多智能体协同系统，以下几个能力将成为技术团队不可回避的核心能力：意图解析能力需要将用户自然语言转化为结构化任务链（如 AWEL DSL、JSON Graph 等）；任务拆解与分发能力使用树状或 DAG 结构，支持任务自动拆解与并发执行；智能体能力注册与调度机制类似“插件系统”，支持注册各类智能体，并调度调用；记忆与上下文管理实现智能体在多轮协作中共享上下文、传递中间变量；异常处理与回滚机制构建面向业务场景的“容错能力”，支持失败重试、断点续跑；安全与权限机制管理智能体访问数据的边界，防止越权调用或数据泄漏；未来的 AI，不再只是一个“回答者”，而是一个主动工作的“数字合作者”。从数据分析到知识总结，从任务执行到流程协同，我们已经看到多智能体系统开始扮演团队中的新角色。“让系统成为员工的一部分，而不仅仅是工具。”这不仅是数据智能化的进化方向，更是企业迈向新一代智能生产力时代的起点。下一阶段，智能体将不再只是调用函数的代理，而是逐步演化为具备推理、感知、社交能力的数字个体，真正融入业务、参与决策、持续学习与成长。我正在参加【案例共创】第4期 基于华为开发者空间+仓颉/DeepSeek/MCP完成应用构建开发实践 https://bbs.huaweicloud.com/forum/thread-02127182415062274055-1-1.html

尊敬的客户：您好！为了确保人工智能入门级开发者认证课程的时效性和技术领先性，帮助开发者更好地掌握最新的人工智能开发技能，同时紧跟华为云的最新技术动态，我们将对认证内容进行更新，预计2025年6月30日正式更新上线！ 更新内容如下：1. 新增前沿大模型介绍，让初学AI的开发者对大模型有初步认识；2. 新增针对大模型的应用技术MCP，从其工作原理、操作流程讲起，重点突出使用MCP能带来的效率提升、成本降低等优势，使开发者明白其价值所在；3. 新增基于RAG技术的交通大模型应用作为实践案例，详细剖析其在实际交通场景中的应用方式、取得的成效以及解决的问题，让开发者直观感受RAG技术的强大。届时我们还将开展相关开发者认证上新活动，详情请关注华为云开发者学堂论坛-热门活动 相关通知。发布日期：2025年6月4日

尊敬的客户：您好！为了确保人工智能工作级开发者认证课程的时效性和技术领先性，帮助开发者更好地掌握最新的人工智能开发技能，同时紧跟华为云的最新技术动态，我们将对认证内容进行全面升级，预计2025年8月22日正式更新上线！ 更新内容如下：1、课程逻辑重构：以大模型开发应用为主线，包含数据处理分析、大模型理论基础、基于大模型的AI应用开发、大模型优化等；同时精简机器学习/深度学习基础内容，移除EI服务介绍及API调用实操内容，聚焦大模型能力培养与认证；2、新增Transformer与大模型基础、大模型进阶与模型开发、大模型应用开发三大模块，涵盖核心技术：Transformer原理、多模态大模型（CLIP/Stable Diffusion）、强化学习与大模型训练（RLHF）、大模型评测方法等，夯实大模型核心技术基础；3、新增数据搜集、处理与分析，强调数据质量对模型的影响；4、新增AI伦理与未来趋势、具身智能等前沿议题；5、新增配套实验：Bert/GPT文本理解与生成实验，大模型微调与评测，Prompt工程构建聊天机器人，基于RAG的问答机器人开发，智能Agent开发等，强化大模型实操技能训练；6、扩展开发工具链：引入ollama、LlamaIndex、LangChain、Dify等新一代AI应用开发工具，替代了原认证中的ModelArts Pro套件，贴近开发者最新典型使用场景。 届时我们还将开展相关开发者认证上新活动，详情请关注华为云开发者学堂论坛-热门活动 相关通知。发布日期：2025年6月4日

尊敬的客户：您好！为了确保云迁移工作级开发者认证课程的时效性和技术领先性，帮助开发者更好地掌握最新的云迁移技能，同时紧跟华为云的最新技术动态，我们将对认证内容进行全面升级，预计2025年8月30日正式更新上线！一、新增内容1、新增Landing Zone设计和实施章节及配套实验，涵盖需求调研、方案实设计与实施等内容，使开发者掌握Landing Zone的设计和实施的关键要素；2、新增云服务器迁移实验：存储迁移、缓存迁移，增强开发者对于存储、缓存迁移的实操能力；3、新增IMS介绍与操作指导、键值存储KVS、kafka迁移工具介绍和工具迁移方案、 MGC介绍等内容，使开发者对计算、存储、消息队列等云服务及迁移工具有更为全面的掌握。二、更新内容1、云迁移方法论：方法论更新为云采用框架，包括迁移的战略制定与顶层规划、迁移调研分析和评估规划、迁移上云方案设计和试点规划、迁移上云实施和运维；2、云迁移实施案例：更新多个企业的迁移案例；3、云服务产品规格、实验界面：根据最新产品版本更新；4、移除原认证中“华为云核心服务”章节，使内容更聚焦于迁移技能；5、优化原有教材的逻辑结构：数据库服务器使用和迁移、容器服务使用和迁移、容器迁移实验等内容，去掉冗余内容，调整呈现结构，优化开发者学习体验。届时我们还将开展相关开发者认证上新活动，详情请关注华为云开发者学堂论坛-热门活动 相关通知。发布日期：2025年6月4日