B站影视

首页 > 资讯信息 > 内地电影

华为开发者空间，基于仓颉与DeepSeek的MCP智能膳食助手

B站影视 内地电影 2025-10-27 15:02 2

摘要：MCP，全称Model Context Protocol，中文叫“模型上下文协议”。你可以把它想象成AI的“USB 接口” --让不同的AI模型、工具和应用程序能用统一的方式交流。那么我的理解是：它更像是一个适配器来调节各种AI不同的接口达到一致的效果，让AI

本案例由开发者：给无眠点压力提供

最新案例动态，请查阅《【案例共创】华为开发者空间，基于仓颉与DeepSeek的MCP智能膳食助手》「链接」。小伙伴快来领取华为开发者空间进行实操吧

MCP，全称Model Context Protocol，中文叫“模型上下文协议”。你可以把它想象成AI的“USB 接口” --让不同的AI模型、工具和应用程序能用统一的方式交流。那么我的理解是：它更像是一个适配器来调节各种AI不同的接口达到一致的效果，让AI的交流更加简单，即使没有身份预设，走MCP是完美的让AI成为你的最佳助手。

随着人们对健康饮食关注度的提升，越来越多用户希望借助AI助手实现个性化的饮食分析与管理。然而，目前市面上的饮食类应用普遍存在如下痛点：

缺乏智能分析：大多数仅记录卡路里，无法提供专业点评与优化建议；知识更新不及时：难以结合最新营养研究进行推荐与判断；缺乏可扩展性：难以适配特定人群（如高血压、糖尿病、健身人群等）的差异需求。

本项目旨在构建一个基于大语言模型（DeepSeek）和结构化协议（MCP）的智能饮食健康助手，通过自然语言交互，帮助用户实现饮食数据结构化、健康风险识别与个性化建议生成。

说明：

登录华为开发者空间工作台，领取云主机。在云主机登录ModelArts Studio(MaaS)控制台，领取DeepSeek-V3百万免费Tokens。华为开发者空间 - 云主机 桌面打开CodeArts IDE for Python构建本地MCP服务项目，实现饮食语义解析、食物识别、健康标签打标、Prompt生成与请求路由。华为开发者空间 - 云主机 桌面打开CodeArts IDE for Cangjie构建本地仓颉AI机器人，接收经过标准化构造的请求Prompt，返回结构化营养建议、健康分析与饮食优化建议。

本案例预计花费0元。

资源名称规格单价（元）时长（分钟）

面向广大开发者群体，华为开发者空间提供一个随时访问的“开发桌面云主机”、丰富的“预配置工具集合”和灵活使用的“场景化资源池”，开发者开箱即用，快速体验华为根技术和资源。

如果还没有领取云主机进入工作台界面后点击配置云主机，选择Ubuntu操作系统。

进入华为开发者空间工作台界面，点击打开云主机 > 进入桌面连接云主机。

华为云提供了单模型200万免费Tokens，包含DeepSeek-V3满血版等，我们可以登录华为云ModelArts Studio(MaaS)控制台领取免费额度，这里我们选择DeepSeek-R1满血版来搭建我们的专属AI聊天机器人。

在云主机桌面底部菜单栏，点击打开火狐浏览器。用火狐浏览器访问ModelArts Studio首页：https://www.huaweicloud.com/product/modelarts/studio.html，点击ModelArts Studio控制台跳转到登录界面，按照登录界面提示登录，即可进入ModelArts Studio控制台。

根据系统提示签署免责声明。

进入ModelArts Studio控制台首页，区域选择西南-贵阳一，在左侧菜单栏，选择模型推理 > 在线推理 > 预置服务 > 免费服务，选择DeepSeek-V3-32K模型，点击领取额度，领取200万免费token。

领取后点击调用说明，可以获取到对应的API地址、模型名称。

API Key管理，进入API Key管理界面。点击右上角的创建API Key，编辑标签和描述，点击确定。

点击右侧复按钮，将密钥复制保存到本地。

注：API Key仅会在新建后显示一次，若API Key丢失，需要新建API Key。

通过本节操作，我们在ModelArts Studio控制台获取到三个关键数据：API地址、模型名称和API Key。

