from openai import OpenAI

client = OpenAI()

class SingleAgentHarness:
   """单Agent全闭环Harness范式实现"""
   def __init__(self, system_rules: str, max_retry: int = 3):
      self.system_rules = system_rules # Harness核心：定死行为边界
      self.max_retry = max_retry # 重试兜底机制
      self.conversation_history = [] # 会话状态管理

   def _tool_call_control(self, tool_name: str, params: dict) -> str:
   """   Harness工具管控：白名单机制，防止AI乱调用工具"""
      allowed_tools = ["read_file", "write_file", "run_bash", "excel_process"]
      if tool_name not in allowed_tools:
         return f"错误：禁止调用工具{tool_name}，仅允许调用{allowed_tools}"
      # 此处可扩展具体工具实现
      return f"工具{tool_name}执行完成，参数：{params}"

   def _result_validate(self, result: str, task: str) -> tuple[bool, str]:
      """Harness闭环核心：后置结果校验，不合格直接打回"""
      validate_prompt = f"""
      核心任务：{task}
      交付结果：{result}
      校验规则：
      1. 是否完整完成核心任务
      2. 是否符合预设的系统规则
      3. 是否有错误、遗漏、不符合要求的内容
      只输出【PASS】或【FAIL】，并说明具体原因
      """
      response = client.chat.completions.create(
      model="gpt-4o",
         messages=[{"role": "user", "content": validate_prompt}],
         temperature=0
      )
        validate_result = response.choices[0].message.content
      return "PASS" in validate_result, validate_result

   def run(self, task: str) -> str:
   """Harness全链路闭环执行入口"""
      # 1. 前置规则注入，从源头管控行为
      self.conversation_history = [
         {"role": "system", "content": self.system_rules},
         {"role": "user", "content": f"核心任务：{task}，严格遵守系统规则执行"}
      ]
      retry_count = 0

      while retry_count < self.max_retry:
         # 2. 模型推理执行
         response = client.chat.completions.create(
            model="gpt-4o",
            messages=self.conversation_history,
            temperature=0.3
         )
         result = response.choices[0].message.content
         self.conversation_history.append({"role": "assistant", "content": result})

      # 3. 工具调用管控
      if "调用工具" in result:
          tool_result = self._tool_call_control("demo_tool", {"param": "demo"})
         self.conversation_history.append({"role": "user", "content": f"工具执行结果：{tool_result}"})
         continue

       # 4. 后置结果校验，闭环核心
      is_pass, validate_msg = self._result_validate(result, task)
      if is_pass:
            return f"任务执行完成：\n{result}\n校验结果：{validate_msg}"

         # 5. 失败重试，注入错误信息
         retry_count += 1
         self.conversation_history.append({
            "role": "user",
               "content": f"结果不符合要求，失败原因：{validate_msg}，请重新修改，重试次数：{retry_count}/{self.max_retry}"
         })

       return f"任务执行失败，达到最大重试次数，最终结果：{result}"

# 直接使用示例
if __name__ == "__main__":
      # 定义Harness核心规则，定死Agent的行为边界
      system_rules = """
      你是Java代码生成助手，必须严格遵守以下规则：
      1. 所有代码符合SpringBoot 3.2规范，严格遵循分层架构
      2. 所有入参必须做校验，禁止使用System.out.println，必须用SLF4J日志
      3. 所有SQL必须防止注入，必须使用参数化查询
      4. 生成的代码必须包含完整注释、单元测试
      """
      # 初始化Harness
      harness = SingleAgentHarness(system_rules=system_rules, max_retry=3)
      # 执行任务
      result = harness.run("生成用户管理CRUD接口，包含Controller、Service、Repository、Entity")
      print(result)

package com.harness.multiagent;

import java.util.*;

/**
* 多Agent协同分工范式 总控Harness核心实现
*/
public class MultiAgentMasterHarness {

