看到 CodeArts IDE 里内置的 Snap 智能编程助手，代码补全和对话看着不错。个人开发者（非企业）能免费使用吗？在哪里申请开通？

2026华为开发者大赛·小红书Skill创新赛【果办OfficeAce】 只需三步，果办OfficeAce即可帮你轻松完成 Skill 作品的创作以及提交。了解什么是果办OfficeAce?果办OfficeAce 内置了技能开发能力，你可以通过自然语言的方式与智能体对话，轻松创建属于你的 Skill 作品，并通过对话的方式实现作品提交，只需三步即可完成。 1、轻松准备：果办OfficeAce支持Windows、macOS、鸿蒙电脑客户端，下载安装并登录后订阅个人体验版，可免费使用。官网获取下载包：link图1 2、开始创作：在使用果办OfficeAce过程中，可通过2种方式快速创建skill：方式一：直接创建 Skill：打开果办OfficeAce，在对话框中输入你的创意和要求，即可实现 Skill 创作。生成完成后，可在果办OfficeAce的技能列表-我的技能中查看新创建的 Skill。提示词示例：帮我创建一个 Skill，我输入一款产品名和卖点，自动生成小红书风格的种草笔记，包括场景化开头、真实体验感正文、痛点共鸣点和带 emoji 的话题标签。图2图3图4方式二：从任务中沉淀 Skill：先让果办OfficeAce帮你完成一件事，比如直接对它说“帮我给这款冷萃咖啡液写一篇小红书种草笔记，目标人群是每天靠咖啡续命但没空排队买咖啡的打工人，突出 3 秒速溶和 0 糖 0 脂两个卖点，风格要真实体验感、不要硬广。”图5图6之后可以多轮调整结果，确认完成后，对果办OfficeAce说“把上面的经验沉淀为skill，便于后续复用。”果办OfficeAce就会把刚才的种草文案写法（目标人群定位、卖点提炼、场景化开头、真实体验感表达、话题标签组织等）提炼成一个可复用的种草笔记Skill，以后只需输入产品名和卖点，就能直接生成同款质量的小红书种草笔记。图7图8· 亮点功能（推荐使用）：开启“技能自演进”，让 Skill 越用越好：Skill 在执行任务遇到报错或被你纠正时，果办OfficeAce会自动从中提炼经验并更新技能，持续提升执行准确率和任务完成质量，工作流程分为三步：触发：Skill 执行任务时遇到报错，或你对输出进行纠正；提炼：OfficeAce 从错误或纠正中提取关键经验（如指令措辞修正、参数取值调整、操作流程优化等），生成技能更新内容；生效：按你配置的自演进策略，更新内容自动生效，或经你确认后再生效。自演进策略可按需配置：“自动生效”（演进内容直接生效，适合成熟技能无感优化）、“询问”（生成建议供你确认后再生效，适合关键技能人工把关）、“关闭”（该技能不演进），单个技能的策略优先级高于全局策略。你还可以在技能详情页查看版本历史，随时回滚或合并历史版本，用得越多，Skill 越懂你。图9 3、提交作品：当作品完成开发，你可以直接在对话窗口中对果办OfficeAce说：“我正在参加小红书Skill创新赛，帮我把这个Skill发布到华为云Skill市场和小红书Skill市场。”· 果办OfficeAce会帮你把已创建好的Skill直接上传到“华为云 Skill 市场”，过程中会有一些询问信息，按照实际确认即可。上传成功后，果办将返回该 Skill 在华为云 Skill 市场的相关信息及链接。图10图11· 点击果办OfficeAce提供的小红书 Skill Hub 链接，进入该页面上传已打包好的Skill压缩包，按页面提示填写作品信息并提交，即可完成。图12   相关链接：2026华为开发者大赛·小红书Skill创新赛 专题页面

 CEIC2026消费电子创新大会/1024程序员节·果办OfficeAce CodeLabs华为云果办OfficeAce办公智能体场景实战：从想法到交付，AI搭档现场上岗   一、实操案例介绍1.1 实操内容概述果办OfficeAce是华为云办公场景专属桌面AI助手，面向业务人员提供专家中心、灵魂与技能定制、多专家 @协作和 PPT 引导生成四大核心能力。本案例将指导开发者完成"新品上市作战室"完整场景实操，深度体验 AI 专家团队从深度调研到思辨验证到专业交付的完整工作流。面向角色：业务人员 / 产品经理 / 市场运营难度：零门槛实操预计用时：60-90 分钟1.2 果办OfficeAce 核心能力概览果办OfficeAce 是华为云推出的办公场景专属桌面 AI 助手，不同于传统对话式 AI，它能直接操控软件、处理文件、执行流程，是坐在你桌面上的数字办公专家。基于华为云智果 AgentArts 智能体平台底座，具备企业级安全合规保障。能力名称说明能力 1🎯 专家中心开箱即用的智能体团队，预置洞察研究专家、产品研发专家、商务销售专家等多角色，无需配置即可发起任务。能力 2🧬 灵魂与技能定制为智能体定义人设（SOUL）、配置专业技能（SKILL），打造符合企业业务需求的专属专家团队。能力 3🤝 专家团会诊单独开启专家团会话，组长统筹拆解任务、多专家并行处理、交叉校验，模拟跨部门协作，降低单一视角偏差。能力 4📊 PPT 生成版式规划引擎确保文本、图表、布局全要素可编辑；对话式直接生成，支持对话式微调。二、前置准备条件说明Windows 10/11操作系统环境华为云账号用于登录使用直接下载使用下载即用安装步骤：下载 果办OfficeAce 客户端访问产品页点击"下载 Windows 客户端"，获取安装包（如 OfficeClaw-0.2.0-windows-x64-setup.exe，版本号可能更新）安装并启动双击安装包 → 选择安装目录（建议默认路径）→ 点击"安装" → 等待完成 → 点击"完成"启动登录使用启动客户端 → 使用华为云账号登录（支持扫码或账号密码）→ 登录成功即可直接使用三、实战场景：新品上市作战室场景背景：你是一家科技公司的产品负责人，公司即将推出一款全新的智能办公产品。从零开始，你需要完成行业趋势研究、产品策略推演、上市方案制定，最终交付一份高质量的 PPT 汇报材料——全部在 OfficeAce 中完成。📊 行业洞察 → 🧠 策略推演 → 📑 方案制定 → 📊 PPT 交付通过这一完整场景，你将逐步掌握深度研究、专家团会诊、版式精调和全渠道远程协作四大实践能力，理解 AI 专家团队从"想法"到"交付"的完整闭环。3.1 体验点 1：Deep Research 深度研究与内容溯源⏱️预计用时：约 15 分钟向洞察研究专家发起"新品上市行业趋势"深度研究任务，果办OfficeAce 自研搜索引擎多维度检索、分阶段执行、交叉验证，产出有逻辑有实证的研究报告；每条结论可溯源至原始来源，理解"不是让一个模型编，而是让多源证据说话"的研究范式。操作步骤：打开 果办OfficeAce 主界面，选择"市场洞察分析师"在专家中心找到"市场洞察分析师"智能体，点击进入对话窗口。该专家预置了深度研究能力，可直接发起研究任务。发起深度研究任务在对话窗口输入："请对'智能办公行业2026年发展趋势'进行深度研究，覆盖市场规模、竞品分析、用户需求变化、技术演进方向四个维度"。提示：研究任务越具体，产出质量越高。可指定行业、时间范围、关注维度观察多阶段执行过程OfficeAce 将自动进行：① 关键词扩展与检索策略规划 → ② 多源并行检索（行业报告、新闻、论文、论坛）→ ③ 信息提取与交叉验证 → ④ 结构化报告生成。可在界面上实时查看各阶段进度查看研究报告与内容溯源研究完成后，查看结构化报告。点击任意结论，可查看该结论的原始来源链接——包括数据出处、报告来源、引用文献，验证"有实证、可溯源"。追问与迭代基于报告内容追问细节，如"竞品中谁在 AI PPT 方面布局最深？"，专家将基于已有研究上下文给出有据可查的回答，并标注新引用来源。体验收获：理解 Deep Research 的"多源证据说话"范式，掌握如何向研究专家发起高质量研究任务，体验从"AI 编内容"到"AI 查证据"的范式转变。  3.2 体验点 2：专家团会诊——新品上市策略推演⏱️ 约 15 分钟在体验点 1 中，市场洞察分析师已完成行业趋势深度研究。现在进入专家广场，直接使用平台预置的"产品商业化团队"，在独立的专家团会话中将研究结论作为背景输入，由团队中产品商业化负责人、市场机会分析师、商业模式策略师、定价与财务分析师等多位专家从各自专业视角进行策略推演，产出结构化的新品上市策略评估报告。操作步骤：进入专家广场，选择预置团队 - 打开 OfficeAce，进入专家广场页面，在"广场"标签页浏览预置团队卡片。 - 找到"产品商业化团队"，点击团队卡片进入专家团对话窗口。💡 提示："产品商业化团队"是 OfficeAce 出厂预置团队，由产品商业化负责人领衔，包含市场机会分析师、商业模式策略师、定价与财务分析师、上市与销售增长专家、商业交付与签约专家五位成员，专攻产品商业化全流程。无需创建，点击即用。 - 💡 也可以切换到"我的"标签页，点击"新建"→"新建专家团"，自行选择专家成员组建团队，适用于预置团队不完全匹配业务场景的情况。发起策略推演任务 - 在专家团对话窗口输入以下内容（以下数据为示例，实际操作时请替换为体验点 1 研究报告中你得出的关键结论）："我们计划推出一款面向中小企业的智能办公助手。根据前期行业研究，智能办公市场2026年规模预计突破百亿，竞品定价区间在99-299元/月，用户最关注的核心能力是文档智能处理和多渠道协作。请专家团从产品定位、商业模式、定价策略、上市节奏等维度进行策略推演，给出可行性评估和风险提示。"💡 提示：将体验点 1 的研究关键结论带入专家团会话，作为策略推演的事实依据。也可将研究报告导出后拖入对话窗口，专家团自动读取。观察专家团协作过程 - 组长（产品商业化负责人）接收任务后，自动拆解并分派给对应专家：① 市场机会分析师评估市场机会与竞争格局 → ② 商业模式策略师设计商业模式与价值主张 → ③ 定价与财务分析师测算定价与单位经济 → ④ 上市与销售增长专家规划首批客户与渠道切入 → ⑤ 各专家在独立上下文中并行处理 → ⑥ 组长汇总多方意见，交叉校验，标注分歧与共识。 参与讨论，引导深入 - 在专家团产出过程中，可随时追问或补充信息。如"竞品定价区间在 99-299 元/月，我们的成本结构能否支撑？"组长将把追问转发给相关专家，专家基于新信息调整分析。查看策略评估报告 - 专家团完成分析后，组长自动输出结构化的策略评估报告，包含：各方共识结论、关键分歧点、风险提示与行动建议。报告可导出为文档，作为后续 PPT 汇报的内容素材。📖 知识点： OfficeAce 预置专家团开箱即用，无需配置即可发起多专家协作。组长智能体负责任务拆解与汇总校验，多位专家成员在各自独立上下文中并行处理专业子任务。这种"组长统筹+专家并行"的架构既保证了各专家的分析深度（独立上下文不受其他专家 token 占用影响），又通过组长的交叉校验降低了单一视角的偏差。✅ 体验收获： 掌握预置专家团的使用方法，体验"组长统筹+专家并行"的多智能体协作架构，理解如何将前期研究结论作为背景输入驱动专家团策略推演。   3.3 体验点 3：生成 PPT——从研究到交付，版式规划引擎与风格精调⏱️ 约 20 分钟将体验点 1 的行业研究结论与体验点 2 的策略推演结果整合，直接在专家团会话中让专家生成一份面向高管汇报的 PPT。生成的 PPT 文本、图表、布局全要素可编辑，支持对话式微调，理解从"内容拼凑"到"版式合规"的专业交付闭环。操作步骤：在专家团会话中发起 PPT 生成任务 - 继续在体验点 2 的专家团会话中，输入："基于前面的行业研究结论和策略推演结果，生成一份面向高管汇报的新品上市 PPT，包含行业背景、产品定位、竞争分析、上市策略、风险与应对、总结展望六个章节"- 💡 提示：专家团已掌握本次会话中的研究背景与策略分析上下文，无需重复提供。也可将体验点 1 的研究报告导出后拖入对话窗口作为补充素材。查看生成的 PPT - 专家团根据策略分析结论自动生成 PPT：智能匹配内容与版式、自动排版文本与图表、确保全要素可编辑。生成完成后可直接预览。对话式微调 - 对生成的 PPT 通过对话进行微调，如“第14页优化文字显示效果”、"第3页增加一个竞品对比表格"、"把整体配色改为深蓝色系"、"第5页的风险分析再展开一些"。每次微调即时生效，无需重新生成。导出 PPT - 满意后导出 .pptx 文件，可直接在 PowerPoint 中打开编辑。 亮点：亮点说明📐 全要素可编辑生成的 PPT 中文本、图表、布局均为原生可编辑元素，非图片贴图🧩 智能版式匹配根据内容类型自动选择最佳版式：数据用图表、流程用图示、对比用表格💬 对话式迭代生成后可通过自然语言对话持续微调，无需重新生成✅ 体验收获： 掌握在对话中直接生成 PPT 的方法，体验从研究结论到策略分析到 PPT 交付的完整闭环，理解对话式微调的专业交付流程。   3.4 体验点 4：技能广场与全渠道远程协作⏱️预计用时：约 15 分钟用户从技能广场一键安装数据分析、文案撰写等新技能扩展专家能力；通过微信扫码直连 果办OfficeAce，在手机上远程操控电脑发文件、查数据、执行任务；同时绑定飞书/钉钉/小艺多渠道，出差途中也能随时指挥专家团队，理解"桌面专家 + 口袋遥控"的办公新范式。操作步骤：访问技能广场，安装新技能在 OfficeAce 左侧导航找到"技能广场"，浏览可用技能列表（如：AI Excel 数据分析、文案撰写、法律文书、邮件管理等）。点击"安装"一键添加，安装后对应专家立即获得新能力。提示：技能安装后即时生效，无需重启。可随时卸载或更新。体验 AI Excel 数据分析（可选）安装"AI Excel"技能后，将一张多表联动的 Excel 文件拖入对话窗口，输入"生成月度利润透视表"。OfficeAce 自动识别表间关联、整合数据、生成透视表——以前 VLOOKUP 套来套去的操作现在一句话搞定。3、 微信扫码直连 OfficeAce在 OfficeAce 设置中找到"渠道绑定"→"微信"，用手机微信扫描二维码完成绑定。绑定后，在手机微信中即可与 OfficeAce 对话，远程操控电脑：发文件、查数据、执行任务。4、 绑定飞书/钉钉/小艺多渠道在"渠道绑定"中继续绑定飞书、钉钉、小艺等渠道。绑定后，无论在哪个平台都能与同一组 OfficeAce 专家团队对话，上下文与记忆跨渠道同步。5、 远程指挥实战模拟出差场景：在手机微信中向 OfficeAce 发送"把刚才生成的 PPT 发到我邮箱""查一下今天有没有重要邮件需要回复"。体验在手机上远程操控桌面专家团队完成任务。  渠道接入方式：渠道接入方式能力💬 微信扫码一键直连远程操控电脑、发文件、查数据、执行任务🐦 飞书扫码一键直连消息聚合、任务分发、文件协作📌 钉钉账号授权绑定工作消息同步、日程管理、任务跟踪🎤 小艺语音助手接入语音指令操控、免手打交互 体验收获：理解"桌面专家 + 口袋遥控"的办公新范式，掌握技能广场扩展专家能力的方法，体验多渠道远程协作的便捷性。  四、实操收获收获说明📊 一份完整的研究报告有逻辑、有实证、可溯源的行业趋势研究报告，可直接用于决策参考🤝 一份策略评估报告专家团多视角推演后的新品上市策略分析，含共识、分歧、风险点与行动建议📑 一份专业级 PPT内容完整、排版专业、全要素可编辑的 .pptx 文件，可直接用于汇报🧠 OfficeAce 全流程操作技能掌握深度研究、专家团会诊、PPT 生成、技能安装、多渠道绑定的完整操作链路💡 AI 办公新范式认知理解"不是让 AI 编，而是让多源证据说话""桌面专家 + 口袋遥控"等新工作理念🚀 效率提升体感从行业研究到 PPT 交付全流程在 60-90 分钟内完成，传统方式需数天  五、果办OfficeAce 功能速查表功能模块核心能力本训练营对应体验专家中心开箱即用多角色智能体体验点 1、2Deep Research自研搜索引擎、多阶段执行、交叉验证、内容溯源体验点 1专家团会诊组长统筹+专家并行、交叉校验、降低单一视角偏差体验点 2持续记忆独立上下文管理、跨会话记忆留存体验点 2PPT 生成对话式直接生成、版式规划引擎、全要素可编辑体验点 3技能广场一键安装/卸载、持续扩展专家能力体验点 4多渠道接入微信/飞书扫码直连、钉钉/小艺绑定体验点 4AI Excel多表关联、自动透视、数据清洗体验点 4（可选）AI DocWord 文档一键生成、格式规范扩展能力安全合规工具调用安全护栏、数据脱敏加密、本地部署全程底层保障 从想法到交付，AI 搭档现场上岗通过"新品上市作战室"这一完整场景，你体验了 果办OfficeAce 中 AI 专家团队从深度调研到思辨验证到专业交付的完整工作流。不再是"一个人对着空白文档发呆"，而是"指挥一支 AI 专家团队协同作战"——研究有证据、讨论有碰撞、交付有版式、协作无边界。 📊 深度研究 🤝 专家团会诊 📑 版式精调 📱 全渠道协作 🧠 持续记忆 1024程序员节·果办OfficeAce CodeLabs果办OfficeAce 产品页：cid:link_0智果 AgentArts 智能体平台 · 华为云办公场景专属桌面 AI 助手