本案例采用本地MCP服务 + 本地仓颉AI机器人对话。本章节讲解如何构建本地MCP服务。

在云主机桌面打开CodeArts IDE for Python。

在新打开的CodeArts IDE for Python的提示界面或通过文件 > 新建 > 工程，打开新建工程配置界面。

在新建工程配置界面，编辑项目名称为food_mcp，然后点击创建。

在CodeArts IDE for Python左侧资源管理器 > food_mcp工程右侧的新建文件按钮，并将文件命名为mcp_server.py，下一步在此文件中开始编写代码。

依次导入FastAPI框架构建Web服务、使用Pydantic定义数据模型、导入JSON处理模块、导入类型注解工具、导入流式响应支持、导入DeepSeek API客户端（流式/异步流式）、导入食物文本匹配工具、导入日志工具、导入带TTL的缓存系统、导入系统模块（日志配置）。

# -*- coding: utf-8 -*-# mcp_server.py"""基于仓颉 + DeepSeek + MCP 的智能膳食分析助手"""from fastapi import FastAPIfrom pydantic import BaseModelimport jsonfrom typing import List, Dict, Anyfrom fastapi.responses import StreamingResponsefrom deepseek_client import stream_deepseek, async_stream_deepseekfrom text_match import extract_food_simple, extract_foodfrom loguru import loggerfrom cachetools import TTLCacheimport sys

从food_tags.json文件加载食物标签数据，创建全局字典FOOD_TAGS存储食物属性：

营养标签 (tags)副作用/禁忌 (effects)相克食物 (avoid_with)饮食类型 (diet_type)# 加载食物字典with open("food_tags.json", encoding="utf-8") as f: FOOD_TAGS: Dict[str, Dict[str, Any]] = json.load(f)

创建FastAPI应用实例：定义输入数据模型（仅含 input 字符串字段），初始化全局缓存（500条容量，30分钟过期），配置日志系统（输出到控制台，INFO级别）。

app = FastAPIclass Input(BaseModel): input: str# 全局缓存：prompt -> full_reply_textCACHE = TTLCache(maxsize=500, ttl=1800) # 30分钟自动过期logger.removelogger.add(sys.stdout, level="INFO", enqueue=False, backtrace=False)

定义营养师基础角色设定；创建多场景提示词模板：

nutrition_review：常规饮食分析；diet_plan：减脂期食谱生成（含热量表格）；effects_inquiry：食物过量风险分析；avoid_inquiry：食物相克关系分析；

统一要求输出包含5个核心部分：总体评价、推荐摄入、过量风险、不宜同食、行动建议。

# ---- 基础角色词 ----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}" ),}食物提取：识别输入中的食物名称，区分精确匹配和模糊匹配结果，记录匹配日志。信息聚合：聚合所有匹配食物的属性标签，收集副作用信息，提取相克食物列表，确定饮食类型。场景模式识别：基于关键词检测用户意图，自动路由到合适的处理场景。提示词构建：动态生成场景化提示词，注入食物属性作为关键约束，要求Markdown列表格式输出。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, }接收用户输入文本；返回结构化分析结果（不含AI生成内容）：识别出的食物列表、营养标签、副作用信息、相克食物、场景模式、构建的提示词。@app.post("/mcp")async def route_prompt(data: Input): user_input = data.input result = _process_input(user_input) return result

实现Server-Sent Events (SSE)流式响应，四阶段事件流：

meta 事件：发送食物分析元数据；token 事件：流式传输AI生成内容；智能缓存：相同提示词30分钟内直接返回缓存；done 事件：标记响应结束；

支持实时显示AI生成过程，优化重复请求响应速度。

@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")

Ctrl + S键保存代码。

注： 此时工程中会报错，这是因为开发环境中缺少必要的包文件，此处无须理会，我们将在后续步骤“8.1 编写配置文件requirements.txt,配置依赖包”中解决依赖包的问题。

在CodeArts IDE for Python左侧资源管理器 > food_mcp工程右侧的新建文件按钮，并将文件命名为deepseek_client.py，下一步在此文件中开始编写代码。

