AI Agent Harness Engineering 研发流程规范：从需求分析到版本迭代的标准化

写在前面的话：作为一名在云原生+大模型领域摸爬滚打近8年的架构师（注：任务设定调整为贴近Agent领域的资深从业者），我见过太多团队在Agent开发初期“野蛮生长”：要么是几个Prompt工程师临时凑出来一个Demo扔给产品，要么是全栈工程师凭感觉把大模型API、工具链、向量库随便堆在一起——上线后三天两头出幻觉、工具调用链断裂、用户上下文丢失、延迟爆炸。直到2024年Q2我们团队主导的「企业级智能客服Agent平台」上线时踩过9次大版本回滚、27次线上核心模块P0级故障的坑，才痛定思痛，建立了一套从需求拆解的“分层Prompt框架”锚点，到Agent运行时的“可观测性-可调试性-可干预性三位一体”设计，再到灰度发布的“幻觉阈值自动监控+人工快速干预闭环”迭代的完整Harness Engineering（可以译为“AI智能体研发工程化体系”）规范。

本文将系统梳理这套规范，覆盖从需求到迭代的全生命周期，附大量企业级Python/Go混合代码示例、数学模型（幻觉检测、工具链调度优化）、Mermaid架构图/时序图/ER图，并给出踩坑后的最佳实践Tips，目标是帮助团队把Agent开发从“手工作坊”带入“工业化流水线”，将Agent项目的上线周期缩短60%以上、P0级故障发生率降低90%、用户留存率提升35%（我们内部真实数据）。

---

1. 核心概念与问题背景

1.1 核心概念定义

1.1.1 什么是AI Agent？

AI Agent（人工智能智能体）是指具备感知能力、推理决策能力、行动执行能力、学习进化能力的自主软件实体。它可以通过大语言模型（LLM）、多模态模型（VLM）、工具链（Function Calling、Plugin、API）、知识库（向量库、图数据库）、记忆系统（短期记忆、长期记忆、工作记忆）来完成复杂的任务，比如企业级智能客服、代码审计机器人、自动化运维专家、智能内容创作助手等。

为了更清晰地定义企业级AI Agent，我们可以用一个数学表达式来描述其完整的生命周期：

$$A(t)=\mathcal{P}\left(\mathcal{M}\left(\mathcal{S}(t),\mathcal{K},\mathcal{H}(t),\mathcal{I}(t)\right)\right)$$

其中：

• $A(t)$：Agent在时间$t$时刻的最终输出/决策/行动
• $\mathcal{S}(t)$：Agent在时间$t$时刻的感知输入（包括用户Query、传感器数据、历史环境交互数据）
• $\mathcal{K}$：Agent的静态知识库（企业文档、业务规则、产品手册等，通常存储在向量库+图数据库的混合存储中）
• $\mathcal{H}(t)$：Agent的长期记忆（截至时间$t$的用户对话历史、工具执行历史、错误修正历史等，通常存储在关系型数据库+NoSQL数据库的混合存储中）
• $\mathcal{I}(t)$：Agent的短期工作记忆/指令上下文（包括当前任务的核心指令、临时变量、工具调用链的中间结果等，通常存储在Redis这类高性能内存数据库中）
• $\mathcal{M}$：Agent的推理决策模型（核心是LLM/VLM，配合工具调度器、幻觉检测器、上下文压缩器、Prompt优化器等组件）
• $\mathcal{P}$：Agent的行动执行器（包括工具执行器、内容生成器、API调用器等，需要处理工具重试、错误降级、响应格式化等问题）

为了直观展示这个数学模型，我们可以画一个核心要素组成的Mermaid架构图：

1.1.2 什么是AI Agent Harness Engineering？

Harness在英文中的意思是“马具、挽具”，也可以引申为“控制、利用、工程化支撑体系”。AI Agent Harness Engineering（AI智能体研发工程化体系）是一套专门为AI Agent开发设计的、覆盖全生命周期的标准化流程、工具链、最佳实践、质量保障体系，它的核心目标是：