学习平台怎么选，卡组，题集等自建

赛事资源/更多官方课程/案例赛道一和赛道二的赛题解读链接有Bug跳转失败 

您好想问一下，这个比赛创新大赛官网产业赛道的报名也是截止到9月25日吗，还是具体听学校的通知安排呢？例如我们学校申报截止到9月8日，这意味着我们要在9月8日前完成所有的比赛项目内容吗，尤其是项目计划书这个很难短时间完成呢；或者是否允许先报名然后学校申报上，之后慢慢推进多次设计和调整呢？两个赛道看到需要提交具体的项目产品或者演示视频等等，但是现在只看到创新大赛官网上有个项目计划书的提交，所有东西都提交到这里吗，还是后期会开放其他的提交窗口？

想请教下，反复读了赛题说明还是有点懵。目前认为赛题一是要做一个完整可落地的智能体产品，赛题二就只需要开发一个符合规范的 Skill 技能包在华为智能体上跑就行。想问问实际两个赛道的要求到底是什么样？赛题二是不是只做个 Skill 就满足要求，还是也要做完整的应用？求了解的帮忙解答下，谢谢！

华为全联接大会2026·实战训练营华为云果办OfficeAce办公智能体场景实战：从想法到交付，AI搭档现场上岗  一、实操案例介绍1.1 实操内容概述OfficeAce是华为云办公场景专属桌面AI助手，面向业务人员提供专家中心、灵魂与技能定制、多专家 @协作和 PPT 引导生成四大核心能力。本案例将指导开发者完成"新品上市作战室"完整场景实操，深度体验 AI 专家团队从深度调研到思辨验证到专业交付的完整工作流。面向角色：业务人员 / 产品经理 / 市场运营难度：零门槛实操预计用时：60-90 分钟1.2 OfficeAce 核心能力概览OfficeAce 是华为云推出的办公场景专属桌面 AI 助手，不同于传统对话式 AI，它能直接操控软件、处理文件、执行流程，是坐在你桌面上的数字办公专家。基于华为云智果 AgentArts 智能体平台底座，具备企业级安全合规保障。能力名称说明能力 1🎯 专家中心开箱即用的智能体团队，预置洞察研究专家、产品研发专家、商务销售专家等多角色，无需配置即可发起任务。能力 2🧬 灵魂与技能定制为智能体定义人设（SOUL）、配置专业技能（SKILL），打造符合企业业务需求的专属专家团队。能力 3🤝 专家团会诊单独开启专家团会话，组长统筹拆解任务、多专家并行处理、交叉校验，模拟跨部门协作，降低单一视角偏差。能力 4📊 PPT 生成版式规划引擎确保文本、图表、布局全要素可编辑；对话式直接生成，支持对话式微调。二、前置准备条件说明Windows 10/11操作系统环境华为云账号用于登录使用直接下载使用下载即用安装步骤：下载 OfficeAce 客户端访问产品页点击"下载 Windows 客户端"，获取安装包（如 OfficeClaw-0.2.0-windows-x64-setup.exe，版本号可能更新）安装并启动双击安装包 → 选择安装目录（建议默认路径）→ 点击"安装" → 等待完成 → 点击"完成"启动登录使用启动客户端 → 使用华为云账号登录（支持扫码或账号密码）→ 登录成功即可直接使用三、实战场景：新品上市作战室场景背景：你是一家科技公司的产品负责人，公司即将推出一款全新的智能办公产品。从零开始，你需要完成行业趋势研究、产品策略推演、上市方案制定，最终交付一份高质量的 PPT 汇报材料——全部在 OfficeAce 中完成。📊 行业洞察 → 🧠 策略推演 → 📑 方案制定 → 📊 PPT 交付通过这一完整场景，你将逐步掌握深度研究、专家团会诊、版式精调和全渠道远程协作四大实践能力，理解 AI 专家团队从"想法"到"交付"的完整闭环。3.1 体验点 1：Deep Research 深度研究与内容溯源⏱️预计用时：约 15 分钟向洞察研究专家发起"新品上市行业趋势"深度研究任务，OfficeAce 自研搜索引擎多维度检索、分阶段执行、交叉验证，产出有逻辑有实证的研究报告；每条结论可溯源至原始来源，理解"不是让一个模型编，而是让多源证据说话"的研究范式。操作步骤：打开 OfficeAce 主界面，选择"市场洞察分析师"在专家中心找到"市场洞察分析师"智能体，点击进入对话窗口。该专家预置了深度研究能力，可直接发起研究任务。发起深度研究任务在对话窗口输入："请对'智能办公行业2026年发展趋势'进行深度研究，覆盖市场规模、竞品分析、用户需求变化、技术演进方向四个维度"。提示：研究任务越具体，产出质量越高。可指定行业、时间范围、关注维度观察多阶段执行过程OfficeAce 将自动进行：① 关键词扩展与检索策略规划 → ② 多源并行检索（行业报告、新闻、论文、论坛）→ ③ 信息提取与交叉验证 → ④ 结构化报告生成。可在界面上实时查看各阶段进度查看研究报告与内容溯源研究完成后，查看结构化报告。点击任意结论，可查看该结论的原始来源链接——包括数据出处、报告来源、引用文献，验证"有实证、可溯源"。追问与迭代基于报告内容追问细节，如"竞品中谁在 AI PPT 方面布局最深？"，专家将基于已有研究上下文给出有据可查的回答，并标注新引用来源。体验收获：理解 Deep Research 的"多源证据说话"范式，掌握如何向研究专家发起高质量研究任务，体验从"AI 编内容"到"AI 查证据"的范式转变。 3.2 体验点 2：专家团会诊——新品上市策略推演⏱️ 约 15 分钟在体验点 1 中，市场洞察分析师已完成行业趋势深度研究。现在进入专家广场，直接使用平台预置的"产品商业化团队"，在独立的专家团会话中将研究结论作为背景输入，由团队中产品商业化负责人、市场机会分析师、商业模式策略师、定价与财务分析师等多位专家从各自专业视角进行策略推演，产出结构化的新品上市策略评估报告。操作步骤：进入专家广场，选择预置团队 - 打开 OfficeAce，进入专家广场页面，在"广场"标签页浏览预置团队卡片。 - 找到"产品商业化团队"，点击团队卡片进入专家团对话窗口。💡 提示："产品商业化团队"是 OfficeAce 出厂预置团队，由产品商业化负责人领衔，包含市场机会分析师、商业模式策略师、定价与财务分析师、上市与销售增长专家、商业交付与签约专家五位成员，专攻产品商业化全流程。无需创建，点击即用。 - 💡 也可以切换到"我的"标签页，点击"新建"→"新建专家团"，自行选择专家成员组建团队，适用于预置团队不完全匹配业务场景的情况。发起策略推演任务 - 在专家团对话窗口输入以下内容（以下数据为示例，实际操作时请替换为体验点 1 研究报告中你得出的关键结论）："我们计划推出一款面向中小企业的智能办公助手。根据前期行业研究，智能办公市场2026年规模预计突破百亿，竞品定价区间在99-299元/月，用户最关注的核心能力是文档智能处理和多渠道协作。请专家团从产品定位、商业模式、定价策略、上市节奏等维度进行策略推演，给出可行性评估和风险提示。"💡 提示：将体验点 1 的研究关键结论带入专家团会话，作为策略推演的事实依据。也可将研究报告导出后拖入对话窗口，专家团自动读取。观察专家团协作过程 - 组长（产品商业化负责人）接收任务后，自动拆解并分派给对应专家：① 市场机会分析师评估市场机会与竞争格局 → ② 商业模式策略师设计商业模式与价值主张 → ③ 定价与财务分析师测算定价与单位经济 → ④ 上市与销售增长专家规划首批客户与渠道切入 → ⑤ 各专家在独立上下文中并行处理 → ⑥ 组长汇总多方意见，交叉校验，标注分歧与共识。 参与讨论，引导深入 - 在专家团产出过程中，可随时追问或补充信息。如"竞品定价区间在 99-299 元/月，我们的成本结构能否支撑？"组长将把追问转发给相关专家，专家基于新信息调整分析。查看策略评估报告 - 专家团完成分析后，组长自动输出结构化的策略评估报告，包含：各方共识结论、关键分歧点、风险提示与行动建议。报告可导出为文档，作为后续 PPT 汇报的内容素材。📖 知识点： OfficeAce 预置专家团开箱即用，无需配置即可发起多专家协作。组长智能体负责任务拆解与汇总校验，多位专家成员在各自独立上下文中并行处理专业子任务。这种"组长统筹+专家并行"的架构既保证了各专家的分析深度（独立上下文不受其他专家 token 占用影响），又通过组长的交叉校验降低了单一视角的偏差。✅ 体验收获： 掌握预置专家团的使用方法，体验"组长统筹+专家并行"的多智能体协作架构，理解如何将前期研究结论作为背景输入驱动专家团策略推演。 3.3 体验点 3：生成 PPT——从研究到交付，版式规划引擎与风格精调⏱️ 约 20 分钟将体验点 1 的行业研究结论与体验点 2 的策略推演结果整合，直接在专家团会话中让专家生成一份面向高管汇报的 PPT。生成的 PPT 文本、图表、布局全要素可编辑，支持对话式微调，理解从"内容拼凑"到"版式合规"的专业交付闭环。操作步骤：在专家团会话中发起 PPT 生成任务 - 继续在体验点 2 的专家团会话中，输入："基于前面的行业研究结论和策略推演结果，生成一份面向高管汇报的新品上市 PPT，包含行业背景、产品定位、竞争分析、上市策略、风险与应对、总结展望六个章节"- 💡 提示：专家团已掌握本次会话中的研究背景与策略分析上下文，无需重复提供。也可将体验点 1 的研究报告导出后拖入对话窗口作为补充素材。查看生成的 PPT - 专家团根据策略分析结论自动生成 PPT：智能匹配内容与版式、自动排版文本与图表、确保全要素可编辑。生成完成后可直接预览。对话式微调 - 对生成的 PPT 通过对话进行微调，如“第14页优化文字显示效果”、"第3页增加一个竞品对比表格"、"把整体配色改为深蓝色系"、"第5页的风险分析再展开一些"。每次微调即时生效，无需重新生成。导出 PPT - 满意后导出 .pptx 文件，可直接在 PowerPoint 中打开编辑。 亮点：亮点说明📐 全要素可编辑生成的 PPT 中文本、图表、布局均为原生可编辑元素，非图片贴图🧩 智能版式匹配根据内容类型自动选择最佳版式：数据用图表、流程用图示、对比用表格💬 对话式迭代生成后可通过自然语言对话持续微调，无需重新生成✅ 体验收获： 掌握在对话中直接生成 PPT 的方法，体验从研究结论到策略分析到 PPT 交付的完整闭环，理解对话式微调的专业交付流程。 3.4 体验点 4：技能广场与全渠道远程协作⏱️预计用时：约 15 分钟用户从技能广场一键安装数据分析、文案撰写等新技能扩展专家能力；通过微信扫码直连 OfficeAce，在手机上远程操控电脑发文件、查数据、执行任务；同时绑定飞书/钉钉/小艺多渠道，出差途中也能随时指挥专家团队，理解"桌面专家 + 口袋遥控"的办公新范式。操作步骤：访问技能广场，安装新技能在 OfficeAce 左侧导航找到"技能广场"，浏览可用技能列表（如：AI Excel 数据分析、文案撰写、法律文书、邮件管理等）。点击"安装"一键添加，安装后对应专家立即获得新能力。提示：技能安装后即时生效，无需重启。可随时卸载或更新。体验 AI Excel 数据分析（可选）安装"AI Excel"技能后，将一张多表联动的 Excel 文件拖入对话窗口，输入"生成月度利润透视表"。OfficeAce 自动识别表间关联、整合数据、生成透视表——以前 VLOOKUP 套来套去的操作现在一句话搞定。3、 微信扫码直连 OfficeAce在 OfficeAce 设置中找到"渠道绑定"→"微信"，用手机微信扫描二维码完成绑定。绑定后，在手机微信中即可与 OfficeAce 对话，远程操控电脑：发文件、查数据、执行任务。4、 绑定飞书/钉钉/小艺多渠道在"渠道绑定"中继续绑定飞书、钉钉、小艺等渠道。绑定后，无论在哪个平台都能与同一组 OfficeAce 专家团队对话，上下文与记忆跨渠道同步。5、 远程指挥实战模拟出差场景：在手机微信中向 OfficeAce 发送"把刚才生成的 PPT 发到我邮箱""查一下今天有没有重要邮件需要回复"。体验在手机上远程操控桌面专家团队完成任务。  渠道接入方式：渠道接入方式能力💬 微信扫码一键直连远程操控电脑、发文件、查数据、执行任务🐦 飞书扫码一键直连消息聚合、任务分发、文件协作📌 钉钉账号授权绑定工作消息同步、日程管理、任务跟踪🎤 小艺语音助手接入语音指令操控、免手打交互 体验收获：理解"桌面专家 + 口袋遥控"的办公新范式，掌握技能广场扩展专家能力的方法，体验多渠道远程协作的便捷性。 四、实操收获收获说明📊 一份完整的研究报告有逻辑、有实证、可溯源的行业趋势研究报告，可直接用于决策参考🤝 一份策略评估报告专家团多视角推演后的新品上市策略分析，含共识、分歧、风险点与行动建议📑 一份专业级 PPT内容完整、排版专业、全要素可编辑的 .pptx 文件，可直接用于汇报🧠 OfficeAce 全流程操作技能掌握深度研究、专家团会诊、PPT 生成、技能安装、多渠道绑定的完整操作链路💡 AI 办公新范式认知理解"不是让 AI 编，而是让多源证据说话""桌面专家 + 口袋遥控"等新工作理念🚀 效率提升体感从行业研究到 PPT 交付全流程在 60-90 分钟内完成，传统方式需数天  五、OfficeAce 功能速查表功能模块核心能力本训练营对应体验专家中心开箱即用多角色智能体体验点 1、2Deep Research自研搜索引擎、多阶段执行、交叉验证、内容溯源体验点 1专家团会诊组长统筹+专家并行、交叉校验、降低单一视角偏差体验点 2持续记忆独立上下文管理、跨会话记忆留存体验点 2PPT 生成对话式直接生成、版式规划引擎、全要素可编辑体验点 3技能广场一键安装/卸载、持续扩展专家能力体验点 4多渠道接入微信/飞书扫码直连、钉钉/小艺绑定体验点 4AI Excel多表关联、自动透视、数据清洗体验点 4（可选）AI DocWord 文档一键生成、格式规范扩展能力安全合规工具调用安全护栏、数据脱敏加密、本地部署全程底层保障 从想法到交付，AI 搭档现场上岗通过"新品上市作战室"这一完整场景，你体验了 OfficeAce 中 AI 专家团队从深度调研到思辨验证到专业交付的完整工作流。不再是"一个人对着空白文档发呆"，而是"指挥一支 AI 专家团队协同作战"——研究有证据、讨论有碰撞、交付有版式、协作无边界。 📊 深度研究 🤝 专家团会诊 📑 版式精调 📱 全渠道协作 🧠 持续记忆 华为全联接大会2026·实战训练营果办OfficeAce 产品页：cid:link_0智果 AgentArts 智能体平台 · 华为云办公场景专属桌面 AI 助手