# deepseek_client.py# 基于 DeepSeek API 的客户端import requestsimport jsonfrom typing import Generatorimport httpx

从config.json文件加载API配置，获取DeepSeek API密钥、端点URL、获取当前使用的模型名称。提供集中配置管理，便于维护和变更。

# 从 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"]实现同步流式API调用，通过stream=True参数启用流式传输；使用response.iter_lines逐行处理服务器推送事件(SSE)；解析data:开头的有效数据块；过滤结束标记[DONE]；从JSON中提取增量内容(delta)；使用生成器逐步返回文本片段；支持实时显示生成过程。# 新增：流式推理生成器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实现异步流式API调用，使用httpx库替代requests实现异步；通过AsyncClient和client.stream处理异步流；使用resp.aiter_lines异步迭代响应行；解析逻辑与同步版本一致；适用于FastAPI等异步框架，支持高并发场景；设置timeout=None防止长文本生成超时。

参数配置：指定使用的AI模型，构造对话消息结构，通过stream=True启用流式传输，设置生成token上限（防止过长响应），请求包含用量统计信息（可选）
核心处理逻辑：识别有效的SSE数据行（以 data: 开头），跳过空行和元数据行，处理流结束信号[DONE]，解析JSON格式的增量数据，提取choices[0].delta.content 中的文本片段，过滤空内容片段，异常处理确保解析失败时不中断流程。

# 异步流式生成器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

Ctrl + S键保存代码。

在CodeArts IDE for Python左侧资源管理器 > food_mcp工程右侧的新建文件按钮，并将文件命名为test_pipeline.py，下一步在此文件中开始编写代码。

# -*- coding: utf-8 -*-# test_pipeline.pyimport requestsfrom deepseek_client import call_deepseekimport jsonfrom requests.exceptions import ChunkedEncodingError

测试目的：验证常规饮食分析场景；测试MCP的食物识别和模式判断能力；验证DeepSeek对日常饮食的分析建议。
测试数据：典型中式餐饮组合（炸鸡+奶茶，牛肉火锅）。
预期输出：MCP识别出炸鸡、奶茶、牛肉等食物；MCP选择nutrition_review模式；DeepSeek输出包含健康评价和建议。

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)

测试目的：验证减脂食谱生成功能；测试关键词触发diet_plan模式；验证特定食材的食谱定制能力。
测试数据：明确减肥意图和指定食材。
预期输出：MCP识别减肥关键词，选择diet_plan模式；DeepSeek生成包含热量估算的食谱表。

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)

测试目的：验证食物过量风险分析功能；测试关键词触发effects_inquiry模式；验证副作用信息的准确输出。
测试数据：直接询问食物过量影响。
预期输出：MCP识别"吃多了"关键词，选择effects_inquiry模式，DeepSeek输出科学依据的副作用说明。

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)

测试目的：验证食物相克关系分析功能；测试关键词触发avoid_inquiry模式；验证相克原因和替代方案输出。
测试数据：直接询问两种食物兼容性。
预期输出：MCP识别"能一起吃吗"关键词，选择avoid_inquiry模式，DeepSeek输出不宜同食的科学解释。

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)流式接口测试：验证SSE流式接口功能；测试元数据和内容分块传输；验证实时输出效果。MCP服务集成测试：模拟客户端调用MCP的/mcp接口，验证服务可用性和响应格式，检查返回数据结构完整性，打印中间结果便于调试。DeepSeek集成测试：验证 DeepSeek 客户端功能；测试提示词构建质量；检查生成内容的相关性；评估响应时间。异常处理机制：处理网络传输异常；避免测试因网络问题中断；确保资源释放；提供友好的错误提示。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

Ctrl + S键保存代码。

在CodeArts IDE for Python左侧资源管理器 > food_mcp工程右侧的新建文件按钮，并将文件命名为text_match.py，下一步在此文件中开始编写代码。

# -*- coding: utf-8 -*-# text_match.pyimport jiebaimport difflibfrom 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]

Ctrl + S键保存代码。

在CodeArts IDE for Python左侧资源管理器 > food_mcp工程右侧的新建文件按钮，并将文件命名为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": "控制每日总咖啡因