1. 降低Agent开发的门槛：让没有Prompt经验的全栈工程师也能快速开发出稳定的企业级Agent
2. 提高Agent开发的效率：通过复用Prompt模板、工具链组件、知识库构建模板、可观测性组件，将上线周期缩短60%以上
3. 保障Agent运行的稳定性：通过可观测性三位一体、幻觉监控、灰度发布、快速回滚等机制，将P0级故障发生率降低90%
4. 提升Agent的用户体验：通过分层Prompt框架锚定业务规则、上下文压缩器避免Prompt溢出、强化学习优化工具调度，将用户留存率提升35%

为了更清晰地对比传统软件开发工程化体系和AI Agent Harness Engineering体系的差异，我们可以列一个核心属性维度对比的Markdown表格：

核心属性维度	传统软件开发工程化体系	AI Agent Harness Engineering体系
核心开发语言/工具	Java/Python/Go/JavaScript + 框架（Spring/Django）	LLM/VLM API + Prompt模板库 + 工具链 + 向量库 + 记忆系统
需求拆解方法	功能拆解 + 用例图 + 流程图	分层Prompt框架锚点拆解 + 业务规则图 + 幻觉边界定义 + 工具链清单
核心质量指标	功能正确性 + 性能（TPS/延迟） + 稳定性（可用性）	事实一致性（无幻觉率） + 任务完成率 + 用户满意度 + 工具调用准确率
调试方法	断点调试 + 日志分析	可观测性三位一体 + Prompt调试控制台 + 幻觉追溯工具
测试方法	单元测试 + 集成测试 + 端到端测试 + 性能测试	幻觉测试 + 任务完成率测试 + 对抗性测试 + 多模态测试 + 性能测试
发布方法	蓝绿发布 + 滚动发布	幻觉阈值自动监控的灰度发布 + 多Prompt模板的A/B测试 + 快速人工干预回滚
迭代方法	功能迭代 + 性能优化	幻觉修正 + 工具链优化 + Prompt迭代 + 知识库更新 + 强化学习微调
核心难点	高并发 + 分布式一致性 + 性能瓶颈	幻觉控制 + 工具链调度优化 + 上下文管理 + Prompt工程化

1.1.3 核心组成部分的ER实体关系图

为了更清晰地展示Harness Engineering体系核心组成部分之间的关系，我们可以画一个Mermaid ER图：

1.2 问题背景与痛点

1.2.1 需求阶段的痛点

1. 需求模糊不清：产品经理往往只会说“我要一个像ChatGPT一样的智能客服”，但不会明确说明“哪些问题必须用知识库回答、哪些问题必须调用内部API、哪些问题必须拒绝回答、幻觉的边界在哪里”
2. Prompt工程化缺失：Prompt工程师往往凭感觉写Prompt，没有标准化的框架，导致同一个需求不同的工程师写出来的Prompt效果差异巨大，而且难以维护和迭代
3. 业务规则与Prompt分离：业务规则往往写在产品文档里，而不是嵌入到Prompt中或者知识库的图规则里，导致LLM经常违反业务规则，比如给VIP用户错误的折扣、泄露内部数据等

1.2.2 开发阶段的痛点

1. 组件复用率低：每个Agent项目都要从零开始搭建LLM API代理、向量库查询接口、工具链接口、记忆系统接口，导致开发效率极低
2. 上下文管理混乱：没有标准化的短期工作记忆/长期记忆策略，导致LLM经常忘记用户之前说的话，比如用户刚说“我要查一下我的订单号123456的物流”，下一句LLM就问“你的订单号是什么”
3. 工具链调度困难：没有标准化的工具调度器，导致LLM经常调用错误的工具、调用工具的顺序不对、调用工具的参数不对，比如查物流应该调用“物流查询API”，但LLM调用了“订单查询API”；查订单号123456的物流应该先调用“订单查询API”获取物流单号，再调用“物流查询API”，但LLM直接用订单号调用了“物流查询API”

1.2.3 测试阶段的痛点

1. 传统测试方法无效：传统的单元测试、集成测试、端到端测试只能测试代码的正确性，无法测试LLM的输出是否符合业务规则、是否有幻觉、工具调用是否正确
2. 测试用例覆盖不足：企业级Agent的用户Query可能有百万级甚至千万级，手动写测试用例覆盖所有场景几乎不可能
3. 对抗性测试缺失：没有对抗性测试，导致Agent很容易被用户的恶意Query攻击，比如泄露内部数据、生成有害内容等