每个文件都无法打开，还怎么工作

  这几天开始思考过程经常出现乱码 或 循环操作文件的时候路径的时候里面也会夹杂的乱码字符  

一点击×就直接叉掉了，但是智能体还在运行的过程中，可以完善一下，如果智能体还在运行，这个×点击的时候系统提示一下，确定要关闭吗

华为云智果AgentArts金融信贷AI智能体实战通过了解华为云智果AgentArts智能体平台、RAG检索增强及工作流编排等核心技术，实操搭建合规高效的企业信贷审核智能体。点击链接，即刻体验：cid:link_0   

前言OpenHarmony 轻量系统的固件构建通常会卡在三个地方：构建机架构、预编译工具链、源码和依赖下载。尤其是像 xiaohong (atomgit 开源的软硬件一体项目) 这类面向 WS63 芯片的项目，工具链、Python 依赖、repo 同步、clang 路径等细节只要有一处没对齐，就很容易在编译中途报错。这次我们尝试把 xiaohong 固件构建流程放到华为云上完成：通过 AI Shell 创建云端 ECS、准备构建环境、下载源码、配置工具链并完成固件编译。整体体验下来，AI Shell 比较适合这类“步骤多、依赖多、容易踩坑但流程可沉淀”的云端自动化任务。案例介绍xiaohong 是基于 OpenHarmony 的迷你系统，专为 WS63 芯片设计。本案例介绍如何使用华为云 AI Shell 从零开始编译 xiaohong 固件，包括环境准备、源码下载、工具链配置、编译过程以及常见问题解决方案。最终产物为：ws63-liteos-app_all.fwpkg：完整固件ws63-liteos-app_load_only.fwpkg：仅加载固件为什么适合用 AI Shell这个任务并不是单条命令能解决的问题，而是一个典型的云端构建流水线：需要创建合适规格和架构的 ECS。需要安装大量系统依赖和 Python 依赖。需要同步较大的 OpenHarmony 相关源码和预编译工具。需要处理工具链路径、软链接和环境变量。编译完成后还要下载固件并清理云资源。如果每次都手工操作，容易遗漏步骤。使用 AI Shell 的价值在于：可以用自然语言描述目标，再把重复流程沉淀为 skill 或脚本，后续复用时只需要发起任务即可。核心要点架构选择：必须使用 x86_64 架构的 ECS，预编译工具链不兼容 aarch64。源码下载：使用 repo 工具，并在 repo init 时添加 --git-lfs。工具链配置：需要配置 RISC-V 编译器 PATH，并创建 clang 软链接。Python 依赖：需要安装 kconfiglib、pycparser、markupsafe 等模块。构建命令：使用 ./build.sh --product-name xiaohong --gn-args is_debug=false。资源清理：构建完成后及时保存固件并释放 ECS，避免资源持续计费。适用对象企业个人开发者高校学生案例时间本案例总时长预计60分钟。资源总览创建华为云资源需要收费，请按需充值。本案例中编译环境所创建的云资源预计花费10元。资源名称规格单价（元）AI Shell体验版免费华为云资源按需10整体流程流程可以理解为：用户在开发者空间向 AI Shell 下达编译 xiaohong 固件任务。AI Shell 调用 xiaohong build skill 或自动化脚本。AI Shell 创建 x86_64 ECS 并准备构建环境。ECS 完成源码同步、工具链配置、固件编译。用户下载固件产物，并按需销毁 ECS。案例步骤0. 进入 AIShell参考教程探索智能 Shell 交互新范式 详解 AI Shell 完整用法或者产品文档AI Shell，云上开发运维效率升级 进入到 AI Shell 界面进入到 AIShell 我们应该能看如下界面，和我们常用的 OpenCode、AtomCode 等产品类似，可以输入：1. 使用 xiaohong-build-skill此处，我们不再介绍 AIShell 的详细功能和能力，留给大家自行探索。首先我们输入/确认一下所有功能是否都已经开启（默认是开启了所有的功能），如下图：虽然我们一句话就能实现 xiaohong 编译环境搭建、编译运行，但是为了了解背后的细节，我们先让 AIShell 帮我熟悉熟悉：帮我看看这是什么： https://atomgit.com/huqi/xiaohong-build-skill，如何使用？此处我们理解权限安全规则，这里按需选择 Allow once 或者 Allow always，我选择的是 Allow always，可以左右键切换选择并回车确认：接着我们就让 AIShell 执行这个 skill,如果遇到：帮我实际运行这个 skill 来编译 xiaohong 固件从右侧任务列表我们可以看到类似的：检查前置条件检查华为云凭证创建 ESC 实例构建配置环境下载源码编译固件下载固件到本地清理资源如果我们的华为云账号有幸绑定了短信，在 ECS 创建完我们也能收到短信，当然也能去控制台查看，类似：接下来只需静静等待 Task 被一一执行完。理想情况下，我们会看到 AI Shell 会继续自动执行下去，比如进入到 ECS 中安装依赖：比如下载源码：最终能看到编译完成：2. 下载固件下载固件的方式有很多种，比如让 AIShell 上传到 OBS ，当然我们也可以去 ESC 实例里手动下载：3. 后续后续可以让 AIShell 指导我们烧录固件：题外话：让 AIShell 帮我修改源码，重新编译固件写入我专属的引导语释放资源最后记得让 AIShell 释放资源：帮我释放这次创建的所有资源Q&A⚠️ The maximum number of model requests in a single turn is exceeded原因是触发了 MaaS 的限流，只需回复 “继续” 就行如果我开发的不是 xiaohong 而是其他平台如 小智 等，那怎么办？在我们看来，底层逻辑都是相通的，我们的目的是搭建编译环境–编译–获取产物，理论上只需要把相关的指导文档发给 AIShell 就行，类似的： 我想搭建环境编译 https://github.com/78/xiaozhi-esp32 ，应该怎么做？本文正在参与：【案例共创】【第12期】基于华为云AI Shell完成云资源管理、云服务运维和应用部署