在CodeArts IDE for Python左侧资源管理器 > food_mcp工程右侧的新建文件按钮，并将文件命名为config.json，下一步在此文件中开始编写配置文件。

{ "DEEPSEEK_API_KEY": "API Key", "DEEPSEEK_URL": "API地址", "MCP_PORT": 8001, "MODEL_NAME": "模型名称"}

注：需要替换在步骤“2. 免费领取DeepSeek R1满血版”中获取到的API地址、模型名称和API Key。

在CodeArts IDE for Python左侧资源管理器 > food_mcp工程右侧的新建文件按钮，并将文件命名为requirements.txt，下一步在此文件中开始编写配置文件。

fastapiuvicornrequestshttpxjiebacachetoolsloguru

点击终端，运行如下命令，系统自动安装requirements.txt文件中的依赖包。

pip install -r requirements.txt -i https://mirrors.huaweicloud.com/repository/pypi/simple

注：安装完毕后，代码文件中如果显示仍然缺少依赖，请关闭ide，重新打开即可。

点击终端，运行如下命令，启动MCP服务。

uvicorn mcp_server:app --port 8001

点击终端右上角的“+”，新建终端，在新的终端界面执行如下命令，测试程序运行结果：

python test_pipeline.py

在执行测试成功的示例图中，我们就可以看到流式响应的效果。

本案例采用本地MCP服务 + 本地仓颉AI机器人对话。本章节讲解如何构建本地仓颉AI机器人。

返回云主机桌面，打开CodeArts IDE for Cangjie。

在新打开的CodeArts IDE for Cangjie的提示界面或通过文件 > 新建 > 工程，打开新建工程配置界面。

在新建工程配置界面，编辑项目名称为food_bot，然后点击创建。

在CodeArts IDE for Cangjie左侧资源管理器 > food_bot工程右侧的新建文件按钮，并将文件命名为config.json，下一步在此文件中开始编写配置文件。

{ "model": "", "api_key": "", "base_url": "http://localhost:8001/mcp_stream", "system_prompt": ""}

Ctrl + S键保存配置文件代码。

在CodeArts IDE for Cangjie左侧资源管理器 > food_bot工程中找到配置文件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 = {}

Ctrl + S键保存配置文件代码。在弹出的选项中选择yes，修改配置文件cjpm.toml需要重启LSPServer。

在CodeArts IDE for Cangjie左侧资源管理器 > food_bot工程目录中找到src/main.cj，下一步在此文件中开始编写代码。

package food_botimport std.console.Consoleimport std.collection.ArrayList初始化与欢迎信息：打印欢迎语，提示退出指令；初始化history列表存储用户输入与AI回复的元组 (用户输入, AI回复)。用户输入处理：读取控制台输入，跳过空输入；支持exit/exit退出程序；支持clear清空对话历史。流式模式动态切换：根据命令行参数stream和API地址特征（是否以 /mcp_stream 结尾）智能启用流式传输。聊天接口调用：根据流式模式选择调用stream_chat（流式）或chat（非流式）；自动传递当前输入prompt、环境配置env_info和历史对话history。健壮性设计：首次调用失败时自动重试一次；重试仍失败则终止程序。对话历史管理：成功响应后保存当前对话到history，维持多轮对话上下文。func cli_chat(env_info: EnvInfo, stream!: Bool) { println("\n欢迎使用智能膳食分析助手，输入exit退出") var history: ArrayList = ArrayList 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= 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): 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}

在CodeArts IDE for Cangjie左侧资源管理器 > food_bot工程目录中找到src目录，点击food_bot工程右侧的新建文件按钮，将文件命名为env_info.cj，下一步在此文件中开始编写代码。