1.2.4 上线阶段的痛点

1. 稳定性保障不足：没有可观测性三位一体的设计，导致Agent出问题时很难定位原因，比如用户说“我的物流查不到”，但不知道是LLM的问题、工具调用的问题、还是向量库查询的问题
2. 灰度发布缺失：直接全量发布，导致一旦出问题就会影响所有用户
3. 快速回滚困难：没有多Prompt模板的版本管理、没有快速人工干预的机制，导致回滚需要很长时间

1.2.5 迭代阶段的痛点

1. 幻觉修正困难：没有幻觉追溯工具，导致很难找到幻觉产生的原因，比如Agent说“我们的产品有终身保修”，但不知道是Prompt的问题、还是知识库的问题、还是LLM本身的问题
2. 知识库更新困难：没有标准化的知识库更新流程，导致知识库更新后Agent的效果反而变差
3. Prompt迭代困难：没有Prompt版本管理、没有A/B测试的机制，导致很难判断新的Prompt是否比旧的Prompt效果好

---

2. 需求分析阶段：分层Prompt框架锚点与业务边界定义

2.1 核心概念与方法

2.1.1 什么是分层Prompt框架？

分层Prompt框架（Layered Prompt Framework，LPF）是我们团队在踩过无数Prompt工程的坑之后总结出来的一套标准化Prompt编写框架，它的核心思想是将复杂的Agent需求拆解成多个独立的、可复用的、可迭代的Prompt层，每个Prompt层只负责一个明确的任务，然后通过一个“Prompt组装器”将这些层组装成最终的Prompt发送给LLM。

2.1.2 分层Prompt框架的组成部分

分层Prompt框架通常由以下7个核心层组成（可以根据实际需求增加或删除层）：

1. 系统身份层（System Identity Layer）：定义Agent的身份、角色、职责、语气等，比如“你是某电商平台的VIP智能客服，你的职责是为VIP用户提供专业、热情、高效的服务，你的语气要友好、耐心、正式，但不要过于生硬”
2. 业务规则层（Business Rules Layer）：定义Agent必须遵守的所有业务规则，包括正向规则（必须做什么）和反向规则（绝对不能做什么），比如“正向规则1：如果用户是VIP用户，必须优先处理他们的问题；正向规则2：如果用户查物流，必须先获取订单号，再获取物流单号，最后查询物流信息；反向规则1：绝对不能泄露内部数据，比如员工工资、内部会议内容；反向规则2：绝对不能生成有害内容，比如色情、暴力、政治敏感内容；反向规则3：绝对不能给非VIP用户提供VIP专属服务，比如专属折扣、专属客服通道”
3. 幻觉边界层（Hallucination Boundary Layer）：定义Agent必须拒绝回答的问题和必须从知识库或工具中获取答案的问题，比如“问题类型1：如果问题涉及企业尚未公开的产品信息、技术信息、财务信息，必须拒绝回答，并说‘抱歉，这个问题我暂时无法回答，请关注我们的官方公告’；问题类型2：如果问题涉及企业的产品信息、服务信息、业务规则，必须优先从知识库中获取答案，如果知识库中没有答案，再调用工具获取；问题类型3：如果问题涉及用户的个人信息、订单信息、物流信息，必须调用内部API获取，绝对不能凭感觉回答”
4. 工具链清单层（Toolchain Inventory Layer）：列出Agent可以调用的所有工具，包括工具名称、工具描述、工具参数、工具返回值格式等，这一层通常会配合Function Calling的格式来写，比如“你可以调用以下工具来完成任务：

• 工具名称：get_user_info，工具描述：获取用户的基本信息，包括用户ID、用户类型（VIP/普通用户）、注册时间等，工具参数：user_id: str（必填，用户的唯一标识符），工具返回值格式：JSON
• 工具名称：get_order_info，工具描述：获取用户的订单信息，包括订单号、订单状态、商品信息、物流单号等，工具参数：user_id: str（必填）、order_id: str（可选，订单号，如果不填则返回用户最近10笔订单的信息），工具返回值格式：JSON
• 工具名称：get_logistics_info，工具描述：获取订单的物流信息，包括物流单号、物流公司、物流状态、物流轨迹等，工具参数：logistics_id: str（必填，物流单号），工具返回值格式：JSON”

