本案例由开发者：给无眠点压力提供
一、概述
1. 案例介绍
MCP，全称 Model Context Protocol，中文叫“模型上下文协议”。你可以把它想象成 AI 的“USB 接口” --让不同的 AI 模型、工具和应用程序能用统一的方式交流。那么我的理解是：它更像是一个适配器来调节各种 AI 不同的接口达到一致的效果，让 AI 的交流更加简单，即使没有身份预设，走 MCP 是完美的让 AI 成为你的最佳助手。
随着人们对健康饮食关注度的提升，越来越多用户希望借助 AI 助手实现个性化的饮食分析与管理。然而，目前市面上的饮食类应用普遍存在如下痛点：
缺乏智能分析：大多数仅记录卡路里，无法提供专业点评与优化建议；
知识更新不及时：难以结合最新营养研究进行推荐与判断；
缺乏可扩展性：难以适配特定人群（如高血压、糖尿病、健身人群等）的差异需求。
本项目旨在构建一个基于大语言模型（DeepSeek）和结构化协议（MCP）的智能饮食健康助手，通过自然语言交互，帮助用户实现饮食数据结构化、健康风险识别与个性化建议生成。
2. 适用对象
3. 案例时间
本案例总时长预计 90 分钟。
4. 案例流程
说明：
登录华为开发者空间工作台，领取云主机。在云主机登录 ModelArts Studio(MaaS)控制台，领取 DeepSeek-V3 百万免费 Tokens。
华为开发者空间 - 云主机 桌面打开 CodeArts IDE for Python 构建本地 MCP 服务项目，实现饮食语义解析、食物识别、健康标签打标、Prompt 生成与请求路由。
华为开发者空间 - 云主机 桌面打开 CodeArts IDE for Cangjie 构建本地仓颉 AI 机器人，接收经过标准化构造的请求 Prompt，返回结构化营养建议、健康分析与饮食优化建议。
5. 资源总览
本案例预计花费 0 元。
二、环境与资源准备
1. 配置开发者空间
面向广大开发者群体，华为开发者空间提供一个随时访问的“开发桌面云主机”、丰富的“预配置工具集合”和灵活使用的“场景化资源池”，开发者开箱即用，快速体验华为根技术和资源。
如果还没有领取云主机进入工作台界面后点击配置云主机，选择 Ubuntu 操作系统。
进入华为开发者空间工作台界面，点击打开云主机 > 进入桌面连接云主机。
2. 免费领取 DeepSeek R1 满血版
华为云提供了单模型 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 服务项目
本案例采用本地 MCP 服务 + 本地仓颉 AI 机器人对话。本章节讲解如何构建本地 MCP 服务。
1. 新建项目 food_mcp
在云主机桌面打开 CodeArts IDE for Python。
在新打开的 CodeArts IDE for Python 的提示界面或通过文件 > 新建 > 工程，打开新建工程配置界面。
在新建工程配置界面，编辑项目名称为food_mcp，然后点击创建。
2. 功能实现 mcp_server.py
在 CodeArts IDE for Python 左侧资源管理器 > food_mcp 工程右侧的新建文件按钮，并将文件命名为mcp_server.py，下一步在此文件中开始编写代码。
2.1 模块导入与初始化
依次导入 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
复制代码
2.2 数据加载
从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)
复制代码
2.3 API 服务初始化
创建 FastAPI 应用实例：定义输入数据模型（仅含 input 字符串字段），初始化全局缓存（500 条容量，30 分钟过期），配置日志系统（输出到控制台，INFO 级别）。
app = FastAPI()
class Input(BaseModel): input: str
# 全局缓存：prompt -> full_reply_textCACHE = TTLCache(maxsize=500, ttl=1800) # 30分钟自动过期
logger.remove()logger.add(sys.stdout, level="INFO", enqueue=False, backtrace=False)
复制代码
2.4 提示词模板系统
定义营养师基础角色设定；创建多场景提示词模板：
统一要求输出包含 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}" ),}
复制代码
2.5 核心处理函数
食物提取：识别输入中的食物名称，区分精确匹配和模糊匹配结果，记录匹配日志。
信息聚合：聚合所有匹配食物的属性标签，收集副作用信息，提取相克食物列表，确定饮食类型。
场景模式识别：基于关键词检测用户意图，自动路由到合适的处理场景。
提示词构建：动态生成场景化提示词，注入食物属性作为关键约束，要求 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, }
复制代码
2.6 API 端点实现同步端点(/mcp)
@app.post("/mcp")async def route_prompt(data: Input): user_input = data.input result = _process_input(user_input) return result
复制代码
2.7 API 端点实现流式端点(/mcp_stream)
实现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,配置依赖包”中解决依赖包的问题。
3. 功能实现 deepseek_client.py
在 CodeArts IDE for Python 左侧资源管理器 > food_mcp 工程右侧的新建文件按钮，并将文件命名为deepseek_client.py，下一步在此文件中开始编写代码。
3.1 模块导入与初始化
# deepseek_client.py# 基于 DeepSeek API 的客户端import requestsimport jsonfrom typing import Generatorimport httpx
复制代码
3.2 配置加载模块
从 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"]
复制代码
3.3 同步调用函数
实现同步调用 DeepSeek API；
构建 API 请求头（包含认证信息）；
构造符合 DeepSeek 要求的请求体；
发送 POST 请求并获取完整响应；
解析响应并返回生成的完整文本内容；
适用于不需要实时反馈的简单场景。
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"]
复制代码
3.4 同步流式生成器
# 新增：流式推理生成器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
复制代码
3.5 异步流式生成器
实现异步流式 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 键保存代码。
注： 此时工程中会报错，这是因为开发环境中缺少必要的包文件，此处无须理会，我们将在后续步骤“8.1 编写配置文件 requirements.txt,配置依赖包”中解决依赖包的问题。
4. 功能实现 test_pipeline.py
在 CodeArts IDE for Python 左侧资源管理器 > food_mcp 工程右侧的新建文件按钮，并将文件命名为test_pipeline.py，下一步在此文件中开始编写代码。
4.1 模块导入与初始化
# -*- coding: utf-8 -*-# test_pipeline.pyimport requestsfrom deepseek_client import call_deepseekimport jsonfrom requests.exceptions import ChunkedEncodingError
复制代码
4.2 饮食分析测试test_diet_analysis()
测试目的：验证常规饮食分析场景；测试 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)
复制代码
4.3 减脂计划测试test_diet_plan()
测试目的：验证减脂食谱生成功能；测试关键词触发 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)
复制代码
4.4 副作用查询测试test_effects_inquiry()
测试目的：验证食物过量风险分析功能；测试关键词触发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)
复制代码
4.5 食物相克测试test_avoid_inquiry()
测试目的：验证食物相克关系分析功能；测试关键词触发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)
复制代码
4.6 流式接口与核心功能测试 test_stream()
流式接口测试：验证 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
复制代码
4.7 测试执行模式
模块化测试用例设计；可选择性执行单个测试；便于快速验证特定功能；支持迭代开发中的持续测试。
if __name__ == "__main__": # test_diet_analysis() # test_diet_plan() # test_effects_inquiry() # test_avoid_inquiry() test_stream()
复制代码
Ctrl + S 键保存代码。
注： 此时工程中会报错，这是因为开发环境中缺少必要的包文件，此处无须理会，我们将在后续步骤“8.1 编写配置文件 requirements.txt,配置依赖包”中解决依赖包的问题。
5. 功能实现 text_match.py
在 CodeArts IDE for Python 左侧资源管理器 > food_mcp 工程右侧的新建文件按钮，并将文件命名为text_match.py，下一步在此文件中开始编写代码。
5.1 模块导入与初始化
# -*- coding: utf-8 -*-# text_match.pyimport jiebaimport difflibfrom typing import Iterable, List, Set, Tuple
复制代码
5.2 核心功能：食物关键词提取
从用户输入文本中识别食物关键词，支持精确匹配和模糊匹配两种模式，返回三种匹配结果：
所有匹配到的食物列表（去重）
精确匹配的食物列表
模糊匹配的食物列表
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
复制代码
5.3 辅助功能：简化接口
提供向后兼容的简化接口；只返回匹配到的食物列表（不区分精确/模糊）；保持旧模块的调用方式不变。
# 向后兼容的简化接口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 键保存代码。
注： 此时工程中会报错，这是因为开发环境中缺少必要的包文件，此处无须理会，我们将在后续步骤“8.1 编写配置文件 requirements.txt,配置依赖包”中解决依赖包的问题。
6. 编写配置文件 food_tags.json
在 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": "控制每日总咖啡因 < 400mg" }}
复制代码
7. 编写配置文件 config.json
在 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。
8. 编写配置文件 requirements.txt,配置依赖包
8.1 编写配置文件 requirements.txt
在 CodeArts IDE for Python 左侧资源管理器 > food_mcp 工程右侧的新建文件按钮，并将文件命名为requirements.txt，下一步在此文件中开始编写配置文件。
fastapiuvicornrequestshttpxjiebacachetoolsloguru
复制代码
8.2 配置依赖包
点击终端，运行如下命令，系统自动安装requirements.txt文件中的依赖包。
pip install -r requirements.txt -i https://mirrors.huaweicloud.com/repository/pypi/simple
复制代码
注：安装完毕后，代码文件中如果显示仍然缺少依赖，请关闭 ide，重新打开即可。
9. 调试 MCP
点击终端，运行如下命令，启动 MCP 服务。
uvicorn mcp_server:app --port 8001
复制代码
点击终端右上角的“**+**”，新建终端，在新的终端界面执行如下命令，测试程序运行结果：
在执行测试成功的示例图中，我们就可以看到流式响应的效果。
四、构建本地仓颉 AI 机器人
本案例采用本地 MCP 服务 + 本地仓颉 AI 机器人对话。本章节讲解如何构建本地仓颉 AI 机器人。
1. 新建项目 food_bot
返回云主机桌面，打开 CodeArts IDE for Cangjie。
在新打开的 CodeArts IDE for Cangjie 的提示界面或通过文件 > 新建 > 工程，打开新建工程配置界面。
在新建工程配置界面，编辑项目名称为food_bot，然后点击创建。
2. 编写配置文件 config.json
在 CodeArts IDE for Cangjie 左侧资源管理器 > food_bot 工程右侧的新建文件按钮，并将文件命名为config.json，下一步在此文件中开始编写配置文件。
{ "model": "", "api_key": "", "base_url": "http://localhost:8001/mcp_stream", "system_prompt": ""}
复制代码
Ctrl + S 键保存配置文件代码。
3. 编写配置文件 cjpm.toml
在 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。
4. 功能实现 src/main.cj
在 CodeArts IDE for Cangjie 左侧资源管理器 > food_bot 工程目录中找到src/main.cj，下一步在此文件中开始编写代码。
4.1 模块导入与初始化
package food_botimport std.console.Consoleimport std.collection.ArrayList
复制代码
4.2 交互式聊天核心 (cli_chat)
初始化与欢迎信息：打印欢迎语，提示退出指令；初始化 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<(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 } } }}
复制代码
4.3 主入口函数 (main)
初始化程序环境配置（load_env_info()）。
解析命令行参数，仅支持 --stream 标志开启流式输出模式。
调用核心聊天函数 cli_chat() 启动交互界面。
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}
复制代码
Ctrl + S 键保存代码。
5. 功能实现 src/env_info.cj
在 CodeArts IDE for Cangjie 左侧资源管理器 > food_bot 工程目录中找到 src 目录，点击 food_bot 工程右侧的新建文件按钮，将文件命名为env_info.cj，下一步在此文件中开始编写代码。
5.1 模块导入与初始化
package food_botimport encoding.json.stream.*import std.fs.Fileimport std.fs.Pathimport std.io.ByteArrayStream
复制代码
5.2 环境配置类 (EnvInfo)
配置数据容器：存储 AI 服务必需的 4 个核心参数；包含默认值防止空值异常。
JSON 序列化/反序列化：实现双向 JSON 转换接口，支持从 JSON 流构建对象 (fromJson)，支持将对象输出为 JSON (toJson)。
健壮性设计：字段级默认值（如 API 密钥占位符），忽略未知 JSON 字段的容错处理，严格的 JSON 结构校验（BeginObject/EndObject）。
安全防护：API 密钥默认使用掩码值、序列化时自动处理敏感数据。
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() }}
复制代码
5.3 配置样本生成器 (save_env_info)
配置模板生成：自动创建标准配置文件模板 (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()}
复制代码
5.4 配置加载器 (load_env_info)
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}
复制代码
Ctrl + S 键保存代码。
6. 功能实现 src/chat.cj
在 CodeArts IDE for Cangjie 左侧资源管理器 > food_bot 工程目录中找到 src 目录，点击 food_bot 工程右侧的新建文件按钮，将文件命名为chat.cj，下一步在此文件中开始编写代码。
6.1 模块导入与初始化
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()
复制代码
6.2 核心数据结构
消息角色枚举 (RoleType)
定义对话消息的三种角色类型（用户/助手/系统）及转换方法。
// ===== 可配置常量 =====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 }}
复制代码
消息结构体 (Message)
封装单条对话消息，实现角色和内容的双向 JSON 转换。
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() }}
复制代码
聊天请求结构体 (ChatRequest)
多场景构造器：直接使用消息列表；基于提示词+历史对话+系统提示构建。智能上下文构建：系统提示 > 历史对话循环 > 用户输入。
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() }}
复制代码
响应相关结构体 Choice
封装 AI 返回的选择项（支持流式增量）。
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() }}
复制代码
响应相关结构体 Usage
实现统计 token 消耗。
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() }}
复制代码
响应相关结构体 ChatResponse
ChatResponse，用于实现完整的响应容器。
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 ) }}
复制代码
6.3 HTTP 通信核心
请求构建器 (build_http_client)
后端类型自适应：检测 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<(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)}
复制代码
非流式聊天 (chat)
客户端 > AI 服务：发送完整请求。
AI 服务 > 客户端：返回完整 JSON。
客户端 > 客户端：解析第一选择项内容。
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}
复制代码
流式聊天 (stream_chat)
双模式处理：
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 }}
复制代码
Ctrl + S 键保存代码。
7. 调试 AIChat
点击终端图标，打开终端，执行命令：
程序运行成功，输入测试语句例如：我中午吃了炸鸡和奶茶还有香菇 我晚上吃点什么比较好呢？。
我们看到日志返回成功。切换到 CodeArts IDE for Python，切换终端窗口，我们可以看到 MCP 服务端接受到请求的日志为“未命中”，这是正常响应。
切换到 CodeArts IDE for Cangjie，再次输入测试语句例如：我中午吃了炸鸡和奶茶还有香菇 我晚上吃点什么比较好呢？。
再次切换回 CodeArts IDE for Python，我们发现终端窗口中输出“命中”缓存的日志。
两次请求的数据，若请求未命中，则去 DeepSeek 请求并返回正常的输出结果；若请求相同数据，则命中缓存，返回缓存中的结果。
五、项目总结
1. 核心能力
食品实体识别能力
系统具备从自然语言中准确识别出饮食相关的食品实体与关键词的能力。例如，用户输入：“我今天吃了煎饼果子、卤牛肉、奶茶”，系统应能够自动识别出食品列表：
支持中文短语、多食物组合、模糊词等多种表达方式。
饮食结构化分析能力（MCP 中间层处理）
系统应通过 MCP 模块完成对用户输入饮食内容的语义解析与结构化处理，主要包括：
应提取食品清单；
应识别饮食行为时段（如早餐、晚餐、夜宵等）；
应对识别出的食物打上健康标签（如高糖、高油脂、高钠、低纤维等）；
应构造符合 LLM 输入规范的 Prompt，以支撑后续模型理解与生成。
结构化分析结果应作为 DeepSeek 模型调用的核心中间数据。
合理饮食建议生成能力（基于 DeepSeek）
系统应利用 DeepSeek 大模型，根据 MCP 构造的 Prompt 生成具有科学性、专业性与可操作性的饮食建议，包括：
对当前饮食存在的健康风险进行分析（如糖分摄入超标、缺乏蔬菜等）；
提供可替代食材建议（如炸鸡 → 烤鸡，奶茶 → 绿茶）；
给出搭配优化方案（如餐前饮水、搭配高纤维蔬菜等）；
提出个性化提示（如适合三高人群、健身人群、孕期饮食等）。
模型输出以自然语言形式呈现，逻辑清晰、结构明确，风格应贴近真实营养师表达方式。
2. 技术选型
本项目旨在构建一套基于多模块协作的智能膳食分析助手，通过自然语言输入，实现对个人饮食行为的结构化分析与健康建议生成，服务于日常健康管理、饮食优化与疾病预防等场景。
系统采用模块化设计，主要由三大核心组成部分构成：
3. 项目亮点
使用 MCP 解耦输入处理与模型调用，提升可扩展性与灵活性；
DeepSeek 专注自然语言生成任务，MCP 预处理提高 prompt 精度；
仓颉终端可后续拓展为图形界面、Web 应用或移动端入口；
本地配置与食物字典可按需调整，适配多种用户类型（如减脂、控糖、健身等）；
架构支持多模型/多来源接入，可拓展至 Claude、SparkDesk、ChatGPT 等后端模型。
4. 后续拓展
food_bot：通过 vue 等各种 UI 前端适配不同前端风格的机器人。
food_mcp：独立开发为自己更加需要的模式，更加契合的"USB"，让 AI 更加懂你。
5.关于缓存方面
缓存可以采用本地文件、redis 等更加高性能的方案来替换，这样就可以让实现由 AI 回答到 AI 理解到了所表达和呈现的每一个需求点。
评论