package food_botimport encoding.json.stream.*import std.fs.Fileimport std.fs.Pathimport std.io.ByteArrayStream配置数据容器：存储AI服务必需的4个核心参数；包含默认值防止空值异常。JSON序列化/反序列化：实现双向JSON转换接口，支持从JSON流构建对象 (fromJson)，支持将对象输出为JSON (toJson)。健壮性设计：字段级默认值（如API密钥占位符），忽略未知JSON字段的容错处理，严格的JSON结构校验（BeginObject/EndObject）。安全防护：API密钥默认使用掩码值、序列化时自动处理敏感数据。& 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.startObject while(r.peek != EndObject) { let n = r.readName match (n) { case "model" => temp_model = r.readValue case "api_key" => temp_api_key = r.readValue case "base_url" => temp_base_url = r.readValue case "system_prompt" => temp_system_prompt = r.readValuecase _ => } } 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 }}配置模板生成：自动创建标准配置文件模板 (env_sample.json)，包含带说明的占位值。开发者辅助：为新用户提供配置参考，展示正确的JSON结构格式，输出优化：使用美化格式 (WriteConfig.pretty) 增强可读性，清理旧文件避免冲突。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}配置文件加载：从config.json加载运行时配置，硬编码路径确保一致性。异常处理：文件不存在时抛出明确异常，终止程序防止后续错误。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= 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}

Ctrl + S键保存代码。

在CodeArts IDE for Cangjie左侧资源管理器 > food_bot工程目录中找到src目录，点击food_bot工程右侧的新建文件按钮，将文件命名为chat.cj，下一步在此文件中开始编写代码。

package food_botimport encoding.json.stream.*import net.http.ClientBuilderimport net.http.HttpHeadersimport net.http.HttpRequestBuilderimport net.tls.TlsClientConfigimport net.tls.CertificateVerifyModeimport std.collection.ArrayListimport std.io.ByteArrayStreamimport std.time.Durationimport std.unicode.UnicodeExtension // for String.trim{ 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): 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 }}消息结构体 (Message)

封装单条对话消息，实现角色和内容的双向JSON转换。