5. 对话历史层（Dialogue History Layer）：包含用户和Agent的历史对话记录，这一层通常会经过上下文压缩器的处理，避免Prompt溢出
6. 短期工作记忆层（Short-term Working Memory Layer）：包含当前任务的核心指令、临时变量、工具调用链的中间结果等，这一层通常存储在Redis中
7. 当前用户Query层（Current User Query Layer）：包含用户当前的输入，比如文本、语音转文本、图像转文本等

2.1.3 分层Prompt框架的数学模型

分层Prompt框架的最终Prompt $P_{final}$ 可以用一个数学表达式来描述：

$$P_{final}=\mathcal{A}\left(P_{identity},P_{rules},P_{boundary},P_{tools},\mathcal{C}\left(P_{history},L_{max\_history}\right),P_{working\_memory},P_{current\_query}\right)$$

其中：

• $\mathcal{A}$：Prompt组装器，负责将各个Prompt层按照指定的顺序组装成最终的Prompt
• $P_{identity}$：系统身份层Prompt
• $P_{rules}$：业务规则层Prompt
• $P_{boundary}$：幻觉边界层Prompt
• $P_{tools}$：工具链清单层Prompt
• $\mathcal{C}$：上下文压缩器，负责压缩对话历史层Prompt，使其长度不超过 $L_{max\_history}$（这个值通常是LLM上下文窗口的20%-30%）
• $P_{history}$：原始对话历史层Prompt
• $L_{max\_history}$：对话历史层的最大允许长度（Token数）
• $P_{working\_memory}$：短期工作记忆层Prompt
• $P_{current\_query}$：当前用户Query层Prompt

为了直观展示分层Prompt框架的组装流程，我们可以画一个Mermaid流程图：

2.2 需求拆解的具体操作步骤

2.2.1 步骤1：召开需求拆解会议

需求拆解会议必须由产品经理、Prompt工程师、全栈工程师、业务专家、测试工程师共同参加，会议时长通常为2-4小时，会议的核心目标是：

1. 明确Agent的核心功能和边界：回答“Agent能做什么？不能做什么？”
2. 定义业务规则的正向/反向清单：回答“Agent必须遵守哪些规则？”
3. 定义幻觉边界的问题类型清单：回答“Agent必须拒绝回答哪些问题？必须从哪里获取答案？”
4. 定义工具链的清单：回答“Agent需要调用哪些工具？工具的参数和返回值是什么？”
5. 定义测试用例的核心场景：回答“Agent需要覆盖哪些核心测试场景？”

2.2.2 步骤2：填写《Agent需求拆解文档》

《Agent需求拆解文档》是我们团队标准化的需求文档模板，它包含以下10个核心章节：

1. 项目概述：包括项目名称、项目背景、项目目标、项目范围、目标用户群体
2. 核心功能清单：用表格列出Agent的所有核心功能，包括功能ID、功能名称、功能描述、优先级（P0/P1/P2/P3）
3. 业务边界定义：用表格列出Agent能做的事情和不能做的事情，包括边界ID、边界类型（能做/不能做）、边界描述
4. 业务规则清单：用表格列出Agent必须遵守的所有业务规则，包括规则ID、规则类型（正向/反向）、规则描述、规则来源（产品文档/业务专家/法律合规）
5. 幻觉边界定义：用表格列出Agent必须拒绝回答的问题类型和必须从哪里获取答案的问题类型，包括边界ID、边界类型（拒绝回答/知识库获取/工具获取）、边界描述、拒绝回答的话术/获取答案的优先级
6. 工具链清单：用表格列出Agent可以调用的所有工具，包括工具ID、工具名称、工具描述、工具类型（内部API/外部API/内置工具）、工具参数（用JSON Schema描述）、工具返回值（用JSON Schema描述）、工具调用的前置条件、工具调用的错误处理策略
7. 核心测试场景清单：用表格列出Agent需要覆盖的所有核心测试场景，包括场景ID、场景描述、用户Query示例、预期输出示例、优先级（P0/P1/P2/P3）
8. 质量指标定义：用表格列出Agent的所有质量指标，包括指标ID、指标名称、指标定义、目标值、统计方法、统计周期
9. 非功能需求：包括性能需求（TPS/延迟）、可用性需求、可扩展性需求、安全性需求、可维护性需求
10. 时间规划：包括需求分析阶段、开发阶段、测试阶段、上线阶段、迭代阶段的时间节点

