一、摘要昇思大模型工作流编排是基于 MindSpore、MindSpore-Transformers、MindSpeed、MindSpore Science 等组件把数据预处理、模型调用、工具调用、子 Agent、后处理、结果校验、重试分支等独立节点串联成端到端业务链路适配昇腾硬件的应用层调度体系区别于模型内部的流水线并行PP。核心能力1.节点解耦每个节点是独立可复用单元包含数据加载、LLM 推理、向量检索、外部 API 调用、自定义 Python 函数、分支判断、循环重试支持同步 / 异步调度、断点续跑。2.昇腾原生适配节点内模型推理复用 MindIE、MindSpore 推理后端自动接管昇腾 NPU 显存、HCCL 通信支持多卡推理、KV 缓存复用和 CANN 算子栈打通。3.两种编排范式○ 工程流水线RAG、微调、评测、模型导出等固定 AI 任务流水线MindSpore-Transformers Trainer 回调、ModelArts Workflow○ Agent 动态编排MindSpore Science 多子智能体ReAct 规划、工具选择、自我纠错动态生成执行路径用于科研智能体、复杂业务 Agent昇思Min...。4.可观测性节点耗时、NPU 算力占用、输入输出日志、异常捕获对接 TensorBoard、自定义埋点。典型业务链路示例用户 Query → 文档加载 切分 → 向量库检索 → Prompt 组装 → MindSpore LLM 推理 → 结果校验 → 格式化输出。二、开发实操环境依赖pip install mindspore mindformers mindspeed最简代码顺序式工作流编排RAG 链路import mindspore as ms from mindformers import AutoModel, AutoTokenizer # 1. 定义工作流节点函数每个节点独立可单独调试 def load_doc_node(path: str): 节点1文档读取与文本切分 with open(path, r, encodingutf-8) as f: text f.read() chunks [text[i:i512] for i in range(0, len(text), 512)] return {chunks: chunks} def retrieval_node(chunks, query): 节点2检索这里简化为关键词匹配可替换向量库 res [c for c in chunks if query in c] context \n.join(res[:2]) return {context: context, query: query} def llm_infer_node(context, query, tokenizer, model): 节点3昇思大模型推理节点跑在昇腾NPU prompt f基于上下文回答问题\n上下文{context}\n问题{query} inputs tokenizer(prompt, return_tensorsms) output model.generate(**inputs, max_new_tokens256) answer tokenizer.decode(output[0], skip_special_tokensTrue) return {answer: answer} # 2. 工作流调度器串行编排可扩展分支/重试 def workflow_runner(workflow_nodes, init_input): state init_input for node_func in workflow_nodes: state.update(node_func(**state)) return state if __name__ __main__: ms.set_context(device_targetAscend, modems.GRAPH_MODE) model AutoModel.from_pretrained(Qwen-7B-MindSpore) tokenizer AutoTokenizer.from_pretrained(Qwen-7B-MindSpore) # 注册节点顺序 pipeline [load_doc_node, retrieval_node, lambda context,query: llm_infer_node(context,query,tokenizer,model)] result workflow_runner(pipeline, {path: ./doc.txt, query: 昇思工作流编排是什么}) print(result[answer])进阶分支 异常重试编排import time def validate_answer_node(answer:str): 校验节点分支判断 if len(answer) 20: return {retry: True} return {retry: False} # 改造runner增加循环重试、分支跳转 def runner_with_branch(nodes, init_input, max_retry2): state init_input retry_cnt 0 while retry_cnt max_retry: state workflow_runner(nodes[:-1], state) state.update(validate_answer_node(state[answer])) if not state[retry]: break retry_cnt 1 time.sleep(0.5) return stateMindSpore Science Agent 编排多子智能体MindSpore Science 内置 SurveyAgent、PlanAgent、ExecuteAgent 等 Sub-Agent支持声明式 Workflow 配置适合科研场景自动实验流程通过 Skill 系统调用仿真软件、实验设备、外部工具子 Agent 之间传递中间结果支持自我修正减少人工串联成本昇思Min...。云侧低代码编排ModelArts Workflow 提供可视化拖拽节点直接拖拽数据集、MindSpore 训练、模型评估、MindIE 部署节点生成 DAG 工作流自动调度昇腾算力适合企业级批量 AI 任务。三、问题总结1.NPU 显存溢出跨节点显存未释放根因多节点串行执行时MindSpore 计算图常驻显存上一轮推理 Tensor 没有主动释放。对策节点边界处增加ms.reset_memory()推理节点使用 MindIE 静态图推理推理结束释放 KV 缓存大并发场景采用异步调度 推理池。2.节点间数据传递序列化异常根因直接传递 MindSpore Tensor 对象跨异步节点 / 进程无法序列化。对策节点边界只传递 numpy 数组、字符串、字典Tensor 仅在模型推理节点内部创建销毁。3.Agent 动态编排无限循环、上下文污染根因子 Agent 错误输出的报错信息持续追加到 prompt。对策参考 MindSpore Science 设计增加 Critic 校验 Agent过滤无效报错信息限制上下文窗口长度设置最大规划步数。4.分布式多卡工作流调度冲突根因工作流调度器和模型内部 PP/TP 并行调度混淆HCCL 通信组冲突。对策区分两层并行工作流是应用业务 DAG调度任务流水线并行是模型内部层切分在单模型推理节点内部配置pipeline_stages不要在工作流调度层处理模型并行。四、应用场景场景 1企业 RAG 知识库问答工作流节点 DAG文档解析→分块→向量化入库→检索过滤→Prompt 构造→MindSpore LLM 推理→敏感词过滤→结果返回异步队列支持多用户并发节点可单独扩缩容推理节点池复用昇腾 NPU。场景 2大模型微调流水线节点数据集加载 清洗 → 数据去冗余MindSpore-Transformers 数据通信域去冗余→ LoRA 微调训练 → 评估指标计算 → safetensors 权重保存 → MindIR 导出全链路可定时调度失败节点单独重试无需从头跑完整训练。场景 3科研多智能体工作流MindSpore Science文献调研 Agent → 方案假设 Agent → 实验规划 Agent → 仿真 / 湿实验执行 Agent → 结果批判 Agent动态分支实验失败自动调整参数重新规划打通仿真软件 API 与昇腾算力。场景 4多模态应用编排图像预处理昇腾 DVPP→ 视觉 Encoder 推理 → 文本 LLM → 结果渲染异构模态节点独立调度Mpipe 多模态流水线优化减少空泡提升多模态端到端吞吐昇思Min...。工程化最佳实践● 节点无状态设计输入输出结构化方便单元测试● 生产环境优先 MindIE 做推理节点图编译一次多次调度● 工作流和模型并行解耦业务 DAG 负责任务流转框架内部负责张量并行 / 流水线并行● 关键节点增加超时、重试、降级策略记录全链路 Trace。