& 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= None var temp_content: String = "" while (let Some(v) r.startObject while(r.peek != EndObject) { let n = r.readName match(n) { case "role" => temp_role = r.readValue> case "content" => temp_content = r.readValue 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>(role_type_to_str(this.role)) w.writeName("content").writeValue(this.content) w.endObject w.flush }}聊天请求结构体 (ChatRequest)

多场景构造器：直接使用消息列表；基于提示词+历史对话+系统提示构建。
智能上下文构建：系统提示 > 历史对话循环 > 用户输入。

, 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, system_prompt: String, max_tokens: Int64, stream: Bool ){ // construction function with prompt and system_prompt this.model = model this.messages = ArrayList([ 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, system_prompt: String, stream: Bool ){ // construction function with prompt and default arguments this.model = model this.messages = ArrayList([ 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(this.model) w.writeName("messages").writeValue>(this.messages) w.writeName("max_tokens").writeValue(this.max_tokens) w.writeName("stream").writeValue(this.stream) w.endObject w.flush }}响应相关结构体Choice) { 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= None while (let Some(v) r.startObject while(r.peek != EndObject) { let n = r.readName match (n) { case "index" => temp_index = r.readValue case "message" => temp_message = r.readValue> case "delta" => temp_delta = r.readValue> case "finish_reason" => temp_finish_reason = r.readValue> case "logprobs" => temp_logprobs = r.readValue> 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(this.index) w.writeName("message").writeValue>(this.message) w.writeName("delta").writeValue>(this.delta) w.writeName("finish_reason").writeValue>(this.finish_reason) w.writeName("logprobs").writeValue>(this.logprobs) w.endObject w.flush }}响应相关结构体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.startObject while(r.peek != EndObject) { let n = r.readName match (n) { case "prompt_tokens" => temp_prompt_tokens = r.readValue case "completion_tokens" => temp_completion_tokens = r.readValue case "total_tokens" => temp_total_tokens = r.readValue 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(this.prompt_tokens) w.writeName("completion_tokens").writeValue(this.completion_tokens) w.writeName("total_tokens").writeValue(this.total_tokens) w.endObject w.flush }}响应相关结构体ChatResponse

ChatResponse，用于实现完整的响应容器。

) { 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= ArrayList= None while (let Some(v) r.startObject while(r.peek != EndObject) { let n = r.readName match (n) { case "id" => temp_id = r.readValue> case "request_id" => temp_request_id = r.readValue> case "system_fingerprint" => temp_system_fingerprint = r.readValue> case "model" => temp_model = r.readValue case "object" => temp_object = r.readValue case "created" => temp_created = r.readValue case "choices" => temp_choices = r.readValue> case "usage" => temp_usage = r.readValue> 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 ) }}后端类型自适应：检测URL是否含/mcp_stream区分自定义/标准API；生成差异化请求体。安全连接：HTTPS启用信任所有证书模式（TrustAll）；设置域名验证（get_domain）流式支持：添加text/event-stream请求头；设置长超时（300秒）。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, stream!: Bool){ // prepare input data // If we are targeting the custom `/mcp_stream` backend we must send // a very simple JSON body `{ "input": "" }` 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: Arrayif (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)}非流式聊天 (chat)= None var res_text = "" try { // call api let response = client.send( request ) // read result (support max revice 100k data) let buffer = Array(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= 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}流式聊天 (stream_chat)数据格式关键事件自定义后端事件流(event: xxx)token, ping, error, done标准OpenAIdata: JSON对象delta.content, [DONE]{ 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(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= 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= 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 }}

Ctrl + S键保存代码。

点击终端图标，打开终端，执行命令：

cjpm run

程序运行成功，输入测试语句例如：我中午吃了炸鸡和奶茶还有香菇 我晚上吃点什么比较好呢？。

我们看到日志返回成功。切换到CodeArts IDE for Python，切换终端窗口，我们可以看到MCP服务端接受到请求的日志为“未命中”，这是正常响应。

切换到CodeArts IDE for Cangjie，再次输入测试语句例如：我中午吃了炸鸡和奶茶还有香菇 我晚上吃点什么比较好呢？。

再次切换回CodeArts IDE for Python，我们发现终端窗口中输出“命中”缓存的日志。

两次请求的数据，若请求未命中，则去DeepSeek请求并返回正常的输出结果；若请求相同数据，则命中缓存，返回缓存中的结果。

食品实体识别能力

系统具备从自然语言中准确识别出饮食相关的食品实体与关键词的能力。例如，用户输入：“我今天吃了煎饼果子、卤牛肉、奶茶”，系统应能够自动识别出食品列表：

["煎饼果子", "卤牛肉", "奶茶"]

支持中文短语、多食物组合、模糊词等多种表达方式。

饮食结构化分析能力（MCP中间层处理）

系统应通过MCP模块完成对用户输入饮食内容的语义解析与结构化处理，主要包括：

应提取食品清单；应识别饮食行为时段（如早餐、晚餐、夜宵等）；应对识别出的食物打上健康标签（如高糖、高油脂、高钠、低纤维等）；应构造符合LLM输入规范的Prompt，以支撑后续模型理解与生成。结构化分析结果应作为DeepSeek模型调用的核心中间数据。合理饮食建议生成能力（基于DeepSeek）

系统应利用DeepSeek大模型，根据MCP构造的Prompt生成具有科学性、专业性与可操作性的饮食建议，包括：

对当前饮食存在的健康风险进行分析（如糖分摄入超标、缺乏蔬菜等）；提供可替代食材建议（如炸鸡 → 烤鸡，奶茶 → 绿茶）；给出搭配优化方案（如餐前饮水、搭配高纤维蔬菜等）；提出个性化提示（如适合三高人群、健身人群、孕期饮食等）。

模型输出以自然语言形式呈现，逻辑清晰、结构明确，风格应贴近真实营养师表达方式。

本项目旨在构建一套基于多模块协作的智能膳食分析助手，通过自然语言输入，实现对个人饮食行为的结构化分析与健康建议生成，服务于日常健康管理、饮食优化与疾病预防等场景。

系统采用模块化设计，主要由三大核心组成部分构成：

模块技术栈职责

缓存可以采用本地文件、redis等更加高性能的方案来替换，这样就可以让实现由AI回答到AI理解到了所表达和呈现的每一个需求点。

来源：华为云开发者联盟

免责声明：本站系转载，并不代表本网赞同其观点和对其真实性负责。如涉及作品内容、版权和其它问题，请在30日内与本站联系，我们将在第一时间删除内容!
相关推荐