随着企业大模型应用从“原型试验”走向“核心生产场景”，如何保障智能体在复杂业务环境下的安全运行、高效运维与精准资产调用，成为企业规模化部署的挑战。AgentArts正是为此而生的企业级智能体平台。什么是智果（AgentArts）智能体平台？智果（AgentArts）智能体平台是面向企业级核心业务场景打造的智能体平台，基于Harness最佳实践构建标准化的开发与治理框架，专注于解决智能体在企业复杂环境下落地难、运行不可靠、运维无抓手等痛点，致力于降低应用门槛，助力企业将AI能力与实际业务深度融合，实现规模化落地。智果（AgentArts）智能体平台向下兼容业界主流开源大模型，向上提供安全可靠的企业级运行底座、长短期记忆、海量行业资产精准调用与全链路运营运维体系，同时内置OfficeAce办公生产力套件，帮助企业快速组建高可信的AI数字员工团队，让大模型真正转化为企业核心生产力。为什么选择智果（AgentArts）智能体平台？企业级智能体的真正价值不仅在于“被开发出来”，更在于在生产环境中“稳定运行”与“持续演进”。企业级安全运行底座安全可靠运行沙箱：针对智能体调用工具及执行代码的场景，提供完全隔离的生产级安全沙箱，严防命令注入与数据外泄，确保企业核心资产安全。弹性高可用调度：承载高并发的业务请求，底层资源支持10ms极速冷启动，从容应对从0到万级并发的流量洪峰，关键业务响应延迟低至500ms，保障企业级智能体在峰值业务场景下的长程稳定运行。企业级多层级记忆与长程复杂任务解锁多层级记忆网络：独创结合短期会话、中层业务上下文与企业全局长期知识库的记忆架构。智能体不仅能“记住”当前的对话，更能理解企业复杂的历史业务背景。海量资产精准调用与高性能网关精准意图路由与参数注入：企业内部成百上千的异构API与行业资产，平台不仅具备海量托管能力，更依靠高精度语义检索与意图识别引擎，实现工具的精准匹配与参数自动注入，解决大模型“工具多选错、参数传不对”的生产级痛点。全链路观测与自动化评估端到端监控：提供覆盖全局的日志、调用链追踪与应用指标分析，摆脱“黑盒”运行，出现异常精准定位至具体节点。量化自动评估：预置高精准评估器，基于企业真实数据集自动打分，直观验证智能体水平与稳定性，驱动业务持续演进。OfficeAce专业办公原生集成多渠道开箱即用：预置多场景精选办公Skills，支持一键无缝接入微信、飞书、钉钉等核心协同办公渠道。本地安全高效协同：具备本地文件安全读写能力，直接在本地高效处理Word、Excel、PPT、PDF等多格式文件，隐私不出户，效率最大化。欢迎来华为开发者空间体验案例，留下您的宝贵意见或建议。