2.2.3 步骤3：基于《Agent需求拆解文档》编写分层Prompt框架的初始版本

Prompt工程师需要基于《Agent需求拆解文档》中的系统身份层定义、业务规则清单、幻觉边界定义、工具链清单来编写分层Prompt框架的初始版本，每个Prompt层都必须独立存储在一个Markdown文件中，并且必须添加版本号和注释，方便后续的维护和迭代。

为了更清晰地展示分层Prompt框架的初始版本，我们可以用某电商平台的VIP智能客服Agent为例，给出每个Prompt层的Python代码示例（代码中会读取对应的Markdown文件）：

首先，我们需要创建一个项目目录结构：

vip_customer_service_agent/
├── prompts/
│   ├── v0.1/
│   │   ├── 01_system_identity.md
│   │   ├── 02_business_rules.md
│   │   ├── 03_hallucination_boundary.md
│   │   ├── 04_toolchain_inventory.md
│   │   └── assembly_config.yaml
└── prompt_manager.py

然后，我们来编写每个Markdown文件：

• prompts/v0.1/01_system_identity.md：

  # 系统身份层（System Identity Layer）
  # 版本号：v0.1
  # 最后更新时间：2024-06-01
  # 更新人：张三（Prompt工程师）
  
  你是「未来商城」的VIP专属智能客服，你的职责是为VIP用户提供专业、热情、高效的服务，你的工作时间是每天的9:00-21:00。
  
  你的语气要求：
  1. 友好、耐心、正式，但不要过于生硬
  2. 要使用「您」来称呼用户，不要使用「你」
  3. 要主动提供帮助，比如用户说「我要查物流」，你要主动说「好的，请您提供一下您的订单号，或者我可以帮您查询您最近10笔订单的信息」
  4. 如果用户的问题解决了，要主动问「您还有其他问题吗？」
  

• prompts/v0.1/02_business_rules.md：

  # 业务规则层（Business Rules Layer）
  # 版本号：v0.1
  # 最后更新时间：2024-06-01
  # 更新人：张三（Prompt工程师）
  
  ## 正向规则（Must Do）
  1. 如果用户是VIP用户，必须优先处理他们的问题，不要让他们等待超过30秒
  2. 如果用户查物流，必须先获取用户的信息，再确认用户是否提供了订单号：
     a. 如果用户提供了订单号，必须先调用 `get_order_info` 工具获取物流单号，再调用 `get_logistics_info` 工具获取物流信息
     b. 如果用户没有提供订单号，必须先调用 `get_order_info` 工具获取用户最近10笔订单的信息，让用户选择要查询的订单
  3. 如果用户查询VIP专属服务（比如专属折扣、专属客服通道），必须先调用 `get_user_info` 工具确认用户的VIP等级，再根据VIP等级提供相应的服务
  4. 如果用户的问题无法从知识库或工具中获取答案，必须主动将用户转接到人工VIP专属客服
  5. 如果用户对服务不满意，必须主动道歉，并询问用户的具体意见
  
  ## 反向规则（Must NOT Do）
  1. 绝对不能泄露内部数据，比如员工工资、内部会议内容、未公开的产品信息、技术信息、财务信息
  2. 绝对不能生成有害内容，比如色情、暴力、政治敏感内容、虚假信息
  3. 绝对不能给非VIP用户提供VIP专属服务，比如专属折扣、专属客服通道
  4. 绝对不能代替用户做决定，比如代替用户申请退款、代替用户修改订单信息，必须让用户自己操作或者提供操作指引
  5. 绝对不能在工作时间之外（21:00-9:00）提供服务，必须说「抱歉，现在是我们的非工作时间，我们的工作时间是每天的9:00-21:00，请您在工作时间内咨询，谢谢您的理解」
  