      // 角色Agent注册中心，每个Agent有独立的Harness
      private final Map<String, AgentHarness> agentRegistry = new HashMap<>();
       private final int maxRetry = 3;

       public MultiAgentMasterHarness() {
            // 注册各角色Agent，每个Agent有独立的规则和Harness
            agentRegistry.put("PRODUCT", new AgentHarness(
               "产品经理Agent",
               """
               "你是专业产品经理，输出PRD必须包含：
               "1. 需求背景、核心目标、用户场景
               "2. 功能详情、业务规则、异常流程
               "3. 验收标准、非功能需求
               "输出必须清晰可落地，开发能直接照着做
               """"
      "));
      "agentRegistry.put("BACKEND", new AgentHarness(
               "后端开发Agent",
               """"
               "你是专业Java后端开发，基于PRD开发必须遵守：
               "1. SpringBoot 3.2规范，分层架构Controller→Service→Repository→Entity
               "2. 入参校验、异常处理、日志规范、防SQL注入
               "3. 代码必须包含完整注释、单元测试
               """"
      "));
      "agentRegistry.put("TEST", new AgentHarness(
               ""测试工程师Agent",
"""
               "你是专业测试工程师，输出测试用例必须包含：
               "1. 功能测试、边界测试、异常测试用例
               "2. 接口测试、性能测试要求
               "3. 验收标准对应的测试点
               """"
       " ));
   "}

/**
* 总控Harness核心执行入口
*/
public Map<String, String> run(String task) {
   Map<String, String> resultMap = new HashMap<>();
   int retryCount = 0;

    while (retryCount < maxRetry) {
       try {
          // 1. 产品Agent输出PRD
          AgentHarness productAgent = agentRegistry.get("PRODUCT");
          String prd = productAgent.run("基于以下需求，输出完整PRD：" + task);
         resultMap.put("PRD", prd);

         // 2. 后端Agent基于PRD开发代码
         AgentHarness backendAgent = agentRegistry.get("BACKEND");
         String code = backendAgent.run("基于以下PRD，输出完整Java后端代码：" + prd);
          resultMap.put("BACKEND_CODE", code);

         // 3. 测试Agent输出测试用例
         AgentHarness testAgent = agentRegistry.get("TEST");
         String testCase = testAgent.run("基于以下PRD，输出完整测试用例：" + prd);
         resultMap.put("TEST_CASE", testCase);

         // 4. 总控Harness最终校验
         boolean isPass = finalValidate(resultMap);
         if (isPass) {
                   resultMap.put("STATUS", "SUCCESS");
                    return resultMap;
          }
          retryCount++;
     } catch (Exception e) {
          retryCount++;
         resultMap.put("ERROR", e.getMessage());
    }
  }

  resultMap.put("STATUS", "FAIL");
   return resultMap;
   }

/**
* 总控Harness最终校验逻辑
*/
private boolean finalValidate(Map<String, String> resultMap) {
     return resultMap.containsKey("PRD")
        && resultMap.containsKey("BACKEND_CODE")
        && resultMap.containsKey("TEST_CASE");
  }

/**
* 子Agent独立Harness实现
*/
static class AgentHarness {
     private final String roleName;
      private final String systemRules;

     public AgentHarness(String roleName, String systemRules) {
        this.roleName = roleName;
        this.systemRules = systemRules;
      }

     public String run(String task) {
        // 实际实现调用大模型+独立校验闭环，参考范式一
         return "【" + roleName + "】执行结果：\n" + task;
      }
   }
}

import os
from typing import Dict, List

class ProgressiveSkillHarness:
   """渐进式技能解锁范式Harness实现"""
   def __init__(self, project_root: str = "./agent_project"):
     self.project_root = project_root
     self.nav_map_path = os.path.join(project_root, "AGENTS.md")
     self.skill_library = {} # 按阶段分类的技能库
     self.phase_progress = {} # 阶段进度持久化
     self._init_project()

  def _init_project(self):
     """初始化项目目录和导航地图"""
     os.makedirs(self.project_root, exist_ok=True)
     # 初始化导航地图，核心是阶段拆分和技能索引
     if not os.path.exists(self.nav_map_path):
        default_nav_map = """
        # AGENT 项目导航地图
        ## 项目阶段拆分
        1. 阶段1：项目初始化 - 技能：环境搭建、仓库初始化、规范定义
        2. 阶段2：架构设计 - 技能：分层架构设计、数据库设计、接口设计
        3. 阶段3：核心功能开发 - 技能：业务代码开发、单元测试
         4. 阶段4：测试与优化 - 技能：集成测试、性能优化、bug修复
        5. 阶段5：上线交付 - 技能：部署配置、上线文档、验收交付

        ## 技能索引
        所有技能文档在 ./docs/skills/ 目录下，仅当前阶段的技能会被注入
        """
        with open(self.nav_map_path, "w", encoding="utf-8") as f:
                f.write(default_nav_map)
        # 加载技能库
        self._load_skill_library()

   def _load_skill_library(self):
        """加载技能库，按阶段分类"""
        skill_dir = os.path.join(self.project_root, "docs", "skills")
        os.makedirs(skill_dir, exist_ok=True)
        # 遍历技能文件，按阶段分类
        for phase in ["phase1", "phase2", "phase3", "phase4", "phase5"]:
                phase_skill_path = os.path.join(skill_dir, phase)
                 os.makedirs(phase_skill_path, exist_ok=True)
                 self.skill_library[phase] = []
                for file in os.listdir(phase_skill_path):
                    if file.endswith(".md"):
                        with open(os.path.join(phase_skill_path, file), "r", encoding="utf-8") as f:
                         self.skill_library[phase].append(f.read())

def get_current_phase_skills(self, current_phase: str) -> str:
      """Harness核心：仅获取当前阶段的技能，渐进式披露"""
      return "\n\n".join(self.skill_library.get(current_phase, []))

def update_phase_progress(self, phase: str, progress: str, result: str):
      """持久化阶段进度到文件系统，跨会话不会丢失"""
      self.phase_progress[phase] = progress
      progress_path = os.path.join(self.project_root, "docs", "progress", f"{phase}_progress.md")
      os.makedirs(os.path.dirname(progress_path), exist_ok=True)
      with open(progress_path, "w", encoding="utf-8") as f:
            f.write(f"# {phase} 进度记录\n\n进度：{progress}\n\n成果：{result}")

def run_phase(self, phase: str, phase_task: str) -> str:
      """执行指定阶段的任务，仅注入当前阶段的技能和规则"""
      # 1. 仅注入导航地图和当前阶段的技能，避免上下文溢出
system_prompt = f"""
      项目导航地图：{open(self.nav_map_path, "r",       encoding="utf-8").read()}
      当前阶段：{phase}
      当前阶段专属技能和规则：{self.get_current_phase_skills(phase)}
      历史阶段进度：{self.phase_progress}
      你只需要完成当前阶段的任务，不要涉及后续阶段的内容
      """
      # 2. 执行任务（实际调用大模型）
      print(f"===== 执行{phase}任务 =====")
      print(f"注入上下文长度：{len(system_prompt)} 字符")
      result = f"{phase}任务执行完成，成果：{phase_task}"
      # 3. 持久化进度
      self.update_phase_progress(phase, "已完成",       result)
      return result

# 使用示例
if __name__ == "__main__":
      harness = ProgressiveSkillHarness(project_root="./user_management_project")
      # 分阶段执行，每个阶段仅注入对应的技能，上下文不会爆炸
      print(harness.run_phase("phase1", "初始化SpringBoot项目，搭建仓库结构，定义编码规范"))
      print(harness.run_phase("phase2", "设计用户管理系统的分层架构、数据库表结构、接口规范"))