一.前言说明（文章最后有部署的python代码）本文介绍了如何通过部署在腾讯云CloudBase上的桥接服务，打通企业微信应用（机器人）与AgentArts运行时的方法。并列举了使用华为云FunctionGraph的方案的差异（未测试）。以下为简单流程：企业微信应用（机器人） → AgentArts 桥接服务（华为云FunctionGrah或者腾讯云CloudBase） → AgentArts运行时 二.为什么需要桥接服务？1、企业微信机器人无法直接调用AgentArts的原因a.协议不兼容维度企业微信机器人AgentArts 智能体运行时消息格式JSON/XML 加密信封REST API JSON认证方式AES-256-CBC + SHA1签名Bearer Token通信模式被动回调（企业微信→服务）主动调用（服务→AgentArts）响应要求必须在 5秒内 返回可能需要 5-30秒 响应 b.安全机制冲突- 企业微信 ：所有消息必须使用 EncodingAESKey 加密，防止中间人攻击- AgentArts ：使用 API Key + Bearer Token 进行身份验证- 两者的安全协议完全独立，无法直接互通 c. 架构模式差异数据格式不匹配- 企业微信 ：发送的是加密的消息对象（包含用户ID、消息类型、时间戳等）- AgentArts ：期望接收简单的 {"input": "用户问题"} 格式- 两者的数据结构完全不同，需要转换   2、AgentArts桥接服务的必要性a.桥接服务的核心作用b.桥接服务承担的职责职责说明消息解密使用企业微信的 Token 和 AESKey 解密收到的加密消息格式转换将企业微信消息格式转换为 AgentArts API 所需的格式会话管理使用用户ID作为Session ID，保持多轮对话记忆异步处理处理 AgentArts 可能的长响应时间（超过5秒）消息加密将 AgentArts 的回复加密后返回给企业微信签名验证验证企业微信请求的合法性，防止伪造请求 c.5秒超时问题的解决方案方案A：同步模式（适合快速响应）用户消息 → 桥接服务 → AgentArts → 立即返回加密响应← 5秒内完成方案B：异步模式（适合长耗时任务）用户消息 → 桥接服务 → 立即返回空响应（告诉企业微信已收到）                              ↓后台线程处理 → AgentArts → 通过企业微信API主动推送回复3、桥接模式必要性的总结a.无法直接调用的根本原因：1. 协议层 ：企业微信使用加密回调，AgentArts使用REST API2. 安全层 ：认证方式和加解密机制完全不同3. 时间层 ：企业微信要求5秒响应，AgentArts可能需要更长时间b.桥接服务的必要性：- 作为 协议转换器 ：实现两种不同协议的互通- 作为 安全网关 ：处理双方的认证和加密- 作为 异步缓冲 ：解决响应超时问题- 作为 会话管理器 ：维护用户对话上下文 一句话总结 ：桥接服务是连接两个完全不同生态系统的桥梁，没有它，企业微信机器人和AgentArts智能体就像说两种不同语言的人，无法直接沟通。三. 企业微信机器人通过CloudBase打通AgentArts整体方案步骤1：配置企业微信自建应用3.1登录企业微信管理后台https://work.weixin.qq.com/wework_admin/frame#/3.2 创建自建应用   - 进入「应用管理」→「应用」→「自建」   - 点击「创建应用」，填写应用名称（如：AI助手）   - 上传应用图标3.3 获取应用配置      - 企业ID ：我的企业 → 企业信息 → 企业ID   - AgentId ：应用详情页 → AgentId   - Secret ：应用详情页 → 点击查看Secret4. 配置接收消息      - 进入应用详情页 → 「接收消息」   - 设置 Token （自定义字符串，如：my_wx_token）   - 点击「随机获取」生成 EncodingAESKey （43字符）   - 填写回调URL（稍后配置CloudBase后填写）需要注意的是部署在腾讯云或者华为云上的桥接服务，需要挂载企业域名，才可以通过腾讯企业微信的验证要求。5. 启用智能机器人功能   - 在应用详情页找到「智能机器人」设置   - 启用机器人功能，配置可见范围步骤2：部署CloudBase云函数1.    创建CloudBase环境   - 登录腾讯云控制台 → 云开发CloudBase   - 创建新环境（选择按量计费，免费额度充足）2. 创建云函数  - 进入「云函数」→「新建」   - 函数名称： wecom-bot-bridge   - 运行环境：Python 3.10   - 创建方式：空白函数3. 配置HTTP   开启HTTP访问为当前域名添加路由例如当前云函数路径为https://cloud1-7gmelj5udf09f32f-1409020909.ap-shanghai.app.tcloudbase.com添加的路由为/wangran-agentarts则访问路径为：https://cloud1-7gmelj5udf09f32f-1409020909.ap-shanghai.app.tcloudbase.com/wangran-agentarts4. 上传代码   - 复制 wecom-bot-tcb.py 的全部代码   - 粘贴到在线编辑器中   - 保存为 index.py   - 入口函数设置为： index.main5. 配置环境变量步骤3：配置AgentArts智能体1.确保智能体已部署- 运行时名称： runtime-17uneunb- 访问域名： defaultgw-mvstsmzsgv.cn-southwest-2.huaweicloud-agentarts.com- API Key： 3de5d906dfc24af98431ba4954f4b589以上数据为举例说明，需要根据实际情况自己更改2.验证云函数的可用性(已桥接智能体）curl.exe -x POST https://cloud1-7gmelj5udf09f32f-1409020909.ap-shanghai.app.tcloudbase.com/wangran-agentarts/test -H "Content-type: application/json" -d "{\"input\":"\"帮我指定一个3天的旅游计划"\"}" 步骤4：连接企业微信和CloudBase- 回到企业微信应用详情页 → 「接收消息」- 回调URL： https://xxx.tcb.qcloud.la/wecom-bot-bridge/callback- Token：与CloudBase环境变量中的 WECOM_TOKEN 一致- EncodingAESKey：与CloudBase环境变量中的 WECOM_ENCODING_AES_KEY 一致2.验证URL- 点击「保存」后，企业微信会发送GET请求验证URL- CloudBase云函数会自动处理验证请求- 验证成功后显示「验证通过」四、完整数据流向用户在群聊中 @机器人    │    ▼┌──────────────────────────────────┐│ 1. 企业微信服务器处理                                           │    • 识别@机器人消息                                            │    • 使用 AES-256-CBC 加密消息                                  │    • 生成 SHA1 签名                                             │    • POST 到 CloudBase 云函数                                   └──────────────────────────────────┘    │    ▼┌──────────────────────────────────┐│ 2. CloudBase 云函数接收                                         │    • HTTP 触发器接收 POST 请求                                  │    • 提取 query 参数：msg_signature, timestamp, nonce           │    • 解析 JSON 请求体：{"encrypt": "...", ...}                 └──────────────────────────────────┘    │    ▼┌──────────────────────────────────┐│ 3. 消息解密（纯Python实现，零依赖）                             │    • 验证 SHA1 签名                                             │    • 使用 EncodingAESKey 解密消息                               │    • 提取明文 JSON：                                            │      {                                                          │        "msgid": "...",                                         │        "msgtype": "text",                                      │        "from": {"userid": "zhangsan"},                         │        "text": {"content": "@AI助手 正方形蓝苹果是什么？"}     │      }                                                          └──────────────────────────────────┘    │    ▼┌──────────────────────────────────┐│ 4. 消息处理                                                     │    • 去重检查（基于 msgid）                                      │    • 过滤非文本消息                                             │    • 去除 @机器人 前缀                                          │    • 提取用户ID作为 session_id（保持多轮对话记忆）              └───────────────────────────────────┘    │    ▼┌───────────────────────────────────┐│ 5. 调用 AgentArts 智能体                                        │    • 使用 urllib 发送 HTTP 请求（零依赖）                      │    • 请求头：                                                   │      - Authorization: Bearer {API_KEY}                         │      - X-Hw-Agentarts-Session-Id: {user_id}                    │    • 请求体：{"input": "正方形蓝苹果是什么？"}                  └───────────────────────────────────┘    │    ▼┌───────────────────────────────────┐│ 6. AgentArts 智能体处理                                         │    • 查询知识库（正方形蓝苹果信息）                             │    • 检索记忆库（历史对话）                                     │    • 调用 LLM 生成回复                                         │    • 返回 JSON：{"response": "根据知识库信息..."}              └───────────────────────────────────┘    │    ▼┌───────────────────────────────────┐│ 7. CloudBase 处理响应                                          │    • 解析 AgentArts 返回的 JSON                                  │    • 提取 response 字段                                        │    • 构造回复消息：                                           │      {"msgtype": "text", "text": {"content": "回复内容"}}      └───────────────────────────────────┘    │    ▼┌───────────────────────────────────┐│ 8. 消息加密                                                     │    • 使用 AES-256-CBC 加密回复消息                             │    • 生成新的 SHA1 签名                                         │    • 构造 JSON 信封：                                          │      {                                                          │        "encrypt": "...",                                       │        "msgsignature": "...",                                  │        "timestamp": "...",                                     │        "nonce": "..."                                           │      }                                                          └───────────────────────────────────┘    │    ▼┌───────────────────────────────────┐│ 9. 返回给企业微信                                               │    • CloudBase 返回加密的 JSON 信封                             │    • HTTP 状态码：200                                          │    • Content-Type: application/json                            └───────────────────────────────────┘    │    ▼┌───────────────────────────────────┐│ 10. 企业微信展示给用户                                         │     • 解密消息                                                  │     • 在群聊中显示机器人的回复                                  └───────────────────────────────────┘五、华为云FunctionGraph替代腾讯云CloudBase的方案对比可以使用华为云FunctionGraph方案方案替代腾讯云CloudBase华为云方案未经验证，需要自行验证，本文仅提供测试代码。wecom-bot-fg.py为华为云FunctionGraph方案的部署代码wecom-bot-tcb.py为腾讯CloudBase云函数方案的部署代码   文件wecom-bot-fg.pywecom-bot-tcb.py部署平台华为云 FunctionGraph腾讯云 CloudBase消息格式JSON（2025新版）JSON（2025新版）回复模式同步（直接返回加密响应）同步（直接返回加密响应）依赖flask,requests,pycryptodome零依赖 （纯标准库）入口方式Flask WSGICloudBase HTTP触发器超时处理同步等待（需AgentArts快响应）同步等待适用场景华为云生态用户腾讯云生态/无依赖要求wecom-bot-tcb.py""" 企业微信智能机器人 → 腾讯云 CloudBase 云函数 → AgentArts 智能体 ============================================================== 本文件专为 CloudBase 云函数设计，使用原生 HTTP 触发器格式。 **零第三方依赖**：只使用 Python 标准库（urllib + hashlib + 内置加解密）。 说明： - 复制全部代码到 CloudBase 在线编辑器，保存为 index.py - 处理程序填: index.main - 不需要 requirements.txt（无外部依赖！） - 环境变量需要配置 7 个（见下方） """ import json import os import re import time import struct import base64 import hashlib import hmac import socket import ssl import logging import urllib.request from typing import Tuple, Optional # ============================================================================ # 配置 # ============================================================================ AGENTARTS_URL = os.getenv("AGENTARTS_URL", "https://defaultgw-mvstsmzsgv.cn-southwest-2.huaweicloud-agentarts.com" "/runtimes/runtime-17uneunb/invocations?endpoint=Latest") AGENTARTS_API_KEY = os.getenv("AGENTARTS_API_KEY", "3de5d906dfc24af98431bc4954f4b589") WECOM_CORP_ID = os.getenv("WECOM_CORP_ID", "") WECOM_TOKEN = os.getenv("WECOM_TOKEN", "") WECOM_ENCODING_AES_KEY = os.getenv("WECOM_ENCODING_AES_KEY", "") LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO") logging.basicConfig( level=getattr(logging, LOG_LEVEL.upper(), logging.INFO), format="%(asctime)s [%(levelname)s] %(message)s", ) logger = logging.getLogger("wecom-bot-tcb") # ============================================================================ # 纯 Python 实现 AES-256-CBC（无第三方依赖） # ============================================================================ class _AES256CBC: """ 纯 Python AES-256-CBC 实现。 仅依赖 hashlib（标准库），无需 pycryptodome / cryptography。 """ @staticmethod def _pad(data: bytes, block_size: int = 16) -> bytes: pad_len = block_size - (len(data) % block_size) return data + bytes([pad_len] * pad_len) @staticmethod def _unpad(data: bytes) -> bytes: pad_len = data[-1] if pad_len < 1 or pad_len > 16: raise ValueError(f"无效填充长度: {pad_len}") return data[:-pad_len] @staticmethod def _xor_bytes(a: bytes, b: bytes) -> bytes: return bytes(x ^ y for x, y in zip(a, b)) @classmethod def _aes_encrypt_block(cls, block: bytes, round_keys: list) -> bytes: """加密单个 16 字节块""" # AES-256 使用 14 轮 state = list(block) # AddRoundKey state = cls._xor_bytes(state, round_keys[0]) for r in range(1, 14): state = cls._sub_bytes(state) state = cls._shift_rows(state) state = cls._mix_columns(state) state = cls._xor_bytes(state, round_keys[r]) # Final round (no MixColumns) state = cls._sub_bytes(state) state = cls._shift_rows(state) state = cls._xor_bytes(state, round_keys[14]) return bytes(state) @classmethod def _aes_decrypt_block(cls, block: bytes, round_keys: list) -> bytes: """解密单个 16 字节块""" state = list(block) state = cls._xor_bytes(state, round_keys[14]) state = cls._inv_shift_rows(state) state = cls._inv_sub_bytes(state) for r in range(13, 0, -1): state = cls._xor_bytes(state, round_keys[r]) state = cls._inv_mix_columns(state) state = cls._inv_shift_rows(state) state = cls._inv_sub_bytes(state) state = cls._xor_bytes(state, round_keys[0]) return bytes(state) # ---- S-Box ---- _SBOX = [ 0x63,0x7c,0x77,0x7b,0xf2,0x6b,0x6f,0xc5,0x30,0x01,0x67,0x2b,0xfe,0xd7,0xab,0x76, 0xca,0x82,0xc9,0x7d,0xfa,0x59,0x47,0xf0,0xad,0xd4,0xa2,0xaf,0x9c,0xa4,0x72,0xc0, 0xb7,0xfd,0x93,0x26,0x36,0x3f,0xf7,0xcc,0x34,0xa5,0xe5,0xf1,0x71,0xd8,0x31,0x15, 0x04,0xc7,0x23,0xc3,0x18,0x96,0x05,0x9a,0x07,0x12,0x80,0xe2,0xeb,0x27,0xb2,0x75, 0x09,0x83,0x2c,0x1a,0x1b,0x6e,0x5a,0xa0,0x52,0x3b,0xd6,0xb3,0x29,0xe3,0x2f,0x84, 0x53,0xd1,0x00,0xed,0x20,0xfc,0xb1,0x5b,0x6a,0xcb,0xbe,0x39,0x4a,0x4c,0x58,0xcf, 0xd0,0xef,0xaa,0xfb,0x43,0x4d,0x33,0x85,0x45,0xf9,0x02,0x7f,0x50,0x3c,0x9f,0xa8, 0x51,0xa3,0x40,0x8f,0x92,0x9d,0x38,0xf5,0xbc,0xb6,0xda,0x21,0x10,0xff,0xf3,0xd2, 0xcd,0x0c,0x13,0xec,0x5f,0x97,0x44,0x17,0xc4,0xa7,0x7e,0x3d,0x64,0x5d,0x19,0x73, 0x60,0x81,0x4f,0xdc,0x22,0x2a,0x90,0x88,0x46,0xee,0xb8,0x14,0xde,0x5e,0x0b,0xdb, 0xe0,0x32,0x3a,0x0a,0x49,0x06,0x24,0x5c,0xc2,0xd3,0xac,0x62,0x91,0x95,0xe4,0x79, 0xe7,0xc8,0x37,0x6d,0x8d,0xd5,0x4e,0xa9,0x6c,0x56,0xf4,0xea,0x65,0x7a,0xae,0x08, 0xba,0x78,0x25,0x2e,0x1c,0xa6,0xb4,0xc6,0xe8,0xdd,0x74,0x1f,0x4b,0xbd,0x8b,0x8a, 0x70,0x3e,0xb5,0x66,0x48,0x03,0xf6,0x0e,0x61,0x35,0x57,0xb9,0x86,0xc1,0x1d,0x9e, 0xe1,0xf8,0x98,0x11,0x69,0xd9,0x8e,0x94,0x9b,0x1e,0x87,0xe9,0xce,0x55,0x28,0xdf, 0x8c,0xa1,0x89,0x0d,0xbf,0xe6,0x42,0x68,0x41,0x99,0x2d,0x0f,0xb0,0x54,0xbb,0x16, ] _INV_SBOX = [ 0x52,0x09,0x6a,0xd5,0x30,0x36,0xa5,0x38,0xbf,0x40,0xa3,0x9e,0x81,0xf3,0xd7,0xfb, 0x7c,0xe3,0x39,0x82,0x9b,0x2f,0xff,0x87,0x34,0x8e,0x43,0x44,0xc4,0xde,0xe9,0xcb, 0x54,0x7b,0x94,0x32,0xa6,0xc2,0x23,0x3d,0xee,0x4c,0x95,0x0b,0x42,0xfa,0xc3,0x4e, 0x08,0x2e,0xa1,0x66,0x28,0xd9,0x24,0xb2,0x76,0x5b,0xa2,0x49,0x6d,0x8b,0xd1,0x25, 0x72,0xf8,0xf6,0x64,0x86,0x68,0x98,0x16,0xd4,0xa4,0x5c,0xcc,0x5d,0x65,0xb6,0x92, 0x6c,0x70,0x48,0x50,0xfd,0xed,0xb9,0xda,0x5e,0x15,0x46,0x57,0xa7,0x8d,0x9d,0x84, 0x90,0xd8,0xab,0x00,0x8c,0xbc,0xd3,0x0a,0xf7,0xe4,0x58,0x05,0xb8,0xb3,0x45,0x06, 0xd0,0x2c,0x1e,0x8f,0xca,0x3f,0x0f,0x02,0xc1,0xaf,0xbd,0x03,0x01,0x13,0x8a,0x6b, 0x3a,0x91,0x11,0x41,0x4f,0x67,0xdc,0xea,0x97,0xf2,0xcf,0xce,0xf0,0xb4,0xe6,0x73, 0x96,0xac,0x74,0x22,0xe7,0xad,0x35,0x85,0xe2,0xf9,0x37,0xe8,0x1c,0x75,0xdf,0x6e, 0x47,0xf1,0x1a,0x71,0x1d,0x29,0xc5,0x89,0x6f,0xb7,0x62,0x0e,0xaa,0x18,0xbe,0x1b, 0xfc,0x56,0x3e,0x4b,0xc6,0xd2,0x79,0x20,0x9a,0xdb,0xc0,0xfe,0x78,0xcd,0x5a,0xf4, 0x1f,0xdd,0xa8,0x33,0x88,0x07,0xc7,0x31,0xb1,0x12,0x10,0x59,0x27,0x80,0xec,0x5f, 0x60,0x51,0x7f,0xa9,0x19,0xb5,0x4a,0x0d,0x2d,0xe5,0x7a,0x9f,0x93,0xc9,0x9c,0xef, 0xa0,0xe0,0x3b,0x4d,0xae,0x2a,0xf5,0xb0,0xc8,0xeb,0xbb,0x3c,0x83,0x53,0x99,0x61, 0x17,0x2b,0x04,0x7e,0xba,0x77,0xd6,0x26,0xe1,0x69,0x14,0x63,0x55,0x21,0x0c,0x7d, ] _RCON = [0x01,0x02,0x04,0x08,0x10,0x20,0x40,0x80,0x1b,0x36] @classmethod def _sub_bytes(cls, state: bytes) -> bytes: return bytes(cls._SBOX[b] for b in state) @classmethod def _inv_sub_bytes(cls, state: bytes) -> bytes: return bytes(cls._INV_SBOX[b] for b in state) @staticmethod def _shift_rows(state: bytes) -> bytes: return bytes([state[0], state[5], state[10], state[15], state[4], state[9], state[14], state[3], state[8], state[13], state[2], state[7], state[12], state[1], state[6], state[11]]) @staticmethod def _inv_shift_rows(state: bytes) -> bytes: return bytes([state[0], state[13], state[10], state[7], state[4], state[1], state[14], state[11], state[8], state[5], state[2], state[15], state[12], state[9], state[6], state[3]]) @staticmethod def _gmul(a: int, b: int) -> int: """GF(2^8) 乘法""" p = 0 for _ in range(8): if b & 1: p ^= a hi = a & 0x80 a = (a << 1) & 0xFF if hi: a ^= 0x1b b >>= 1 return p @classmethod def _mix_columns(cls, state: bytes) -> bytes: result = bytearray(16) for i in range(4): s0, s1, s2, s3 = state[i*4], state[i*4+1], state[i*4+2], state[i*4+3] result[i*4] = cls._gmul(2,s0) ^ cls._gmul(3,s1) ^ s2 ^ s3 result[i*4+1] = s0 ^ cls._gmul(2,s1) ^ cls._gmul(3,s2) ^ s3 result[i*4+2] = s0 ^ s1 ^ cls._gmul(2,s2) ^ cls._gmul(3,s3) result[i*4+3] = cls._gmul(3,s0) ^ s1 ^ s2 ^ cls._gmul(2,s3) return bytes(result) @classmethod def _inv_mix_columns(cls, state: bytes) -> bytes: result = bytearray(16) for i in range(4): s0, s1, s2, s3 = state[i*4], state[i*4+1], state[i*4+2], state[i*4+3] result[i*4] = cls._gmul(14,s0)^cls._gmul(11,s1)^cls._gmul(13,s2)^cls._gmul(9,s3) result[i*4+1] = cls._gmul(9,s0)^cls._gmul(14,s1)^cls._gmul(11,s2)^cls._gmul(13,s3) result[i*4+2] = cls._gmul(13,s0)^cls._gmul(9,s1)^cls._gmul(14,s2)^cls._gmul(11,s3) result[i*4+3] = cls._gmul(11,s0)^cls._gmul(13,s1)^cls._gmul(9,s2)^cls._gmul(14,s3) return bytes(result) @classmethod def _key_expansion(cls, key: bytes) -> list: """AES-256 密钥扩展 → 15 个轮密钥""" Nk, Nr = 8, 14 w = [] for i in range(Nk): w.append(key[i*4:(i+1)*4]) for i in range(Nk, 4 * (Nr + 1)): temp = w[i - 1] if i % Nk == 0: temp = cls._xor_bytes( bytes(cls._SBOX[b] for b in temp[1:] + temp[:1]), bytes([cls._RCON[i // Nk - 1], 0, 0, 0])) elif Nk > 6 and i % Nk == 4: temp = bytes(cls._SBOX[b] for b in temp) w.append(cls._xor_bytes(w[i - Nk], temp)) round_keys = [] for r in range(Nr + 1): round_keys.append(b"".join(w[r*4:(r+1)*4])) return round_keys @classmethod def encrypt_cbc(cls, plaintext: bytes, key: bytes, iv: bytes) -> bytes: """AES-256-CBC 加密""" padded = cls._pad(plaintext, 16) round_keys = cls._key_expansion(key) result = bytearray() prev = iv for i in range(0, len(padded), 16): block = cls._xor_bytes(padded[i:i+16], prev) encrypted = cls._aes_encrypt_block(block, round_keys) result.extend(encrypted) prev = encrypted return bytes(result) @classmethod def decrypt_cbc(cls, ciphertext: bytes, key: bytes, iv: bytes) -> bytes: """AES-256-CBC 解密""" round_keys = cls._key_expansion(key) result = bytearray() prev = iv for i in range(0, len(ciphertext), 16): decrypted = cls._aes_decrypt_block(ciphertext[i:i+16], round_keys) result.extend(cls._xor_bytes(decrypted, prev)) prev = ciphertext[i:i+16] return cls._unpad(bytes(result)) # ============================================================================ # 企业微信消息加解密（纯 Python 实现，零依赖） # ============================================================================ class WXBizMsgCrypt: """企业微信消息加解密：AES-256-CBC + SHA1 签名""" def __init__(self, token: str, encoding_aes_key: str, corp_id: str): self.token = token self.corp_id = corp_id self.aes_key = base64.b64decode(encoding_aes_key + "=") if len(self.aes_key) != 32: raise ValueError(f"AES 密钥长度错误: {len(self.aes_key)} (应为 32)") @staticmethod def _sha1(text: str) -> str: return hashlib.sha1(text.encode("utf-8")).hexdigest() def _make_signature(self, timestamp: str, nonce: str, encrypt_msg: str) -> str: return self._sha1("".join(sorted([self.token, timestamp, nonce, encrypt_msg]))) def _verify_signature(self, sig: str, ts: str, nonce: str, enc: str) -> bool: return self._make_signature(ts, nonce, enc) == sig def _decrypt(self, encrypt_b64: str) -> bytes: raw = _AES256CBC.decrypt_cbc( base64.b64decode(encrypt_b64), self.aes_key, self.aes_key[:16]) content = raw[16:] msg_len = socket.ntohl(struct.unpack("I", content[:4])[0]) return content[4:4 + msg_len] def _encrypt(self, plaintext: bytes) -> str: random_bytes = os.urandom(16) msg_len_bytes = struct.pack("!I", len(plaintext)) raw = random_bytes + msg_len_bytes + plaintext + self.corp_id.encode("utf-8") return base64.b64encode( _AES256CBC.encrypt_cbc(raw, self.aes_key, self.aes_key[:16])).decode("utf-8") def verify_url(self, signature: str, timestamp: str, nonce: str, echostr: str) -> Tuple[int, str]: if not self._verify_signature(signature, timestamp, nonce, echostr): return -1, "签名验证失败" try: return 0, self._decrypt(echostr).decode("utf-8") except Exception as e: return -2, f"解密失败: {e}" def decrypt_msg(self, signature: str, timestamp: str, nonce: str, body_json: dict) -> Tuple[int, dict]: encrypt_b64 = body_json.get("encrypt", "") if not self._verify_signature(signature, timestamp, nonce, encrypt_b64): return -1, {} try: return 0, json.loads(self._decrypt(encrypt_b64).decode("utf-8")) except Exception as e: logger.error(f"消息解密失败: {e}") return -2, {} def encrypt_msg(self, reply: dict, nonce: str, timestamp: str = None) -> str: if timestamp is None: timestamp = str(int(time.time())) encrypt_b64 = self._encrypt( json.dumps(reply, ensure_ascii=False).encode("utf-8")) signature = self._make_signature(timestamp, nonce, encrypt_b64) return json.dumps({ "encrypt": encrypt_b64, "msgsignature": signature, "timestamp": timestamp, "nonce": nonce, }, ensure_ascii=False) # ============================================================================ # HTTP 客户端（urllib，零依赖） # ============================================================================ def http_post_agentarts(url: str, payload: dict, headers: dict = None, timeout: int = 30) -> dict: """POST 到 AgentArts API，自动处理 SSE 流式响应""" headers = headers or {} headers.setdefault("Content-Type", "application/json") body = json.dumps(payload).encode("utf-8") ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE req = urllib.request.Request(url, data=body, headers=headers, method="POST") try: with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp: raw = resp.read() text = raw.decode("utf-8") logger.info(f"[HTTP] status={resp.status}, body_len={len(raw)}") # 尝试直接解析 JSON try: return json.loads(text) except json.JSONDecodeError: pass # 解析 SSE 格式 (data:{...}\n\ndata:{...}\n\n...) chunks = [] for line in text.split("\n"): line = line.strip() if line.startswith("data:"): chunk_json = line[5:].strip() try: chunk = json.loads(chunk_json) chunks.append(chunk) logger.info(f"[SSE] event={chunk.get('event', '?')}, " f"data_keys={list(chunk.get('data', {}).keys()) if isinstance(chunk.get('data'), dict) else '?'}") except json.JSONDecodeError: logger.warning(f"[SSE] 跳过无法解析的行: {chunk_json[:80]}") if not chunks: raise RuntimeError(f"无法解析响应 (status={resp.status}): {text[:300]}") # 从 SSE chunks 中提取结果 # AgentArts 的 SSE 格式: {"event": "message", "answer": "..."} # 或: {"event": "error", "data": {"message": "..."}} last_chunk = chunks[-1] # 检查错误事件 for chunk in chunks: event_type = chunk.get("event", "") if event_type == "error": err_data = chunk.get("data", chunk) error_msg = err_data.get("message", str(err_data)) raise RuntimeError(f"AgentArts 错误: {error_msg}") # 提取回复内容 # 尝试多种字段名: answer, response, output, text, content answer = ( last_chunk.get("answer") or (last_chunk.get("data", {}) or {}).get("answer") or last_chunk.get("response") or last_chunk.get("output") or last_chunk.get("text") ) if answer: return {"response": answer, "status": "success"} # 如果 last chunk 没有直接答案，遍历所有 chunks 拼接 full_answer = "" for chunk in chunks: event_type = chunk.get("event", "") if event_type in ("done", "error", "ping"): continue # 尝试多种字段名 text = (chunk.get("answer") or chunk.get("delta") or chunk.get("content") or chunk.get("text") or chunk.get("response") or (chunk.get("data", {}) or {}).get("answer") or (chunk.get("data", {}) or {}).get("delta") or (chunk.get("data", {}) or {}).get("content") or (chunk.get("data", {}) or {}).get("text") or "") if text: full_answer += text if full_answer: return {"response": full_answer, "status": "success"} # 兜底：返回所有非 done 的 chunks（调试用） content_chunks = [c for c in chunks if c.get("event") not in ("done", "ping")] return { "response": "未能提取回复，所有数据块: " + json.dumps(content_chunks, ensure_ascii=False), "status": "debug", } except urllib.error.HTTPError as e: body_text = e.read().decode("utf-8", errors="replace")[:500] logger.error(f"[HTTP] HTTPError {e.code}: {body_text}") raise RuntimeError(f"HTTP {e.code}: {body_text}") except Exception as e: if "请求失败" in str(e) or "AgentArts" in str(e) or "无法解析" in str(e): raise raise RuntimeError(f"请求失败: {e}") # ============================================================================ # AgentArts 客户端 # ============================================================================ class AgentArtsClient: """AgentArts API 封装（urllib 实现）""" def __init__(self, url: str, api_key: str): self.url = url self.api_key = api_key def chat(self, user_input: str, user_id: str = "", timeout: int = 25) -> str: headers = {"Authorization": f"Bearer {self.api_key}"} if user_id: headers["X-Hw-Agentarts-Session-Id"] = user_id headers["X-Hw-Agentgateway-User-Id"] = user_id try: data = http_post_agentarts(self.url, {"input": user_input}, headers, timeout) return data.get("response") or data.get("output") or str(data) except Exception as e: logger.error(f"AgentArts 调用失败: {e}") return "⚠️ 智能体服务暂不可用，请稍后重试。" # ============================================================================ # 消息处理 # ============================================================================ wx_crypt = None agent_client = None _seen_ids: set = set() MENTION_PATTERN = re.compile(r"@\S+\s*") def _init(): global wx_crypt, agent_client if wx_crypt is None and all([WECOM_TOKEN, WECOM_ENCODING_AES_KEY, WECOM_CORP_ID]): wx_crypt = WXBizMsgCrypt(WECOM_TOKEN, WECOM_ENCODING_AES_KEY, WECOM_CORP_ID) if agent_client is None: agent_client = AgentArtsClient(AGENTARTS_URL, AGENTARTS_API_KEY) # ============================================================================ # CloudBase HTTP 触发器入口 # ============================================================================ def main(event, context): """ CloudBase HTTP 云函数入口。 处理企业微信回调的两类请求： GET /callback?msg_signature=...&echostr=... → URL 验证 POST /callback (加密 JSON) → 消息处理 """ _init() path = event.get("path", "/") method = event.get("httpMethod", "GET") query = event.get("queryStringParameters", {}) or {} logger.info(f"{method} {path}") if method == "GET" and path in ("/", ""): return _resp(200, "OK") # --- 测试接口：直接调 AgentArts（不经过企业微信加密）--- if method == "POST" and path.endswith("/test"): return _handle_test(event) if method == "GET" and path.endswith("/callback"): return _handle_verify(query) if method == "POST" and path.endswith("/callback"): body = event.get("body", "") or "" if event.get("isBase64Encoded", False): import base64 as b64 body = b64.b64decode(body).decode("utf-8") return _handle_message(query, body) return _resp(404, "Not Found") # ============================================================================ # 辅助函数 # ============================================================================ def _resp(status: int, body: str, content_type: str = "text/plain") -> dict: return { "statusCode": status, "headers": {"Content-Type": f"{content_type}; charset=utf-8"}, "body": body, } def _handle_test(event: dict) -> dict: """ 测试接口：直接调用 AgentArts，不经过企业微信加密。 """ # 先看看 CloudBase 到底传了什么（调试用） debug_info = { "path": event.get("path", ""), "httpMethod": event.get("httpMethod", ""), "hasBody": "body" in event, "bodyType": str(type(event.get("body", ""))), "bodyLen": len(event.get("body") or ""), "bodyPreview": str((event.get("body") or "")[:200]), "isBase64": event.get("isBase64Encoded", False), "keys": list(event.keys()), "headerKeys": list((event.get("headers", {}) or {}).keys()), } logger.info(f"[DEBUG] event: {json.dumps(debug_info, ensure_ascii=False)}") body_raw = event.get("body", "") or "" # 尝试解析 try: req = json.loads(body_raw) except json.JSONDecodeError: # 返回调试信息帮助定位 return _resp(400, json.dumps({"error": "invalid JSON", "debug": debug_info}, ensure_ascii=False, indent=2), "application/json") user_input = req.get("input", "") if not user_input: return _resp(400, json.dumps({"error": "缺少 input 字段", "received": req}, ensure_ascii=False), "application/json") logger.info(f"[TEST] 调用 AgentArts: {user_input[:60]}") try: headers = { "Authorization": f"Bearer {AGENTARTS_API_KEY}", "X-Hw-Agentarts-Session-Id": "cloudbase-test-session", } # 同时发送 input 和 query，兼容不同版本的 AgentArts API result = http_post_agentarts(AGENTARTS_URL, {"input": user_input, "query": user_input}, headers, timeout=30) return _resp(200, json.dumps(result, ensure_ascii=False, indent=2), "application/json") except Exception as e: return _resp(500, json.dumps({"error": str(e)}, ensure_ascii=False), "application/json") def _handle_verify(query: dict) -> dict: signature = query.get("msg_signature", "") timestamp = query.get("timestamp", "") nonce = query.get("nonce", "") echostr = query.get("echostr", "") logger.info(f"URL 验证: ts={timestamp}") if wx_crypt is None: return _resp(500, "server not configured") ret, result = wx_crypt.verify_url(signature, timestamp, nonce, echostr) if ret == 0: logger.info("URL 验证成功") return _resp(200, result) logger.error(f"URL 验证失败: {result}") return _resp(403, f"verify failed: {result}") def _handle_message(query: dict, body_str: str) -> dict: signature = query.get("msg_signature", "") timestamp = query.get("timestamp", "") nonce = query.get("nonce", "") try: body_json = json.loads(body_str) except (json.JSONDecodeError, UnicodeDecodeError): return _resp(400, "bad request") if wx_crypt is None: return _resp(500, "server not configured") ret, msg = wx_crypt.decrypt_msg(signature, timestamp, nonce, body_json) if ret != 0: return _resp(403, "decrypt failed") from_id = (msg.get("from", {}) or {}).get("userid", "?") msg_type = msg.get("msgtype", "?") logger.info(f"消息: from={from_id}, type={msg_type}") msg_id = msg.get("msgid", "") if msg_id: if msg_id in _seen_ids: return _resp(200, "") _seen_ids.add(msg_id) if len(_seen_ids) > 5000: _seen_ids.clear() if msg_type != "text": return _resp(200, "") content = (msg.get("text", {}) or {}).get("content", "") if not content or not content.strip(): return _resp(200, "") content = MENTION_PATTERN.sub("", content, count=1).strip() if not content: return _resp(200, "") logger.info(f"用户 {from_id}: {content[:60]}") reply_text = agent_client.chat(content, user_id=from_id) logger.info(f"回复 {from_id}: {reply_text[:60]}...") reply_json = {"msgtype": "text", "text": {"content": reply_text}} encrypted = wx_crypt.encrypt_msg(reply_json, nonce, timestamp) return _resp(200, encrypted, "application/json") """ 企业微信智能机器人 → 云函数 → AgentArts 智能体 ============================================== ⚠️ 重要：企业微信群机器人不支持接收消息回调！ 正确做法是使用「自建应用」→「智能机器人」功能。 支持的云平台： • 腾讯云 SCF（云函数 Web 函数）—— 推荐，个人免费额度充裕 • 华为云 FunctionGraph（HTTP 函数） • 任意 VPS / Docker 部署 架构： 用户在群里 @机器人 / 私聊机器人 │ ▼ 企业微信 POST JSON（加密） → 云函数（Flask） │ ├─ 解密消息（AES-256-CBC + SHA1 签名验证） ├─ 调用 AgentArts API（同步等待） ├─ 加密回复（AES-256-CBC） └─ 返回加密 JSON → 企业微信展示给用户 部署：上传 ZIP 到云函数控制台 依赖：flask, requests, pycryptodome ┌─────────────────────────────────────────────────────────────┐ │ 环境变量从哪里获取（详见整合指南） │ ├────────────────────┬────────────────────────────────────────┤ │ AGENTARTS_URL │ AgentArts 控制台 → 智能体运行时 → │ │ │ 最新版本配置详情 → 访问域名 + 路径 │ │ AGENTARTS_API_KEY │ AgentArts 控制台 → 智能体运行时 → │ │ │ 访问与权限控制 → API Key │ │ WECOM_CORP_ID │ 企业微信管理后台 → 我的企业 → 企业ID │ │ WECOM_TOKEN │ 自定义字符串，在自建应用接收消息页填写 │ │ WECOM_ENCODING_AES │ 自建应用接收消息页 → 点击「随机获取」 │ │ WECOM_AGENT_SECRET │ 自建应用详情页 → Secret（用于主动发消息） │ │ WECOM_AGENT_ID │ 自建应用详情页 → AgentId（用于主动发消息） │ └────────────────────┴────────────────────────────────────────┘ """ import json import os import re import time import struct import base64 import hashlib import socket import logging import threading from typing import Tuple, Optional import requests from Crypto.Cipher import AES from flask import Flask, request, Response # ============================================================================ # 配置（全部从环境变量读取） # ============================================================================ # --- AgentArts --- # 来源：AgentArts 控制台 → 智能体运行时 → 最新版本配置详情 → 访问域名 # 格式：https://{域名}/runtimes/{runtime_name}/invocations?endpoint=Latest AGENTARTS_URL = os.getenv("AGENTARTS_URL", "https://defaultgw-mvstsmzsgv.cn-southwest-2.huaweicloud-agentarts.com" "/runtimes/runtime-17uneunb/invocations?endpoint=Latest") # 来源：AgentArts 控制台 → 智能体运行时 → 访问与权限控制 → API Key AGENTARTS_API_KEY = os.getenv("AGENTARTS_API_KEY", "3de5d906dfc24af98431bc4954f4b589") # --- 企业微信自建应用 --- # 来源：企业微信管理后台 → 我的企业 → 企业ID WECOM_CORP_ID = os.getenv("WECOM_CORP_ID", "") # 来源：自建应用详情页 → Secret（点击查看） WECOM_AGENT_SECRET = os.getenv("WECOM_AGENT_SECRET", "") # 来源：自建应用详情页 → AgentId WECOM_AGENT_ID = os.getenv("WECOM_AGENT_ID", "") # 来源：自定义（字母+数字），在自建应用「接收消息」页填写，两边一致即可 WECOM_TOKEN = os.getenv("WECOM_TOKEN", "") # 来源：自建应用「接收消息」页 → 点击「随机获取」自动生成（43字符） WECOM_ENCODING_AES_KEY = os.getenv("WECOM_ENCODING_AES_KEY", "") # --- 服务配置 --- LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO") logging.basicConfig( level=getattr(logging, LOG_LEVEL.upper(), logging.INFO), format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) logger = logging.getLogger("wecom-bot-fg") # ============================================================================ # 企业微信消息加解密（WXBizMsgCrypt） # 适用于自建应用 / 智能机器人，2025 年 JSON 格式 # ============================================================================ class WXBizMsgCrypt: """ 企业微信消息加解密。 加密算法：AES-256-CBC + PKCS#7 填充 签名算法：SHA1(token + timestamp + nonce + encrypt_msg) 排序后拼接 消息格式：JSON（2025 年版智能机器人格式） 参考：https://developer.work.weixin.qq.com/document/path/101033 """ def __init__(self, token: str, encoding_aes_key: str, corp_id: str): self.token = token self.corp_id = corp_id # EncodingAESKey: 43 字符 Base64 → 32 字节 AES 密钥 self.aes_key = base64.b64decode(encoding_aes_key + "=") if len(self.aes_key) != 32: raise ValueError(f"AES 密钥长度错误: {len(self.aes_key)} (应为 32)") # ---------- SHA1 签名 ---------- @staticmethod def _sha1(text: str) -> str: return hashlib.sha1(text.encode("utf-8")).hexdigest() def _make_signature(self, timestamp: str, nonce: str, encrypt_msg: str) -> str: params = sorted([self.token, timestamp, nonce, encrypt_msg]) return self._sha1("".join(params)) def _verify_signature(self, signature: str, timestamp: str, nonce: str, encrypt_msg: str) -> bool: return self._make_signature(timestamp, nonce, encrypt_msg) == signature # ---------- AES 加解密 ---------- def _pkcs7_pad(self, data: bytes, block_size: int = 32) -> bytes: pad_len = block_size - (len(data) % block_size) return data + bytes([pad_len] * pad_len) @staticmethod def _pkcs7_unpad(data: bytes) -> bytes: pad_len = data[-1] if pad_len < 1 or pad_len > 32: raise ValueError(f"无效填充长度: {pad_len}") return data[:-pad_len] def _aes_encrypt(self, plaintext: bytes) -> bytes: iv = self.aes_key[:16] cipher = AES.new(self.aes_key, AES.MODE_CBC, iv) return cipher.encrypt(self._pkcs7_pad(plaintext)) def _aes_decrypt(self, ciphertext: bytes) -> bytes: iv = self.aes_key[:16] cipher = AES.new(self.aes_key, AES.MODE_CBC, iv) return self._pkcs7_unpad(cipher.decrypt(ciphertext)) # ---------- 消息加解密 ---------- def _decrypt(self, encrypt_b64: str) -> bytes: """解密 Base64 编码的密文 → 返回明文 JSON 字节""" raw = self._aes_decrypt(base64.b64decode(encrypt_b64)) # 解密后格式: random(16) + msg_len(4, network order) + msg + receive_id content = raw[16:] msg_len = socket.ntohl(struct.unpack("I", content[:4])[0]) return content[4:4 + msg_len] def _encrypt(self, plaintext: bytes, corp_id: str) -> str: """加密明文 → 返回 Base64 编码的密文字符串""" random_bytes = os.urandom(16) msg_len_bytes = struct.pack("!I", len(plaintext)) raw = random_bytes + msg_len_bytes + plaintext + corp_id.encode("utf-8") return base64.b64encode(self._aes_encrypt(raw)).decode("utf-8") # ---------- 公开接口 ---------- def verify_url(self, signature: str, timestamp: str, nonce: str, echostr: str) -> Tuple[int, str]: """ URL 验证（GET 请求）。 企业微信配置回调 URL 时调用，需要解密 echostr 并返回明文。 """ if not self._verify_signature(signature, timestamp, nonce, echostr): return -1, "签名验证失败" try: plaintext = self._decrypt(echostr).decode("utf-8") return 0, plaintext except Exception as e: return -2, f"解密失败: {e}" def decrypt_msg(self, signature: str, timestamp: str, nonce: str, body_json: dict) -> Tuple[int, dict]: """ 解密消息（POST 请求，JSON 格式）。 输入 body_json 格式（JSON 信封）: {"encrypt": "...", "msgsignature": "...", "timestamp": "...", "nonce": "..."} 返回解密后的消息 JSON dict。 """ encrypt_b64 = body_json.get("encrypt", "") if not self._verify_signature(signature, timestamp, nonce, encrypt_b64): return -1, {} try: plaintext = self._decrypt(encrypt_b64).decode("utf-8") return 0, json.loads(plaintext) except Exception as e: logger.error(f"消息解密失败: {e}") return -2, {} def encrypt_msg(self, reply: dict, nonce: str, timestamp: str = None) -> str: """ 加密回复消息，返回 JSON 信封字符串。 输入 reply: {"msgtype": "text", "text": {"content": "回复内容"}} 输出: {"encrypt": "...", "msgsignature": "...", "timestamp": "...", "nonce": "..."} """ if timestamp is None: timestamp = str(int(time.time())) plaintext = json.dumps(reply, ensure_ascii=False).encode("utf-8") encrypt_b64 = self._encrypt(plaintext, self.corp_id) signature = self._make_signature(timestamp, nonce, encrypt_b64) return json.dumps({ "encrypt": encrypt_b64, "msgsignature": signature, "timestamp": timestamp, "nonce": nonce, }, ensure_ascii=False) # ============================================================================ # AgentArts 客户端 # ============================================================================ class AgentArtsClient: """AgentArts API 轻量封装""" def __init__(self, url: str, api_key: str): self.url = url self.api_key = api_key self._session: Optional[requests.Session] = None @property def session(self) -> requests.Session: if self._session is None: self._session = requests.Session() self._session.headers.update({ "Content-Type": "application/json", "Authorization": f"Bearer {self.api_key}", }) return self._session def chat(self, user_input: str, user_id: str = "", timeout: int = 25) -> str: """调用 AgentArts，返回文本回复""" headers = {} if user_id: headers["X-Hw-Agentarts-Session-Id"] = user_id headers["X-Hw-Agentgateway-User-Id"] = user_id try: resp = self.session.post( self.url, json={"input": user_input}, headers=headers, timeout=timeout, verify=False, ) resp.raise_for_status() data = resp.json() return data.get("response") or data.get("output") or str(data) except requests.Timeout: logger.error(f"AgentArts 超时: {user_input[:50]}...") return "⏳ 智能体响应超时，请稍后重试。" except requests.RequestException as e: logger.error(f"AgentArts 调用失败: {e}") return "⚠️ 智能体服务暂不可用，请稍后重试。" # ============================================================================ # 企业微信 API 客户端（用于异步推送回复） # ============================================================================ class WeComAPI: """企业微信 API：获取 access_token + 主动发送消息""" _token: Optional[str] = None _expires: float = 0 _lock = threading.Lock() def __init__(self, corp_id: str, secret: str, agent_id: str): self.corp_id = corp_id self.secret = secret self.agent_id = agent_id def _get_token(self) -> str: now = time.time() if self._token and now < (self._expires - 300): return self._token with self._lock: if self._token and now < (self._expires - 300): return self._token resp = requests.get( "https://qyapi.weixin.qq.com/cgi-bin/gettoken", params={"corpid": self.corp_id, "corpsecret": self.secret}, timeout=10, ) data = resp.json() if data.get("errcode") == 0: self._token = data["access_token"] self._expires = now + data.get("expires_in", 7200) logger.info("access_token 已刷新") return self._token raise RuntimeError(f"获取 access_token 失败: {data}") def send_text(self, user_id: str, content: str) -> bool: """主动发送文本消息给指定用户""" token = self._get_token() resp = requests.post( "https://qyapi.weixin.qq.com/cgi-bin/message/send", params={"access_token": token}, json={ "touser": user_id, "msgtype": "text", "agentid": int(self.agent_id), "text": {"content": content[:2048]}, }, timeout=10, ) data = resp.json() if data.get("errcode") != 0: logger.error(f"发送消息失败: {data}") return False return True # ============================================================================ # 核心处理逻辑 # ============================================================================ # 初始化组件（如果变量都配了） wx_crypt = None agent_client = None wecom_api = None def _init(): global wx_crypt, agent_client, wecom_api if wx_crypt is None and all([WECOM_TOKEN, WECOM_ENCODING_AES_KEY, WECOM_CORP_ID]): wx_crypt = WXBizMsgCrypt(WECOM_TOKEN, WECOM_ENCODING_AES_KEY, WECOM_CORP_ID) if agent_client is None: agent_client = AgentArtsClient(AGENTARTS_URL, AGENTARTS_API_KEY) if wecom_api is None and all([WECOM_CORP_ID, WECOM_AGENT_SECRET, WECOM_AGENT_ID]): wecom_api = WeComAPI(WECOM_CORP_ID, WECOM_AGENT_SECRET, WECOM_AGENT_ID) # 去重 _seen_ids: set = set() _seen_lock = threading.Lock() # @机器人 前缀匹配模式 MENTION_PATTERN = re.compile(r"@\S+\s*") def process_and_reply(msg: dict) -> str: """ 处理消息，返回回复文本。 消息格式（解密后的 JSON）: { "msgid": "...", "msgtype": "text", "from": {"userid": "zhangsan"}, "text": {"content": "@AI助手 正方形蓝苹果是什么？"}, "chatid": "...", "chattype": "group" } """ msg_id = msg.get("msgid", "") msg_type = msg.get("msgtype", "") # --- 去重 --- if msg_id: with _seen_lock: if msg_id in _seen_ids: return "" _seen_ids.add(msg_id) if len(_seen_ids) > 5000: _seen_ids.clear() # --- 只处理文本 --- if msg_type != "text": logger.info(f"忽略非文本消息: {msg_type}") return "" content = (msg.get("text", {}) or {}).get("content", "") if not content or not content.strip(): return "" # 去掉 @机器人 前缀 content = MENTION_PATTERN.sub("", content, count=1).strip() if not content: return "你好！请直接输入你想问的问题，我会尽力回答 👋" from_info = msg.get("from", {}) user_id = from_info.get("userid", "unknown") logger.info(f"用户 {user_id}: {content[:60]}") # --- 调用 AgentArts --- reply = agent_client.chat(content, user_id=user_id) logger.info(f"回复 {user_id}: {reply[:60]}...") return reply # ============================================================================ # Flask 应用（FunctionGraph HTTP 函数入口） # ============================================================================ app = Flask(__name__) @app.route("/", methods=["GET", "HEAD"]) def health(): """健康检查""" return Response("OK", status=200) @app.route("/callback", methods=["GET"]) def verify_url(): """ ┌─────────────────────────────────────────┐ │ 回调 URL 验证（GET） │ │ 企业微信在配置回调 URL 时首先发此请求验证 │ │ │ │ 参数（query string）： │ │ msg_signature - 签名 │ │ timestamp - 时间戳 │ │ nonce - 随机数 │ │ echostr - 加密的验证字符串 │ │ │ │ 响应：解密后的 echostr 明文 │ └─────────────────────────────────────────┘ """ _init() signature = request.args.get("msg_signature", "") timestamp = request.args.get("timestamp", "") nonce = request.args.get("nonce", "") echostr = request.args.get("echostr", "") logger.info(f"URL 验证请求: ts={timestamp}, nonce={nonce}") if wx_crypt is None: logger.error("企业微信配置未设置（WECOM_TOKEN 等环境变量缺失）") return Response("server not configured", status=500) ret, result = wx_crypt.verify_url(signature, timestamp, nonce, echostr) if ret == 0: logger.info("URL 验证成功") return Response(result, content_type="text/plain; charset=utf-8") else: logger.error(f"URL 验证失败: {result}") return Response(f"verify failed: {result}", status=403) @app.route("/callback", methods=["POST"]) def handle_message(): """ ┌─────────────────────────────────────────┐ │ 消息回调（POST） │ │ 用户 @机器人 或私聊时，企业微信 POST 到此 │ │ │ │ 请求体（JSON 信封）： │ │ {"encrypt": "...", │ │ "msgsignature": "...", │ │ "timestamp": "...", │ │ "nonce": "..."} │ │ │ │ 响应：加密的 JSON 信封（同步回复） │ └─────────────────────────────────────────┘ """ _init() signature = request.args.get("msg_signature", "") timestamp = request.args.get("timestamp", "") nonce = request.args.get("nonce", "") try: body = json.loads(request.data.decode("utf-8")) except (json.JSONDecodeError, UnicodeDecodeError) as e: logger.error(f"请求解析失败: {e}") return Response("bad request", status=400) if wx_crypt is None: logger.error("企业微信配置未设置") return Response("server not configured", status=500) # --- 解密消息 --- ret, msg = wx_crypt.decrypt_msg(signature, timestamp, nonce, body) if ret != 0: logger.error("消息解密失败") return Response("decrypt failed", status=403) logger.info(f"收到消息: from={msg.get('from', {}).get('userid', '?')}, " f"type={msg.get('msgtype', '?')}") # --- 处理消息 --- reply_text = process_and_reply(msg) if not reply_text: # 无需回复（去重、非文本消息等） return Response("", status=200) # --- 加密并返回回复 --- reply_json = { "msgtype": "text", "text": {"content": reply_text}, } encrypted = wx_crypt.encrypt_msg(reply_json, nonce, timestamp) return Response(encrypted, content_type="application/json; charset=utf-8") # ============================================================================ # 本地调试（FunctionGraph 部署时不会执行此段） # ============================================================================ if __name__ == "__main__": _init() # 腾讯云 SCF Web 函数默认端口 9000，其他环境默认 8080 port = int(os.getenv("TENCENTCLOUD_SERVERPORT", "8080")) logger.info("=" * 55) logger.info("企业微信智能机器人 → AgentArts 转发服务") logger.info(f"AgentArts: {AGENTARTS_URL[:50]}...") logger.info(f"监听端口: {port}") logger.info("=" * 55) app.run(host="0.0.0.0", port=port, debug=False)