昇思大模型应用：工作流编排

skytier

114人浏览 · 2026-10-28 10:16:25

skytier · 2026-10-28 10:16:25 发布

一、摘要

昇思大模型工作流编排，是基于 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", encoding="utf-8") as f:
        text = f.read()
    chunks = [text[i:i+512] 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_tensors="ms")
    output = model.generate(**inputs, max_new_tokens=256)
    answer = tokenizer.decode(output[0], skip_special_tokens=True)
    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_target="Ascend", mode=ms.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_retry=2):
    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 state

MindSpore 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，不要在工作流调度层处理模型并行。