最近做的企微二开，要求机器人不只是聊天——客户说"帮我查下张三的订单然后通知销售"，机器人要能拆解成多个任务、分别调接口执行、把结果汇总回客户。和之前聊的大模型工具接入不同，那篇重点是工具注册和调用循环，这篇重点是任务拆解→多接口协同执行→结果汇总回写的完整闭环。把调了哪些接口记下来。
Eyun 平台开放的企微 API，统一 POST+JSON，鉴权用 App Token 加 appid（Authorization: Bearer eyk_xxxx），路径{BASE_URL}/wx-api/api/<模块>/<动作>，响应封套{code, data, detail, message, time}，code 为 0 成功。
任务拆解：AI 拆成多个步骤
客户说一句复杂的话，AI 先拆解成多个执行步骤：
def decompose_task(user_msg, context): prompt = f""" 用户请求：{user_msg} 可用工具：query_customer, query_order, send_message, update_label, create_ticket 请拆解成具体执行步骤，每步调一个工具。 """ steps = llm.plan(prompt) return stepsAI 返回步骤列表，每步指定调哪个工具、传什么参数。比如"查张三订单通知销售"拆成：
query_customer("张三") 拿到客户 uin
query_order(uin) 拿到订单
send_message(销售 uin, 订单信息) 通知销售
步骤执行：调企微接口
每个步骤对应一次企微接口调用。步骤间有依赖——后一步要用前一步的结果：
import requests BASE = "https://api.eyun.com" HEADERS = {"Authorization": "Bearer eyk_xxxx", "Content-Type": "application/json"} def execute_steps(appid, steps, context): results = {} for step in steps: tool = step["tool"] args = fill_args(step["args"], results) # 用前一步结果填参数 if tool == "query_customer": results[step["id"]] = query_customer(appid, args["keyword"]) elif tool == "query_order": results[step["id"]] = query_order(appid, args["uin"]) elif tool == "send_message": results[step["id"]] = send_message(appid, args["to"], args["content"]) elif tool == "update_label": results[step["id"]] = update_label(appid, args["uin"], args["labels"]) return results def fill_args(arg_template, results): # 把 $step1.uin 这种引用替换成实际值 for key, val in arg_template.items(): if isinstance(val, str) and val.startswith("$"): parts = val[1:].split(".") step_id = parts[0] field = parts[1] arg_template[key] = results[step_id][field] return arg_templatefill_args处理步骤间依赖——第二步的参数引用第一步的结果。
查客户：调联系人接口
def query_customer(appid, keyword): if keyword.isdigit() and len(keyword) >= 11: resp = requests.post( f"{BASE}/wx-api/api/contact/phoneNumberSearch", headers=HEADERS, json={"appid": appid, "phone": keyword} ) else: resp = requests.post( f"{BASE}/wx-api/api/contact/search", headers=HEADERS, json={"appid": appid, "keyword": keyword} ) data = resp.json()["data"] return {"uin": data.get("uin"), "name": data.get("nickName", "")}通知销售：调消息接口
def send_message(appid, to_uin, content): resp = requests.post( f"{BASE}/wx-api/api/message/sendText", headers=HEADERS, json={"appid": appid, "to": to_uin, "content": content} ) return resp.json()["code"] == 0打标签：调标签接口
def update_label(appid, uin, labels): resp = requests.post( f"{BASE}/wx-api/api/label/updateLabel", headers=HEADERS, json={"appid": appid, "uin": uin, "labels": labels} ) return resp.json()["code"] == 0结果汇总：AI 组织回复
所有步骤执行完，AI 把结果汇总成自然语言回客户：
def summarize_and_reply(appid, from_uin, user_msg, results): summary = llm.summarize(user_msg, results) requests.post( f"{BASE}/wx-api/api/message/sendText", headers=HEADERS, json={"appid": appid, "to": from_uin, "content": summary} )AI 拿到所有步骤的结果，生成"已查到张三的订单 SO-123，金额 300 元，已通知销售王五"这种汇总回复。不汇总，客户只看到最后一步的结果，不知道前面查了什么。
完整闭环
def handle_complex_task(appid, from_uin, content): # 1. 拆解任务 steps = decompose_task(content, {}) # 2. 执行步骤 results = execute_steps(appid, steps, {}) # 3. 汇总回复 summarize_and_reply(appid, from_uin, content, results)客户一句话进来，拆解、执行、汇总、回复，整个闭环调多个企微接口。关于消息和联系人接口，在Eyun 企业微信 API 平台有说明。
执行失败的兜底
某步失败要兜底——不能整条链崩：
def execute_steps_safe(appid, steps, context): results = {} for step in steps: try: results[step["id"]] = execute_step(appid, step, results) except Exception as e: results[step["id"]] = {"error": str(e)} if step.get("required", True): # 关键步骤失败，终止并回复 break return results关键步骤失败终止并回复客户"查询失败"，非关键步骤失败继续执行后续。不做兜底，一步失败全链崩，客户等半天没回复。
写在最后
会执行任务的智能机器人这套东西，本质是把企微多个接口用任务拆解串起来——AI 拆解任务、按步骤调 phoneNumberSearch/sendText/updateLabel 等接口、步骤间传结果、汇总回复。把任务拆解做对、步骤依赖处理好、失败兜底做实，机器人真正能执行复杂任务——而不是只会一问一答。
；》展开的关于人工“自我”的随想)
