# 搭建数据格式转换工作流 #### 场景介绍 企业日常运营中经常需要处理CRM、ERP等系统导出的JSON或CSV格式业务数据。这类数据包含大量重复的计算和清洗工作（如汇总销售额、筛选有效订单、按区域分组统计等），大模型虽然擅长文本理解和表达，但不适合进行精确的数学计算和复杂的数据操作。 为弥补这一不足，可以在工作流中引入代码节点，利用其执行Python或Node.js代码的能力来处理数据结构化任务；同时配合大模型节点，将计算后的结果转化为通俗易懂的业务分析报告。 本文将演示如何结合代码节点与大模型节点，以销售部门需要将CRM导出的JSON数据转为可读的周报表格为例，构建一个用于数据格式转换与智能分析的工作流。 #### 效果展示 当输入的原始JSON销售数据如下时，转换的销售周报如[图1]所示。 ``` [ {"client_name": "公司A", "amount": 580000, "product": "企业版套餐", "date": "2025-07-01", "sales_rep": "张三", "region": "华南", "status": "已签约"}, {"client_name": "公司B", "amount": 320000, "product": "标准版套餐", "date": "2025-07-03", "sales_rep": "李四", "region": "华南", "status": "已签约"}, {"client_name": "公司C", "amount": 750000, "product": "旗舰版套餐", "date": "2025-07-05", "sales_rep": "王五", "region": "华北", "status": "已签约"}, {"client_name": "公司D", "amount": 210000, "product": "标准版套餐", "date": "2025-07-06", "sales_rep": "张三", "region": "华南", "status": "跟进中"}, {"client_name": "公司E", "amount": 680000, "product": "企业版套餐", "date": "2025-07-08", "sales_rep": "赵六", "region": "华南", "status": "已签约"}, {"client_name": "公司F", "amount": 450000, "product": "企业版套餐", "date": "2025-07-10", "sales_rep": "李四", "region": "华北", "status": "已签约"}, {"client_name": "公司G", "amount": 180000, "product": "标准版套餐", "date": "2025-07-12", "sales_rep": "王五", "region": "华中", "status": "跟进中"} ] ``` 图1效果示例 ![](https://support.huaweicloud.com/bestpractice-agentarts/zh-cn_image_0000002687949840.png "点击放大") #### 工作流设计 重点介绍工作流中的一些主要节点，每个节点负责特定的任务。以下是各个节点的功能和设计思路。 - 开始节点：作为工作流的入口，开始节点负责接收CRM导出的JSON销售数据。 - 代码节点：通过Python代码解析JSON数据，计算各维度的汇总指标。 - 大模型节点：基于代码节点输出的结构化指标数据，生成自然语言周报。 - 结束节点：展示大模型生成的周报。 图2数据格式转换工作流全景 ![](https://support.huaweicloud.com/bestpractice-agentarts/zh-cn_image_0000002717389591.png "点击放大") #### 操作步骤 搭建工作流详细步骤如下： #### 步骤一：创建工作流 1. 登录[AgentArts智能体平台](https://console.huaweicloud.com/agentarts/#/home/overview)。 2. 单击左侧导航栏"开发中心 \> 智能体管理 "，单击左上角"工作流"页签，进入工作流应用管理界面，单击右上角"创建工作流"。 3. 选择创建"对话型工作流"，在对应的创建页面，配置工作流相关信息，具体如[图4](https://support.huaweicloud.com/bestpractice-agentarts/agentarts_06_0191.html#agentarts_06_0191__fig15344111710461)所示，参数如[表1](https://support.huaweicloud.com/bestpractice-agentarts/agentarts_06_0018.html#agentarts_06_0018__table14636564555)所示。 图3创建工作流 ![](https://support.huaweicloud.com/bestpractice-agentarts/zh-cn_image_0000002687789976.png "点击放大") 表1参数说明 | 参数名称 | 示例 | 说明 | |:---|:---|:---| | 名称 | 数据格式转换工作流 | 支持中英文、数字、下划线、中划线和空格，长度 2\~64字符，但不允许以空格开头或结尾。 | | 描述 | 将CRM导出的JSON数据转为可读的周报表格 | 描述工作流的功能，可直观呈现给用户，长度0\~1024个字符。 | 4. 配置完成后单击"立即创建"，进入工作流编排页面。 #### 步骤二：添加代码节点 单击开始节点和大模型节点中间的![](https://support.huaweicloud.com/bestpractice-agentarts/zh-cn_image_0000002687949846.png)，添加代码节点，如[图4]所示。 图4添加代码节点 ![](https://support.huaweicloud.com/bestpractice-agentarts/zh-cn_image_0000002717549481.png "点击放大") 添加完成后，如[图5]所示。 图5数据格式转换工作流 ![](https://support.huaweicloud.com/bestpractice-agentarts/zh-cn_image_0000002717549471.png "点击放大") #### 步骤三：配置工作流节点 工作流各个核心节点的配置方式如下： 表2节点配置说明 | 节点类型 | 说明 | 示例 | |:---|:---|:---| | 开始节点 | 添加一个必选的输入参数，接收CRM导出的JSON销售数据。 - 参数名称：raw_data - 参数类型：String - 描述：CRM导出的JSON格式销售数据 - 必填：勾选。 | ![](https://support.huaweicloud.com/bestpractice-agentarts/zh-cn_image_0000002687789992.png "点击放大") | | 代码节点 | 此节点通过Python代码解析JSON数据，计算各维度的汇总指标。 - **输入参数：** raw_data：引用开始节点的raw_data。 - **输出参数：** 参数名称：result 参数类型：String 描述：计算后的结构化指标（JSON字符串） - **代码：** 填入如下代码 ``` import json from collections import defaultdict def main(args: dict) -> dict: """ 运行代码节点会调用此函数，请勿对下面的函数名做修改 :param args: 输入固定为args字典类型，kv为输入参数键值对 :return: 输出参数为字典类型，kv为输出参数键值对 """ # 从args字典中获取原始数据字符串 raw_data = args.get('raw_data', '') data = json.loads(raw_data) # ========== 1. 整体指标 ========== total_sales = sum(item['amount'] for item in data) deal_count = len(data) avg_deal = round(total_sales / deal_count, 2) if deal_count > 0 else 0 signed_deals = [d for d in data if d['status'] == '已签约'] signed_amount = sum(d['amount'] for d in signed_deals) pipeline_deals = [d for d in data if d['status'] == '跟进中'] pipeline_amount = sum(d['amount'] for d in pipeline_deals) # ========== 2. Top 客户 ========== sorted_by_amount = sorted(data, key=lambda x: x['amount'], reverse=True) top3_clients = [ {"name": d['client_name'], "amount": d['amount'], "product": d['product']} for d in sorted_by_amount[:3] ] # ========== 3. 按销售代表汇总 ========== rep_stats = defaultdict(lambda: {"count": 0, "total": 0}) for d in data: rep = d['sales_rep'] rep_stats[rep]["count"] += 1 rep_stats[rep]["total"] += d['amount'] rep_summary = [ {"name": rep, "deal_count": stats["count"], "total_amount": stats["total"]} for rep, stats in rep_stats.items() ] rep_summary.sort(key=lambda x: x['total_amount'], reverse=True) # ========== 4. 按区域汇总 ========== region_stats = defaultdict(lambda: {"count": 0, "total": 0}) for d in data: region = d['region'] region_stats[region]["count"] += 1 region_stats[region]["total"] += d['amount'] region_summary = [ {"region": region, "deal_count": stats["count"], "total_amount": stats["total"]} for region, stats in region_stats.items() ] region_summary.sort(key=lambda x: x['total_amount'], reverse=True) # ========== 5. 按产品汇总 ========== product_stats = defaultdict(lambda: {"count": 0, "total": 0}) for d in data: product = d['product'] product_stats[product]["count"] += 1 product_stats[product]["total"] += d['amount'] product_summary = [ {"product": product, "deal_count": stats["count"], "total_amount": stats["total"]} for product, stats in product_stats.items() ] product_summary.sort(key=lambda x: x['total_amount'], reverse=True) # ========== 汇总输出 ========== result = { "overview": { "total_sales": total_sales, "deal_count": deal_count, "avg_deal_size": avg_deal, "signed_amount": signed_amount, "signed_count": len(signed_deals), "pipeline_amount": pipeline_amount, "pipeline_count": len(pipeline_deals) }, "top3_clients": top3_clients, "by_sales_rep": rep_summary, "by_region": region_summary, "by_product": product_summary } # 构造返回字典（严格遵循输出格式） ret = { "result": json.dumps(result, ensure_ascii=False) } return ret ``` | ![](https://support.huaweicloud.com/bestpractice-agentarts/zh-cn_image_0000002687949848.png "点击放大") | | 大模型节点 | 此节点基于代码节点输出的结构化指标数据，生成自然语言周报。 - **模型配置**：选择合适的模型。 - **输入参数：** - query，引用开始节点的query。 - data，引用代码节点的result。 - **提示词配置：** - 系统提示词：定义一段系统提示词，如下所示。 ``` 你是一名资深销售数据分析师，擅长将销售数据转化为清晰、专业的业务分析报告。 报告需包含以下模块： 1. 本周业绩概览（总销售额、成交笔数、平均客单价、已签约 vs 跟进中） 2. Top 客户排名（表格展示前3大客户） 3. 销售代表业绩（表格对比各销售代表的成交额和笔数） 4. 区域业绩分布（表格展示各区域表现） 5. 产品线分析（各产品套餐的销售占比） 6. 业务建议（2~3条基于数据的可操作建议） 数据如下： {{data}} 报告要求： 1. 语言专业但易懂，适合管理层快速阅读 2. 关键数字使用加粗标注 3. 使用表格呈现对比类数据 4. 在报告末尾给出 2~3 条可操作的业务建议 5. 金额统一使用万元为单位，保留两位小数 ``` - 用户提示词：引用当前的输入参数query，即引用用户的原始输入。 ``` {{query}} ``` | ![](https://support.huaweicloud.com/bestpractice-agentarts/zh-cn_image_0000002717389599.png "点击放大") | | 结束节点 | 此节点用于展示大模型生成的周报。 - **输入参数：**result，引用大模型节点的raw_output。 - **指定回复：** ``` {{result}} ``` | ![](https://support.huaweicloud.com/bestpractice-agentarts/zh-cn_image_0000002717549465.png "点击放大") | #### 步骤四：调试工作流 1. 配置完成后，在工作流编排界面单击右上角![](https://support.huaweicloud.com/bestpractice-agentarts/zh-cn_image_0000002717549475.png)，在试运行配置中的raw_data框中，输入CRM导出的JSON销售数据。如[图6]所示。配置完成后单击"开始运行"。 输入JSON销售数据示例： ``` [ {"client_name": "公司A", "amount": 580000, "product": "企业版套餐", "date": "2025-07-01", "sales_rep": "张三", "region": "华南", "status": "已签约"}, {"client_name": "公司B", "amount": 320000, "product": "标准版套餐", "date": "2025-07-03", "sales_rep": "李四", "region": "华南", "status": "已签约"}, {"client_name": "公司C", "amount": 750000, "product": "旗舰版套餐", "date": "2025-07-05", "sales_rep": "王五", "region": "华北", "status": "已签约"}, {"client_name": "公司D", "amount": 210000, "product": "标准版套餐", "date": "2025-07-06", "sales_rep": "张三", "region": "华南", "status": "跟进中"}, {"client_name": "公司E", "amount": 680000, "product": "企业版套餐", "date": "2025-07-08", "sales_rep": "赵六", "region": "华南", "status": "已签约"}, {"client_name": "公司F", "amount": 450000, "product": "企业版套餐", "date": "2025-07-10", "sales_rep": "李四", "region": "华北", "status": "已签约"}, {"client_name": "公司G", "amount": 180000, "product": "标准版套餐", "date": "2025-07-12", "sales_rep": "王五", "region": "华中", "status": "跟进中"} ] ``` 图6输入JSON销售数据 ![](https://support.huaweicloud.com/bestpractice-agentarts/zh-cn_image_0000002717389595.png "点击放大") 2. 在"试运行"对话框中，输入"请转换"，效果如[图7]所示。 图7效果示例 ![](https://support.huaweicloud.com/bestpractice-agentarts/zh-cn_image_0000002687949840.png "点击放大") #### 步骤五：提交版本 1. 工作流试运行成功后，在工作流编排页面，单击右上角"提交版本"。 2. 输入版本号与版本描述，如[图8]所示。 图8提交版本 ![](https://support.huaweicloud.com/bestpractice-agentarts/zh-cn_image_0000002687949378.png "点击放大") #### 相关文档 - 编排工作流时，画布相关的详细操作介绍，请参见[创建工作流](https://support.huaweicloud.com/lowcode-agentarts/agentarts_05_0047.html)。 - 在创建和运行工作流时，可能会遇到的常见问题，请参见《[常见问题](https://support.huaweicloud.com/agentarts_faq/agentarts_09_0001.html)》。 - 工作流创建完成后，如果需要通过API调用，则需要将提交的版本部署至实例，具体操作请参考[发布并部署工作流](https://support.huaweicloud.com/lowcode-agentarts/agentarts_05_0053.html)。