• prompts/v0.1/03_hallucination_boundary.md：

  # 幻觉边界层（Hallucination Boundary Layer）
  # 版本号：v0.1
  # 最后更新时间：2024-06-01
  # 更新人：张三（Prompt工程师）
  
  ## 必须拒绝回答的问题类型
  | 问题类型ID | 问题类型描述 | 拒绝回答的话术 |
  |------------|--------------|----------------|
  | RB001      | 涉及企业尚未公开的产品信息、技术信息、财务信息 | 抱歉，这个问题我暂时无法回答，请关注我们的官方公告 |
  | RB002      | 涉及企业的商业机密 | 抱歉，这个问题我无法回答 |
  | RB003      | 涉及其他用户的个人信息 | 抱歉，我无法查询其他用户的个人信息，请您提供自己的信息 |
  | RB004      | 涉及生成有害内容 | 抱歉，我无法生成这样的内容 |
  
  ## 必须从知识库中获取答案的问题类型
  | 问题类型ID | 问题类型描述 | 获取答案的优先级 |
  |------------|--------------|------------------|
  | KB001      | 涉及企业的产品信息（比如产品参数、产品价格、产品保修政策） | 优先从向量知识库中获取，如果向量知识库中没有答案，再从图规则知识库中获取，如果都没有答案，再调用工具获取（如果有相关工具） |
  | KB002      | 涉及企业的服务信息（比如配送政策、退换货政策、售后服务政策） | 同上 |
  | KB003      | 涉及企业的常见问题（FAQ） | 优先从向量知识库中获取 |
  
  ## 必须从工具中获取答案的问题类型
  | 问题类型ID | 问题类型描述 | 获取答案的优先级 |
  |------------|--------------|------------------|
  | TL001      | 涉及用户的个人信息（比如用户ID、用户类型、注册时间、VIP等级） | 必须调用 `get_user_info` 工具获取 |
  | TL002      | 涉及用户的订单信息（比如订单号、订单状态、商品信息、物流单号） | 必须调用 `get_order_info` 工具获取 |
  | TL003      | 涉及订单的物流信息（比如物流单号、物流公司、物流状态、物流轨迹） | 必须先调用 `get_order_info` 工具获取物流单号，再调用 `get_logistics_info` 工具获取 |
  

