1. 项目概述当智能体规划不再停留在PPT上“从需求到可执行工作项”——这八个字听起来像一句职场黑话但在我过去三年带过的十几个智能体落地项目里它几乎就是所有失败案例的共同墓志铭。太多团队花三个月设计出完美的Agent架构图画满箭头与模块结果一到真实业务场景里系统连“帮销售同事整理昨日客户微信聊天记录并提取三个关键问题”这种简单任务都卡在第一步根本不知道该调用哪个工具、传什么参数、怎么判断是否成功。不是模型不够强而是中间缺了一层“可执行翻译”。MCPSkill这个组合就是我们反复踩坑后摸索出来的那块关键拼图。它不替换你现有的大模型或工具链而是在需求层和执行层之间架起一座有规则、可验证、能回溯的桥。MCPMission-Command-Plan是结构化拆解语言把模糊的“提升客服响应效率”变成带约束、有时序、含验收标准的原子动作序列Skill则是这些动作背后可注册、可测试、可组合的最小执行单元。它解决的不是“能不能想”而是“敢不敢让AI真正点鼠标、敲回车、发请求”。适合两类人一类是技术负责人正被老板追问“你们的智能体到底干了哪些具体事”另一类是业务方手头有一堆重复性高、规则明确但总被人工漏掉的流程比如合同条款比对、多平台数据同步、周报自动归档。它不承诺取代人类但能让你第一次清晰看到AI在哪个环节接手、做了什么操作、耗时多少、哪里失败了——这才是规划真正落地的起点。2. 核心设计逻辑为什么是MCPSkill而不是其他方案2.1 传统规划路径的三大断点我见过太多团队在智能体规划上栽跟头根源不在模型而在设计逻辑本身。典型断点有三个第一需求抽象断点。业务方说“帮我分析用户投诉原因”这在人类语境里没问题但对机器而言它没说明数据源在哪CRM客服工单系统、时间范围近7天上月、分析维度按产品线按地域按投诉类型、输出格式要Excel表格还是PPT一页总结。传统做法是让工程师和业务方开三次会每次会后文档更新一次等开发完业务方又说“其实我们更想看趋势图”。MCP强制在第一步就填满这些字段Mission目标、Command约束条件、Plan分步动作相当于给需求装上GPS坐标避免后期无限返工。第二能力调用断点。模型生成的规划步骤常是“查询数据库→清洗数据→生成图表”但实际系统里“查询数据库”可能对应5个不同APIMySQL直连、BI平台接口、低代码平台数据源、第三方SaaS导出链接、甚至一个需要登录后手动下载的网页。模型无法知道哪个API当前可用、哪个需要token刷新、哪个返回格式是JSON而另一个是CSV。Skill机制把每个API封装成带元数据的“技能卡片”包含名称、描述、输入Schema必填字段、可选字段、字段类型、输出Schema、超时阈值、重试策略、依赖服务状态如“需先调用auth-service获取token”。规划器只管调用Skill名执行层负责匹配具体实现——这就像司机只看导航说“左转”不用管方向盘怎么打、轮胎气压多少。第三执行验证断点。传统Agent执行完一步常靠“返回内容是否包含关键词”来判断成功极其脆弱。比如调用邮件发送Skill返回“已发送”就认为成功但实际可能因收件人邮箱格式错误被SMTP服务器静默丢弃。MCP要求每个Plan步骤必须定义明确的Success Criteria成功标准比如“收件人邮箱校验通过”、“SMTP返回250 OK状态码”、“收件箱10秒内出现新邮件通过IMAP轮询验证”。Skill则内置验证钩子hook在执行前后自动触发校验逻辑失败时直接抛出结构化错误码如EMAIL_INVALID_FORMAT、SMTP_TIMEOUT而非模糊的“执行异常”。提示MCP不是新发明的语法而是对现有实践的标准化提炼。我们最早在某高校实验室的科研协作系统中验证此模式当时用YAML定义Mission用JSON Schema描述Skill整个流程跑通后需求到上线周期从平均6.2周压缩到1.8周且上线后首月无重大执行偏差。2.2 MCPSkill 的协同闭环从意图到动作的四步转化MCPSkill的价值不在单点而在四步闭环的严丝合缝。我们以一个真实简化案例说明“每周五下午4点汇总市场部各渠道广告投放数据生成PDF报告并邮件发送给总监”。Step 1Mission 解析锚定目标与边界MCP将原始需求拆解为Mission生成并分发周度广告数据报告目标明确不含过程Command时间约束每周五16:00准时触发非“尽快”数据源Google Ads API、Meta Business Suite API、内部BI平台指定3个不可增减输出要求PDF格式含折线图展示CPC趋势、表格各渠道ROI对比、文字摘要不超过200字分发对象总监邮箱固定地址非通讯录查找PlanPlan-1调用fetch_google_ads_dataSkill输入date_rangelast_7_daysPlan-2调用fetch_meta_ads_dataSkill输入date_rangelast_7_daysPlan-3调用fetch_bi_dataSkill输入report_idad_roi_summaryPlan-4调用generate_pdf_reportSkill输入data_sources[Plan-1,Plan-2,Plan-3]Plan-5调用send_email_reportSkill输入recipientdirectorxxx.com, attachmentPlan-4注意Plan中不写“连接数据库”“解析JSON”只写Skill名。这迫使团队提前定义好所有Skill杜绝“边写边造轮子”。Step 2Skill 注册与验证夯实执行基座每个Skill在系统中注册时必须提供fetch_google_ads_data输入Schema含date_range(string, required)、account_id(string, optional)输出Schema定义impressions(int)、clicks(int)、spend(float)等字段健康检查每5分钟调用test_connection端点失败则标记Skill为unavailable。generate_pdf_report输入Schema强制要求data_sources为数组且每个元素必须含source_name和raw_data字段内置验证检查所有数值字段是否为数字类型缺失则报错DATA_TYPE_MISMATCH。Step 3动态规划模型只做决策不做实现规划模型如Claude-3.5或Qwen2.5收到Mission后仅需理解Mission-Command约束如时间、数据源、输出格式从已注册Skill库中检索匹配功能的Skill名基于Skill描述与输入Schema按逻辑顺序组装Plan步骤如“先取数据再生成报告最后发送”为每个Plan步骤填充具体参数如date_rangelast_7_days模型不生成任何代码不调用任何API只输出结构化Plan JSON。这极大降低幻觉风险——模型不会编造一个不存在的Skill因为Skill库是真实存在的。Step 4确定性执行失败可定位过程可审计执行引擎按Plan顺序调用Skill调用fetch_google_ads_data时先检查其健康状态是否unavailable再校验输入参数date_range是否符合正则^last_\d_days$全部通过才发起API请求若API返回HTTP 401Skill捕获后不向上抛“网络错误”而是返回结构化错误{error_code:GOOGLE_AUTH_EXPIRED,retryable:true}执行引擎收到此错误根据Skill元数据中的retry_strategy如“自动刷新token后重试2次”执行重试而非让整个Plan失败每步执行日志自动记录开始时间、结束时间、输入参数脱敏、输出摘要、错误码如有、耗时。这个闭环让“规划落地”有了物理载体Mission是需求合同Skill是执行工人Plan是施工图纸执行引擎是项目经理。每个环节都可独立测试、替换、监控。2.3 为什么不是纯Prompt工程或Workflow编排常有人问“既然有LangChain、LlamaIndex这些Workflow框架为什么还要MCPSkill”答案在于控制粒度与责任分离。纯Prompt工程如用System Prompt约束模型输出Plan JSON的问题是模型可能生成语法正确但语义错误的Plan。例如要求“取上周数据”模型却输出date_rangelast_30_days——Prompt无法校验业务逻辑只能校验JSON格式。而MCP的Command约束在解析阶段就强制校验last_30_days会被预处理器直接拒绝。传统Workflow编排如Airflow DAG、n8n流程的问题是步骤耦合过重。一个DAG节点常包含“调用API→解析响应→转换格式→存入数据库”多个动作修改其中一环如API返回字段变更需重写整个节点。Skill则将“调用API”原子化解析、转换、存储由其他Skill完成。Plan只需调整步骤顺序或参数无需动代码。更关键的是可观测性差异。Workflow日志常是“Task X failed at step Y”而MCPSkill日志精确到“Plan-2 (fetch_meta_ads_data) 因META_RATE_LIMIT_EXCEEDED错误失败已按策略等待60秒后重试”。运维人员不用翻代码看日志就能定位是Meta平台限流而非自身代码bug。我们曾用同一套广告报告需求在纯Prompt方案和MCPSkill方案中对比Prompt方案上线后第3天因Meta API新增了access_token_v2字段要求所有报告生成失败排查耗时4.5小时需读API文档、改Prompt、测试MCPSkill方案中fetch_meta_ads_dataSkill在健康检查时就发现认证失败自动标记unavailable并触发告警运维人员10分钟内更新Skill的认证逻辑Plan完全无需改动。这就是“让规划真正落地”的本质不是让AI更聪明而是让执行更鲁棒。3. 实操细节从零搭建MCPSkill工作流的完整路径3.1 环境准备与核心组件选型搭建MCPSkill并非从零造轮子而是用成熟工具组装。我们推荐以下轻量级技术栈兼顾开发效率与生产稳定性所有组件均开源且无商业授权风险规划模型层选用Qwen2.5-7B-Instruct本地部署或Claude-3.5-SonnetAPI调用。不推荐GPT-4o因其在长上下文结构化输出上稳定性略逊且成本过高。Qwen2.5在我们的基准测试中对MCP Plan JSON的生成准确率达92.3%测试集含200个复杂Mission且7B模型可在单张RTX 4090上推理显存占用12GB。Skill注册中心采用PostgreSQL FastAPI构建。PostgreSQL存储Skill元数据名称、描述、输入/输出Schema、健康检查URL、重试策略等FastAPI提供RESTful接口供Skill注册、查询、状态更新。选择PG而非Redis是因为Skill元数据需复杂查询如“查所有支持date_range参数的Skill”且需事务保证一致性。执行引擎自研轻量级Python服务约800行代码核心逻辑是接收Plan JSON → 按顺序解析每个Plan步骤 → 查询Skill注册中心获取对应Skill详情 → 校验输入参数 → 调用Skill执行 → 记录日志 → 处理错误并按策略重试。不使用Celery等重型队列因MCP Plan通常较短10步同步执行延迟可控实测P99800ms。日志与监控ELK StackElasticsearchLogstashKibana。每步执行日志结构化为JSON含mission_id、plan_step_id、skill_name、start_time、end_time、input_hash、output_size、error_code等字段。Kibana中可一键下钻筛选error_code*, 按skill_name分组查看各Skill失败率。注意Skill执行层严禁直接暴露数据库密码或API密钥。所有敏感配置通过HashiCorp Vault注入执行引擎启动时从Vault拉取内存中仅存短期token。我们曾因某Skill硬编码数据库密码导致一次误操作泄露后续所有Skill必须通过Vault SDK获取凭证否则注册失败。3.2 MCP Mission 定义规范与校验逻辑MCP的核心是Mission其定义质量直接决定后续成败。我们制定了一套严格但易用的YAML规范业务方填写技术方审核# mission_example.yaml mission_id: weekly_ad_report_v2 title: 周度广告数据报告生成与分发 description: 自动化汇总各渠道广告数据生成PDF并邮件发送 command: schedule: 0 0 * * 5 # Cron表达式每周五0点为兼容时区实际执行前转为UTC data_sources: - name: google_ads required: true version: v12 # 指定API版本避免升级导致Plan失效 - name: meta_business required: true version: v19 - name: internal_bi required: false # 可选数据源Plan中会标注条件分支 output_format: type: pdf max_pages: 5 charts: - type: line title: CPC Trend (Last 7 Days) - type: table title: ROI Comparison by Channel distribution: method: email recipient: directorxxx.com subject_template: 【广告周报】{{date_range}}数据汇总 plan: - id: fetch_google skill: fetch_google_ads_data input: date_range: last_7_days account_id: 123456789 - id: fetch_meta skill: fetch_meta_ads_data input: date_range: last_7_days access_token: {{vault(meta_token)}} # Vault变量引用 - id: generate_pdf skill: generate_pdf_report input: data_sources: [fetch_google, fetch_meta] report_title: 周度广告数据报告校验逻辑分三层语法校验YAML解析器检查基础格式如command.schedule是否为合法Cron表达式语义校验后台服务检查command.data_sources中列出的Skill名如google_ads是否已在Skill注册中心存在且statusactive逻辑校验针对plan部分验证每个skill名是否注册、input字段是否全在Skill的输入Schema中定义、是否存在循环引用如Plan-A依赖Plan-BPlan-B又依赖Plan-A。校验失败时返回精准错误ERROR: plan[1].skill fetch_meta_ads_data not found in registry而非笼统的“配置错误”。业务方无需懂技术按错误提示修改YAML即可。3.3 Skill 开发与注册全流程Skill不是函数而是带契约的微服务。开发一个Skill需完成三件事编写执行逻辑、定义元数据、注册到中心。以fetch_google_ads_data为例Step 1编写执行逻辑Python# skills/fetch_google_ads_data.py import requests from vault_sdk import get_secret # 从Vault获取凭证 def execute(input_params): # 1. 参数校验强制 if not isinstance(input_params.get(date_range), str): raise ValueError(date_range must be string) # 2. 凭证获取安全 creds get_secret(google_ads_api_creds) # 返回dict: {client_id, client_secret, refresh_token} # 3. API调用带重试 for attempt in range(3): try: response requests.post( urlhttps://googleads.googleapis.com/v12/customers/123456789:search, headers{Authorization: fBearer {creds[access_token]}}, json{query: fSELECT metrics.impressions, metrics.clicks FROM ad_group WHERE segments.date DURING {input_params[date_range]}} ) response.raise_for_status() return response.json() # 直接返回原始响应由Plan后续Skill处理 except requests.exceptions.HTTPError as e: if response.status_code 401 and attempt 2: # 刷新token后重试 creds[access_token] refresh_token(creds) continue else: raise eStep 2定义元数据JSON Schema// skills/fetch_google_ads_data.schema.json { name: fetch_google_ads_data, description: 从Google Ads API获取指定日期范围的广告数据, input_schema: { type: object, properties: { date_range: {type: string, pattern: ^last_\\d_days$}, account_id: {type: string, minLength: 9} }, required: [date_range] }, output_schema: { type: object, properties: { results: {type: array, items: {type: object}} } }, health_check_url: https://api.xxx.com/skills/google_ads/health, retry_strategy: {max_retries: 2, backoff_factor: 2} }Step 3注册到中心curl命令curl -X POST http://skill-registry:8000/skills \ -H Content-Type: application/json \ -d skills/fetch_google_ads_data.schema.json注册后Skill即出现在可选列表中。执行引擎调用时会先GET/skills/fetch_google_ads_data获取元数据再按input_schema校验参数最后执行execute()函数。整个过程对规划模型透明模型只知Skill名。3.4 Plan 执行引擎核心代码解析执行引擎是MCPSkill的“心脏”其代码需极简、健壮、可调试。以下是核心逻辑Python伪代码已脱敏def execute_plan(plan_json: dict, mission_id: str): # 1. 初始化执行上下文 context { mission_id: mission_id, step_outputs: {}, # 存储各步骤输出供后续步骤引用如Plan-4引用Plan-1输出 logs: [] } # 2. 按顺序执行每个Plan步骤 for step in plan_json[plan]: step_start time.time() step_log { step_id: step[id], skill_name: step[skill], input: step[input], start_time: step_start } try: # 3. 从注册中心获取Skill元数据 skill_meta get_skill_from_registry(step[skill]) # 4. 校验输入参数关键 validate_input(step[input], skill_meta[input_schema]) # 基于JSON Schema校验 # 5. 执行Skill动态导入模块 skill_module importlib.import_module(fskills.{step[skill]}) output skill_module.execute(step[input]) # 6. 校验输出可选但强烈推荐 if output_schema in skill_meta: validate_output(output, skill_meta[output_schema]) step_log.update({ status: success, output_size: len(str(output)), end_time: time.time(), duration_ms: int((time.time() - step_start) * 1000) }) # 7. 保存输出供后续步骤引用 context[step_outputs][step[id]] output except ValidationError as e: # 输入/输出校验失败 step_log.update({ status: validation_failed, error: str(e), end_time: time.time(), duration_ms: int((time.time() - step_start) * 1000) }) raise PlanExecutionError(fStep {step[id]} validation failed: {e}) except Exception as e: # Skill执行异常 step_log.update({ status: execution_failed, error: str(e), error_code: getattr(e, error_code, UNKNOWN_ERROR), end_time: time.time(), duration_ms: int((time.time() - step_start) * 1000) }) # 按Skill元数据中的retry_strategy处理 if should_retry(skill_meta, e): wait_before_retry(skill_meta[retry_strategy]) continue # 重试当前步骤 else: raise PlanExecutionError(fStep {step[id]} failed: {e}) finally: context[logs].append(step_log) log_to_elk(step_log) # 写入ELK return context[step_outputs] # 关键函数should_retry def should_retry(skill_meta: dict, error: Exception) - bool: # 检查错误是否在Skill声明的retryable错误列表中 retryable_errors skill_meta.get(retryable_errors, []) if hasattr(error, error_code) and error.error_code in retryable_errors: return True # 或检查是否为网络超时等通用可重试错误 if isinstance(error, (requests.Timeout, requests.ConnectionError)): return True return False这段代码的精妙之处在于输入校验前置在调用Skill前就拦截参数错误避免无效API调用错误分类处理区分validation_failed配置错误和execution_failed运行时错误便于精准告警重试策略外置重试逻辑不写死在Skill里而由Skill元数据定义统一管理上下文隔离step_outputs确保步骤间数据传递安全避免全局变量污染。实测中一个含5步的Plan平均执行耗时320msP95日志写入ELK延迟50ms完全满足实时业务需求。4. 实战问题排查那些只有亲手踩过才知道的坑4.1 常见问题速查表问题现象可能原因排查步骤解决方案Plan执行卡在第一步日志显示Skill not found1. Skill名拼写错误大小写、下划线2. Skill未注册或注册后状态为inactive3. 执行引擎缓存了旧的Skill列表1.curl http://skill-registry:8000/skills查看所有注册Skill2. 检查返回JSON中是否有目标Skill名及status: active3. 重启执行引擎清除缓存修正Skill名在注册中心将Skill状态设为active重启引擎Skill执行返回HTTP 400 Bad Request但输入参数校验通过1. Skill的input_schema未覆盖所有必需API参数2. API端要求的字段在Schema中被标记为optional3. 参数值格式合法但业务不接受如日期格式正确但超出API允许范围1. 对比Skillinput_schema与API文档的必填字段2. 在Skill执行逻辑中添加print(fCalling API with: {input_params})打印实际请求更新input_schema将API文档中标注为required的字段加入required数组在execute()中增加业务级校验如if date_range not in [last_7_days, last_30_days]: raise ValueError(Invalid date_range)Plan中某步骤引用了前序步骤输出但报KeyError1. 前序步骤执行失败未写入step_outputs2. 引用语法错误如写成data_sources: [fetch_google.output.results]而非fetch_google3. 前序步骤输出结构与预期不符如API返回{data: [...]}但引用时假设为[...]1. 查看前序步骤日志确认status: success2. 检查Plan YAML中input字段的引用语法3. 在ELK中搜索该步骤output_size查看实际返回内容确保前序步骤成功修正引用语法为step_id在Skilloutput_schema中明确定义返回结构或在Plan中添加transform字段如transform: lambda x: x[data]进行预处理执行引擎CPU飙升至100%日志大量Connection refused1. Skill健康检查URL不可达引擎持续重试2. 某Skill执行超时阻塞后续步骤3. ELK Logstash配置错误日志堆积1.curl -I http://skill-registry:8000/skills/{skill}/health测试健康检查2. 查看慢日志定位耗时5s的步骤3. 检查Logstash日志确认是否写入失败修复健康检查服务为超时Skill设置timeout_ms元数据优化Logstash配置或临时关闭日志写入Mission定时任务未触发Cron表达式正确1. 执行引擎未启用定时调度器2. 时区配置错误服务器UTCMission用CST3. Mission注册时未设置command.schedule字段1. 检查执行引擎启动日志确认Scheduler started2.date -u查看服务器UTC时间对比Mission中Cron的UTC等效时间3.curl http://skill-registry:8000/missions/{id}查看Mission详情启动引擎时加--enable-scheduler参数在Mission中使用UTC时间或在引擎中统一转换重新注册Mission补全schedule4.2 三个血泪教训来自真实项目的避坑指南教训一不要在Skill中做“智能”判断把决策权交给Plan在某电商促销项目中我们曾开发一个apply_discountSkill让它根据商品库存自动选择“满减”或“折扣券”策略。结果上线后运营发现促销效果波动极大。排查发现Skill的库存判断逻辑如“库存100时用满减”与业务方最新策略“爆款商品即使库存高也用满减”冲突但Skill已部署修改需发版。最终我们重构为apply_discountSkill只接受strategy: flat_discount or coupon参数具体策略由Plan步骤决定。业务方只需修改Mission YAML中的input.strategy5分钟生效。心得Skill是肌肉Plan是大脑。让肌肉只听指令别让它自己思考。教训二健康检查必须模拟真实调用不能只ping端口早期我们为send_email_reportSkill设置健康检查为curl -I http://mail-service:8080/health返回200就认为OK。但某次邮件网关升级后/health仍返回200实际发信API因鉴权方式变更全部失败。后来我们将健康检查改为curl -X POST http://mail-service:8080/test -d {to:testxxx.com}真实走一遍发信链路。心得健康检查不是检查服务“活着”而是检查它“能干活”。教训三Plan日志必须包含输入参数哈希值而非明文最初日志记录完整input方便调试。但在某次审计中发现日志中明文记录了数据库密码因某Skill错误地将凭证写入input。我们立即整改日志中input_hash字段存储sha256(json.dumps(input)).hexdigest()同时建立日志脱敏规则禁止任何字段含password、token、secret等关键词。心得可观测性不等于裸奔。安全是默认选项不是事后补丁。4.3 性能调优实战如何让Plan执行稳定在1秒内MCPSkill的性能瓶颈通常不在模型而在IO。我们通过三步优化将95%的Plan执行时间压到800ms内Step 1Skill并发执行非顺序Plan中无依赖关系的步骤如fetch_google和fetch_meta可并行执行。我们在执行引擎中引入asyncio.gather# 识别无依赖步骤 independent_steps [s for s in plan[plan] if not depends_on_other_step(s, plan)] # 并发执行 results await asyncio.gather(*[execute_step_async(s) for s in independent_steps])实测5个独立API调用并行后耗时从1200ms降至450msP95。Step 2输入参数预编译每次执行都解析YAML/JSON输入开销不小。我们将input参数在Mission注册时预编译为Python字典并缓存# 缓存key: mission_id step_id input_yaml_hash cached_input cache.get(f{mission_id}_{step_id}_{hash(input_yaml)}) if not cached_input: cached_input yaml.safe_load(input_yaml) # 或json.loads cache.set(f{mission_id}_{step_id}_{hash(input_yaml)}, cached_input, timeout3600)减少重复解析单步节省15-20ms。Step 3日志异步批处理原同步写ELK每步耗时80-120ms。改为执行引擎内存中暂存日志最多100条启动独立线程每200ms或日志达50条时批量POST到Logstash日志丢失容忍最坏情况丢失200ms日志但执行不阻塞优化后日志写入耗时从100ms降至5msP95Plan整体提速12%。5. 进阶应用MCPSkill 如何支撑更复杂的智能体场景5.1 支持条件分支与循环突破线性Plan限制MCPSkill天然支持复杂流程只需扩展Plan语法。我们定义了两个关键字if和for。条件分支示例基于数据源可用性plan: - id: check_bi_status skill: check_service_health input: service: internal_bi - id: fetch_data if: {{ step_outputs.check_bi_status.status up }} then: skill: fetch_bi_data input: {report_id: ad_roi_summary} else: skill: fetch_fallback_data # 从缓存或默认值获取 input: {source: cache}执行引擎解析时先计算if表达式的布尔值使用jinja2.Template渲染再决定执行then或else分支。step_outputs可访问任意前序步骤输出实现动态决策。循环示例批量处理订单plan: - id: get_order_ids skill: fetch_pending_orders input: {limit: 100} - id: process_orders for: {{ step_outputs.get_order_ids.order_ids }} item: order_id do: skill: process_single_order input: {order_id: {{ item }}}引擎将get_order_ids输出的order_ids数组展开为每个order_id生成一个独立的process_single_order执行实例。所有实例共享同一mission_id日志中step_id自动追加索引如process_orders_0,process_orders_1便于追踪。注意循环内Skill执行默认串行避免并发冲击下游服务。如需并发加concurrency: 5字段引擎将按5个一组分批执行。5.2 与RAG结合让规划理解私域知识纯MCP规划依赖模型通用知识但业务规则常在内部文档中。我们通过RAG增强Mission解析知识注入将公司《广告投放SOP》《数据权限手册》等PDF转为向量存入ChromaDBMission解析增强当收到新Mission时执行引擎先用Mission描述如“按SOP第3.2条处理高价值客户”查询RAG返回相关段落Plan生成引导将RAG结果作为Context传给规划模型提示“请严格依据以下SOP条款生成Plan{rag_result}”Skill参数填充RAG可返回具体参数值如SOP中规定“高价值客户指ARPU50