• prompts/v0.1/04_toolchain_inventory.md：

  # 工具链清单层（Toolchain Inventory Layer）
  # 版本号：v0.1
  # 最后更新时间：2024-06-01
  # 更新人：张三（Prompt工程师）
  
  你可以调用以下工具来完成任务，请严格按照工具的参数和返回值格式来调用：
  
  ---
  
  ## 工具名称：`get_user_info`
  工具类型：内部API
  工具描述：获取用户的基本信息，包括用户ID、用户类型（VIP/普通用户）、注册时间、VIP等级、VIP到期时间等
  工具调用的前置条件：用户已经登录（如果用户没有登录，必须引导用户登录）
  工具调用的错误处理策略：
  - 如果工具返回 `{"code": 401, "message": "用户未登录"}`，必须引导用户登录
  - 如果工具返回 `{"code": 404, "message": "用户不存在"}`，必须说「抱歉，我无法查询您的信息，请您检查您的登录状态」
  - 如果工具返回 `{"code": 500, "message": "服务器内部错误"}`，必须说「抱歉，系统暂时出现故障，请您稍后再试，或者转接到人工VIP专属客服」
  - 如果工具返回其他错误码，必须说「抱歉，系统暂时出现故障，请您稍后再试」
  
  ### 工具参数（JSON Schema）
  ```json
  {
    "type": "object",
    "properties": {
      "user_id": {
        "type": "string",
        "description": "用户的唯一标识符，从用户的登录信息中获取",
        "minLength": 1
      }
    },
    "required": ["user_id"],
    "additionalProperties": false
  }
  

  工具返回值（JSON Schema）

  {
    "type": "object",
    "properties": {
      "code": {
        "type": "integer",
        "description": "响应状态码，200表示成功，其他表示失败"
      },
      "message": {
        "type": "string",
        "description": "响应消息"
      },
      "data": {
        "type": "object",
        "properties": {
          "user_id": {
            "type": "string",
            "description": "用户的唯一标识符"
          },
          "user_type": {
            "type": "string",
            "enum": ["VIP", "NORMAL"],
            "description": "用户类型"
          },
          "register_time": {
            "type": "string",
            "format": "date-time",
            "description": "用户的注册时间"
          },
          "vip_level": {
            "type": "integer",
            "minimum": 1,
            "maximum": 5,
            "description": "VIP等级（1-5，5级最高），只有user_type为VIP时才有值"
          },
          "vip_expire_time": {
            "type": "string",
            "format": "date-time",
            "description": "VIP到期时间，只有user_type为VIP时才有值"
          }
        },
        "required": ["user_id", "user_type", "register_time"],
        "additionalProperties": false
      }
    },
    "required": ["code", "message"],
    "additionalProperties": false
  }
  

  ---

  工具名称：get_order_info

  工具类型：内部API
  工具描述：获取用户的订单信息，包括订单号、订单状态、商品信息、物流单号等
  工具调用的前置条件：用户已经登录（如果用户没有登录，必须引导用户登录）
  工具调用的错误处理策略：

  • 同 get_user_info 工具的错误处理策略

  工具参数（JSON Schema）

  {
    "type": "object",
    "properties": {
      "user_id": {
        "type": "string",
        "description": "用户的唯一标识符，从用户的登录信息中获取",
        "minLength": 1
      },
      "order_id": {
        "type": "string",
        "description": "订单号，如果不填则返回用户最近10笔订单的信息",
        "minLength": 1,
        "optional": true
      }
    },
    "required": ["user_id"],
    "additionalProperties": false
  }
  

  工具返回值（JSON Schema）

  （为了节省篇幅，这里省略了JSON Schema，实际开发中必须详细定义）

  ---

  工具名称：get_logistics_info

  工具类型：内部API
  工具描述：获取订单的物流信息，包括物流单号、物流公司、物流状态、物流轨迹等
  工具调用的前置条件：用户已经登录，并且已经获取了物流单号
  工具调用的错误处理策略：

  • 同 get_user_info 工具的错误处理策略

  工具参数（JSON Schema）

  （为了节省篇幅，这里省略了JSON Schema，实际开发中必须详细定义）

  工具返回值（JSON Schema）

  （为了节省篇幅，这里省略了JSON Schema，实际开发中必须详细定义）

• prompts/v0.1/assembly_config.yaml：

  # Prompt组装器配置文件
  # 版本号：v0.1
  # 最后更新时间：2024-06-01
  # 更新人：张三（Prompt工程师）
  
  # Prompt层的组装顺序（严格按照这个顺序组装）
  assembly_order:
    - "01_system_identity.md"
    - "02_business_rules.md"
    - "03_hallucination_boundary.md"
    - "04_toolchain_inventory.md"
    - "compressed_dialogue_history"
    - "short_term_working_memory"
    - "current_user_query"
  
  # 上下文压缩器配置
  context_compressor:
    # 对话历史层的最大允许长度（Token数），这里设置为LLM上下文窗口的20%
    # 假设我们使用的是GPT-4o，上下文窗口是128K Tokens，所以这里设置为25600
    max_history_tokens: 25600
    # 上下文压缩方法：bm25（BM25检索）+ llm_summarize（LLM摘要）
    compression_method: "bm25_plus_llm_summarize"
    # BM25检索的Top-K值
    bm25_top_k: 10
    # LLM摘要的最大长度（Token数）
    llm_summarize_max_tokens: 500
  
  # 短期工作记忆配置
  short_term_working_memory:
    # 短期工作记忆的最大允许长度（Token数）
    max_working_memory_tokens: 1024
    # 短期工作记忆的过期时间（秒）
    expire_time: 3600
  

最后，我们来编写prompt_manager.py，实现Prompt组装器的功能：

"""
Prompt管理器，负责管理分层Prompt框架的所有Prompt层，
包括Prompt层的读取、版本管理、上下文压缩、组装等功能
"""
import os
import yaml
import json
import tiktoken
from typing import List, Dict, Any, Optional
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.retrievers import BM25Retriever
from langchain.schema import Document
from openai import OpenAI

# 初始化OpenAI客户端（这里使用的是OpenAI的API，实际开发中可以换成其他LLM的API）
# 注意：需要在环境变量中设置OPENAI_API_KEY
client = OpenAI()

# 初始化TikToken编码器（用于计算Token数）
# 这里使用的是GPT-4o的编码器，实际开发中可以换成其他LLM的编码器
encoding = tiktoken.encoding_for_model("gpt-4o")


class PromptManager:
    def __init__(self, prompts_dir: str, version: str):
        """
        初始化Prompt管理器
        :param prompts_dir: Prompt层的存储目录
        :param version: Prompt框架的版本号
        """
        self.prompts_dir = prompts_dir
        self.version = version
        self.version_dir = os.path.join(prompts_dir, version)
        self.assembly_config_path = os.path.join(self.version_dir, "assembly_config.yaml")
        self.assembly_config = self._load_assembly_config()
        self.prompt_layers = self._load_prompt_layers()

    def _load_assembly_config(self) -> Dict[str, Any]:
        """
        加载Prompt组装器配置文件
        :return: 配置字典
        """
        if not os.path.exists(self.assembly_config_path):
            raise FileNotFoundError(f"Prompt组装器配置文件不存在：{self.assembly_config_path}")
        with open(self.assembly_config_path, "r", encoding="utf-8") as f:
            return yaml.safe_load(f)

    def _load_prompt_layers(self) -> Dict[str, str]:
        """
        加载所有静态Prompt层（不包括compressed_dialogue_history、short_term_working_memory、current_user_query）
        :return: Prompt层字典，key是Prompt层的文件名，value是Prompt层的内容
        """
        prompt_layers = {}
        for filename in os.listdir(self.version_dir):
            if filename.endswith(".md") and filename not in ["assembly_config.yaml"]:
                file_path = os.path.join(self.version_dir, filename)
                with open(file_path, "r", encoding="utf-8") as f:
                    prompt_layers[filename] = f.read().strip()
        # 验证静态Prompt层是否完整
        for filename in self.assembly_config["assembly_order"]:
            if filename.endswith(".md") and filename not in prompt_layers:
                raise FileNotFoundError(f"静态Prompt层不存在：{filename}")
        return prompt_layers

    def _count_tokens(self, text: str) -> int:
        """
        计算文本的Token数
        :param text: 文本
        :return: Token数
        """
        return len(encoding.encode(text))

    def _compress_dialogue_history(self, dialogue_history: List[Dict[str, str]]) -> str:
        """
        压缩对话历史
        :param dialogue_history: 对话历史列表，每个元素是一个字典，包含"role"和"content"
        :return: 压缩后的对话历史
        """
        # 首先将对话历史转换成字符串格式
        history_text = ""
        for item in dialogue_history:
            history_text += f"{item['role']}: {item['content']}\n"
        # 计算原始对话历史的Token数
        original_tokens = self._count_tokens(history_text)
        max_history_tokens = self.assembly_config["context_compressor"]["max_history_tokens"]
        # 如果原始对话历史的Token数小于等于最大允许长度，直接返回
        if original_tokens <= max_history_tokens:
            return history_text.strip()
        # 否则，使用配置的压缩方法压缩对话历史
        compression_method = self.assembly_config["context_compressor"]["compression_method"]
        if compression_method == "bm25_plus_llm_summarize":
            return self._compress_dialogue_history_bm25_plus_llm_summarize(
                dialogue_history, max_history_tokens
            )
        else:
            # 默认使用截断最早的对话记录的方法
            return self._compress_dialogue_history_truncate(dialogue_history, max_history_tokens)

    def _compress_dialogue_history_truncate(self, dialogue_history: List[Dict[str, str]], max_history_tokens: int) -> str:
        """
        使用截断最早的对话记录的方法压缩对话历史
        :param dialogue_history: 对话历史列表
        :param max_history_tokens: 最大允许长度（Token数）
        :return: 压缩后的对话历史
        """
        # 从后往前遍历对话历史，累加Token数，直到超过最大允许长度
        compressed_history = []
        total_tokens = 0
        for item in reversed(dialogue_history):
            item_text = f"{item['role']}: {item['content']}\n"
            item_tokens =