小论文历程

lief2026-03-162026-03-20

[up主专用，视频内嵌代码贴在这]

小论文更新历程

2026年3月16日

任务1.1（初始化项目结构：创建所有核心文件夹和关键文件）

创建远程仓库（github上点击右上角的 + 号 → New repository）
填写仓库信息（以下是推荐填写内容）：
Owner：你的用户名（自动）
Repository name：apemad-framework（建议全小写，用 - 连接）
Description（可选）：通过自适应提示增强多智能体辩论：一种面向任务复杂度的框架以提升决策质量
Public / Private：建议选 Public（便于论文引用和开源）
Add a README file：勾选（这样仓库一开始就有 README.md）
Add .gitignore：选择模板 Python（自动生成常用的忽略规则）
Choose a license：选 MIT License（最宽松的开源许可）
把远程仓库克隆到本地电脑
在仓库首页，点击绿色的 Code 按钮 → 复制 HTTPS 链接（类似这样）：
https://github.com/你的用户名/apemad-framework.git
打开电脑的终端（Windows 用 PowerShell 或 Git Bash，Mac 用 Terminal）
→ 先决定你想把项目放在哪个文件夹，例如：
Windows：D:\Projects\

执行克隆命令（把远程仓库下载到本地）：

1	git clone https://github.com/你的用户名/apemad-framework.git

进入项目文件夹：
1
cd apemad-framework
一些cmd命令（mkdir创建文件夹，cd..返回上一级，touch创建文件）

第一次提交并推送到 GitHub
添加所有文件到暂存区：
提交本次更改：
推送到远程仓库：

1
2
3

git add .
git commit -m "初始化项目结构：创建所有核心文件夹和关键文件"
git push origin main

小插曲（提交推送问题）
出现交互问题

1 2	D:\apemad-framework>git push origin main fatal: unable to access 'https://github.com/permanent-ray/apemad-framework.git/': Failed to connect to github.com port 443: Timed out

最后一步推送出现问题，通过更改hosts文件，Windows大概在C:\Windows\System32\drivers\etc下，将查询到的GitHub IP地址内容140.82.114.3 github.com 追加进hosts文件

D:\apemad-framework>git push origin main
Enumerating objects: 5, done.
Counting objects: 100% (5/5), done.
Delta compression using up to 16 threads
Compressing objects: 100% (2/2), done.
Writing objects: 100% (4/4), 516 bytes | 516.00 KiB/s, done.
Total 4 (delta 0), reused 0 (delta 0), pack-reused 0
To https://github.com/permanent-ray/apemad-framework.git
   06495e8..311cf7f  main -> main

最后成功解决问题！

任务1.2 配置开发环境与依赖（超级详细步骤）
这个任务的目标是让你的电脑能够顺利运行 APEMAD 项目的所有代码，包括安装 Python 环境、创建虚拟环境、安装所有需要的第三方库，并验证安装成功。
我把这一步拆成 10 个极细步骤，每个步骤都包含：

具体操作命令
在哪个目录下执行
为什么要做这一步
可能遇到的常见问题及解决办法
验证方式

前提：你已经完成任务 1.1，当前终端已经在项目根目录（apemad-framework）下。

确认 Python 版本（python –version）
创建虚拟环境（强烈推荐，避免全局污染）
操作（在项目根目录 apemad-framework 下执行）：
1
2
# 创建虚拟环境，命名为 venv
python -m venv venv
激活虚拟环境
1
.\venv\Scripts\activate
激活后提示符变化：前面会出现 (venv) 或 (venv) $
为什么：后续所有 pip install 只影响这个项目，不会污染全局
当前虚拟环境已弃用，仅仅是虚拟环境弃用，其他步骤正常，因为后面python版本不匹配，更换了虚拟环境

创建 requirements.txt 文件（如果还没创建touch requirements.txt）
打开文件（用 VS Code 或记事本），写入以下内容（这是当前最推荐的依赖列表）：

# 核心框架
pyautogen>=0.2.0
langchain>=0.1.0

# LLM 与 transformers
transformers>=4.38.0
torch>=2.0.0
accelerate>=0.27.0

# 嵌入与检索
sentence-transformers>=2.2.2
faiss-cpu>=1.7.4          # 如果有 GPU，可换 faiss-gpu

# 可视化与日志
matplotlib>=3.8.0
seaborn>=0.13.0
wandb>=0.16.0

# RL 微调
stable-baselines3>=2.0.0
gymnasium>=0.29.0

# NLP 工具
spacy>=3.7.0
# 在终端运行：python -m spacy download en_core_web_sm

# 其他工具
pyyaml>=6.0.0
python-dotenv>=1.0.0
tqdm>=4.66.0

安装所有依赖
操作（在激活虚拟环境后执行）：
1
pip install -r requirements.txt

安装 spaCy 模型（用于实体依赖深度计算）
操作：

1	python -m spacy download en_core_web_sm

按回车后会看到类似输出：

1
2
3

Downloading en_core_web_sm...
✔ Download and installation successful
You can now load the package via spacy.load('en_core_web_sm')

立即验证是否成功（非常重要！）
新建一个测试文件：touch test_spacy.py
打开文件，复制粘贴以下内容并保存：

# test_spacy.py
import spacy

try:
    nlp = spacy.load("en_core_web_sm")
    print("spaCy 模型加载成功！")
    print("版本：", spacy.__version__)
    
    # 测试一句简单句子
    doc = nlp("This is a test sentence for entity recognition.")
    print("实体：", [(ent.text, ent.label_) for ent in doc.ents])
except Exception as e:
    print("加载失败！错误信息：", e)

运行测试：python test_spacy.py
预期成功输出（类似这样）：
spaCy 模型加载成功！
版本： 3.7.2
实体： []
确认成功后删除测试文件（可选）:del test_spacy.py

验证所有核心库是否可用（最全面的检查）
为什么要这一步？
确保所有依赖都正常工作，尤其是 torch（GPU支持）、transformers（LLM）、faiss（RAG）、wandb（日志）等关键库。如果这里通不过，后续模块会一直报错。
新建测试文件：touch test_imports.py
打开文件输入：

# test_imports.py
import sys

print("Python 版本:", sys.version)

try:
    import pyautogen
    print("pyautogen 导入成功！")
    # 安全检查版本（新版没有 __version__，我们跳过或用其他方式）
    # 如果你非要版本，可以用 pip show（但这里先注释掉）
    # import subprocess
    # version = subprocess.check_output(["pip", "show", "pyautogen"]).decode().split("Version: ")[1].split("\n")[0]
    # print("pyautogen 版本:", version.strip())
except ImportError as e:
    print("pyautogen 导入失败:", e)

try:
    import autogen
    print("旧版 autogen 导入成功（如果存在）")
except ImportError:
    print("旧版 autogen 未找到（正常，新版已改为 pyautogen）")

try:
    import transformers
    print("Transformers 导入成功，版本:", transformers.__version__)
except ImportError as e:
    print("Transformers 导入失败:", e)

try:
    import torch
    print("Torch 导入成功，版本:", torch.__version__)
    print("CUDA 是否可用:", torch.cuda.is_available())
    if torch.cuda.is_available():
        print("当前 GPU:", torch.cuda.get_device_name(0))
except ImportError as e:
    print("Torch 导入失败:", e)

try:
    import sentence_transformers
    print("Sentence-Transformers 导入成功")
except ImportError as e:
    print("Sentence-Transformers 导入失败:", e)

try:
    import faiss
    print("FAISS 导入成功")
except ImportError as e:
    print("FAISS 导入失败:", e)

try:
    import wandb
    print("WandB 导入成功")
except ImportError as e:
    print("WandB 导入失败:", e)

try:
    import stable_baselines3
    print("Stable-Baselines3 导入成功")
except ImportError as e:
    print("Stable-Baselines3 导入失败:", e)

try:
    import spacy
    nlp = spacy.load("en_core_web_sm")
    print("spaCy + en_core_web_sm 加载成功！")
except Exception as e:
    print("spaCy 加载失败:", e)

print("\n如果上面大部分成功（尤其是 pyautogen），则环境配置完成！")

保存文件后，再次运行：python test_imports.py
预期新输出（大致这样）：
Python 版本: 3.13.9 …
pyautogen 导入成功！
旧版 autogen 未找到（正常，新版已改为 pyautogen）
Transformers 导入成功，版本: 5.3.0
Torch 导入成功，版本: 2.10.0+cpu
CUDA 是否可用: False
Sentence-Transformers 导入成功
FAISS 导入成功
WandB 导入成功
Stable-Baselines3 导入成功
spaCy + en_core_web_sm 加载成功！

如果上面大部分成功（尤其是 pyautogen），则环境配置完成！

任务1.3创建 config/config.yaml 的详细步骤（完整版）
现在你的环境已经完全就绪（pyautogen 导入成功，其他库也正常），我们可以进入任务 1.3：创建配置文件 config/config.yaml。
这个文件的作用是把项目中所有可调参数集中管理，以后改任何设置（LLM 模型、温度、复杂度权重、阈值、RAG 路径等）都只需要改这一个文件，不用到处改代码。非常适合学术项目（方便复现、实验对比）。

写入完整 config.yaml 内容 实时更新

# config/config.yaml
# APEMAD 框架配置文件（所有超参数集中管理）

# ========================
# LLM 设置
# ========================
llm:
  provider: "huggingface"
  model_name: "D:/apemad-framework/local_models/Qwen2.5-3B-Instruct"
  temperature: 0.7
  max_tokens: 512
  top_p: 0.9

# ========================
# 任务复杂度评估参数
# ========================
complexity:
  weights:
    length: 0.25        # α - 输入长度权重
    depth: 0.35         # β - 实体依赖深度
    uncertainty: 0.30   # γ - 初步不确定性
    modality: 0.10      # δ - 多模态复杂度
  thresholds:
    low: 0.35
    medium: 0.65
    high: 1.0
  max_agents_low: 4
  max_agents_medium: 7
  max_agents_high: 12

# ========================
# 不确定性计算设置
# ========================
uncertainty:
  entropy_threshold_high: 0.35     # entropy < 此值 → 高确定
  entropy_threshold_low: 0.65      # entropy > 此值 → 低确定
  sampling_n: 3                    # 多采样次数（variance 计算）
  semantic_threshold: 0.75         # 语义一致性阈值（cosine sim）

# ========================
# 自适应提示融合设置
# ========================
fusion:
  max_depth_low: 1
  max_depth_medium: 2
  max_depth_high: 3
  min_confidence_weight: 0.6       # 低于此权重不注入

# ========================
# 外部验证层设置
# ========================
validation:
  rag:
    embedding_model: "all-MiniLM-L6-v2"
    top_k_low: 3
    top_k_high: 5
    similarity_threshold_strong: 0.75
    similarity_threshold_weak: 0.5
  verifier_llm_temperature: 0.2    # Verifier 判断时低温度，更确定

# ========================
# 监控与终止设置
# ========================
termination:
  global_uncertainty_threshold_factor: 0.3   # < θ * C 终止（θ=0.3）
  change_threshold: 0.05                     # 连续两轮变化 < 此值
  max_rounds_low: 3
  max_rounds_medium: 5
  max_rounds_high: 8

# ========================
# RL 微调设置
# ========================
rl:
  enabled: false                     # 先关闭，实验稳定后再开
  ppo_epochs: 10
  reward_weights:
    accuracy: 1.0
    consistency: 0.8
    cost: -0.3
    hallucination: -0.5
  update_frequency: 10               # 每 10 个任务优化一次

# ========================
# 路径设置（相对项目根目录）
# ========================
paths:
  logs_dir: "../logs"
  results_dir: "../experiments/results"
  rag_index_dir: "../experiments/rag_index"   # FAISS 索引存放处
  paper_figures_dir: "../paper_figures"

# ========================
# 调试与日志
# ========================
debug:
  verbose: true
  log_level: "INFO"                  # DEBUG / INFO / WARNING
  save_prompt_chain: true            # 是否保存每轮 prompt 链可视化

运行测试：python test_config.py
预期输出：
config.yaml 加载成功！
LLM 模型: meta-llama/Meta-Llama-3-8B-Instruct
复杂度权重 α: 0.4
高复杂度代理上限: 12

提交到 GitHub

1
2
3

git add config/config.yaml test_config.py
git commit -m "添加完整 config.yaml 配置文件及读取测试脚本"
git push origin main

删除临时测试文件（可选）del test_config.py

任务1.4编写 LLM 调用统一封装（llm_wrapper.py）——详细步骤
这一步的目标是创建一个统一的 LLM 调用接口，让后续所有模块（辩论池、不确定性计算、Verifier LLM 等）都通过这个接口调用 LLM，而不是到处写重复的 transformers 或 OpenAI 代码。这样做的好处是：

切换模型（本地 LLaMA → GPT-4o → Groq）只需改 config.yaml
统一处理 temperature、max_tokens、logits 返回（用于 entropy 计算）
支持 logits 输出（计算不确定性必须有 logits 或 logprobs）

在正确位置创建文件
进入 core 文件夹（所有核心模块都放在这里）
创建文件：type nul > llm_wrapper.py

编写完整 llm_wrapper.py 代码 实时更新
把下面这段完整代码复制粘贴到 llm_wrapper.py 文件中，然后保存。

# core/llm_wrapper.py
"""
统一的 LLM 调用封装，支持 HuggingFace 本地模型和 OpenAI/Groq API
支持返回 logits（用于不确定性计算）或仅文本输出
"""

import os
import yaml
from typing import Dict, Any, Optional, List, Tuple
from dotenv import load_dotenv

# 加载 .env 文件（存放 API Key）
load_dotenv()

# 读取 config.yaml
CONFIG_PATH = os.path.join(os.path.dirname(os.path.dirname(__file__)), "config", "config.yaml")
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
    CONFIG = yaml.safe_load(f)

LLM_CONFIG = CONFIG["llm"]


class LLMWrapper:
    """统一的 LLM 调用类"""

    def __init__(self):
        self.provider = LLM_CONFIG["provider"].lower()
        self.model_name = LLM_CONFIG["model_name"]
        self.temperature = LLM_CONFIG.get("temperature", 0.7)
        self.max_tokens = LLM_CONFIG.get("max_tokens", 1024)
        self.top_p = LLM_CONFIG.get("top_p", 0.9)

        if self.provider == "huggingface":
            self._init_hf()
        elif self.provider in ["openai", "groq"]:
            self._init_api()
        else:
            raise ValueError(f"不支持的 LLM provider: {self.provider}")

    def _init_hf(self):
        """初始化 HuggingFace 本地模型"""
        from transformers import AutoTokenizer, AutoModelForCausalLM
        import torch

        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        print(f"加载 HuggingFace 模型: {self.model_name} on {self.device}")

        self.tokenizer = AutoTokenizer.from_pretrained(self.model_name)
        self.model = AutoModelForCausalLM.from_pretrained(
            self.model_name,
            torch_dtype=torch.float16 if self.device == "cuda" else torch.float32,
            device_map="auto",
            trust_remote_code=True,
        )
        self.model.eval()

    def _init_api(self):
        """初始化 OpenAI 或 Groq API"""
        from openai import OpenAI

        api_key = os.getenv("OPENAI_API_KEY") if self.provider == "openai" else os.getenv("GROQ_API_KEY")
        if not api_key:
            raise ValueError(f"请在 .env 中设置 {self.provider.upper()}_API_KEY")

        base_url = "https://api.groq.com/openai/v1" if self.provider == "groq" else None
        self.client = OpenAI(api_key=api_key, base_url=base_url)

    def generate(
        self,
        prompt: str,
        system_prompt: Optional[str] = None,
        return_logits: bool = False,
        **kwargs
    ) -> Dict[str, Any]:
        """
        统一生成接口
        Args:
            prompt: 用户输入 prompt
            system_prompt: 系统提示（可选）
            return_logits: 是否返回 logits（仅 HuggingFace 支持）
        Returns:
            dict: {"text": 生成文本, "logits": logits (可选), "usage": token 消耗}
        """
        temperature = kwargs.get("temperature", self.temperature)
        max_tokens = kwargs.get("max_tokens", self.max_tokens)
        top_p = kwargs.get("top_p", self.top_p)

        if self.provider == "huggingface":
            return self._generate_hf(prompt, system_prompt, return_logits, temperature, max_tokens, top_p)
        else:
            return self._generate_api(prompt, system_prompt, temperature, max_tokens, top_p)

    def _generate_hf(
        self,
        prompt: str,
        system_prompt: Optional[str],
        return_logits: bool,
        temperature: float,
        max_tokens: int,
        top_p: float
    ) -> Dict:
        """HuggingFace 本地生成"""
        from transformers import GenerationConfig
        import torch

        full_prompt = f"{system_prompt}\n\n{prompt}" if system_prompt else prompt

        inputs = self.tokenizer(full_prompt, return_tensors="pt").to(self.device)

        generation_config = GenerationConfig(
            temperature=temperature,
            top_p=top_p,
            max_new_tokens=max_tokens,
            do_sample=True if temperature > 0 else False,
            pad_token_id=self.tokenizer.eos_token_id,
            output_scores=return_logits,
            return_dict_in_generate=True,
        )

        with torch.no_grad():
            outputs = self.model.generate(**inputs, generation_config=generation_config)

        generated_ids = outputs.sequences[0][inputs.input_ids.shape[1]:]
        text = self.tokenizer.decode(generated_ids, skip_special_tokens=True)

        result = {"text": text.strip(), "usage": {"prompt_tokens": inputs.input_ids.shape[1], "completion_tokens": len(generated_ids)}}

        if return_logits and hasattr(outputs, "scores"):
            result["logits"] = outputs.scores  # tuple of tensors

        return result

    def _generate_api(
        self,
        prompt: str,
        system_prompt: Optional[str],
        temperature: float,
        max_tokens: int,
        top_p: float
    ) -> Dict:
        """OpenAI/Groq API 生成（不支持 logits）"""
        messages = []
        if system_prompt:
            messages.append({"role": "system", "content": system_prompt})
        messages.append({"role": "user", "content": prompt})

        response = self.client.chat.completions.create(
            model=self.model_name,
            messages=messages,
            temperature=temperature,
            max_tokens=max_tokens,
            top_p=top_p,
        )

        text = response.choices[0].message.content.strip()
        usage = response.usage.dict() if hasattr(response, "usage") else {}

        return {"text": text, "usage": usage}

    def get_logits(self, prompt: str) -> Optional[List]:
        """便捷方法：只返回 logits（用于不确定性计算）"""
        result = self.generate(prompt, return_logits=True)
        return result.get("logits", None)

测试 llm_wrapper.py 是否可用
在项目根目录新建测试文件：touch test_llm_wrapper.py
写入以下代码：

# test_llm_wrapper.py
from core.llm_wrapper import LLMWrapper

llm = LLMWrapper()

# 测试文本生成
response = llm.generate(
    prompt="你好，请用一句话介绍自己。",
    system_prompt="你是一个友好的AI助手。"
)

print("生成文本：", response["text"])
print("Token 使用：", response["usage"])

# 如果是 HuggingFace 模型，测试 logits
if llm.provider == "huggingface":
    logits = llm.get_logits("你好")
    print("Logits 类型：", type(logits) if logits else "None")

运行测试：python test_llm_wrapper.py
预期输出（HuggingFace 示例）：

1
2
3

生成文本： 你好！我是 APEMAD 框架中的智能助手，专注于提升多智能体决策质量。
Token 使用： {'prompt_tokens': xx, 'completion_tokens': xx}
Logits 类型： <class 'tuple'>

提交到 GitHub

1
2
3

git add core/llm_wrapper.py test_llm_wrapper.py
git commit -m "实现 LLM 统一调用封装模块 llm_wrapper.py 并添加测试脚本"
git push origin main

2026年3月17日至19日

tips：因为环境问题有重新配置了，换了一个适当的模型。

降级到 Python 3.11

新建一个 Python 3.11 环境（用 conda，干净隔离）：
打开 Anaconda Prompt（以管理员身份运行）
搜索 “Anaconda Prompt”，右键 → 以管理员身份运行
1
2
3
conda deactivate # 先退出当前 venv
conda create -n apemad_py311 python=3.11 -y
conda activate apemad_py311
进入项目目录：cd D:\apemad-framework
重新安装依赖：pip install -r requirements.txt
安装 spaCy 模型：python -m spacy download en_core_web_sm
安装GPU版torch：pip install torch==2.6.0 torchvision==0.21.0 torchaudio==2.6.0 –index-url https://download.pytorch.org/whl/cu118
重新运行测试：python test_imports.py python test_llm_wrapper.py

“方案 2：手动下载模型文件 + 离线加载（网络不稳时最保险）” tips:也可以选择在线模型

首选：Qwen/Qwen2.5-3B-Instruct（约 6GB，6GB 显存刚好能跑），下载页面：https://huggingface.co/Qwen/Qwen2.5-3B-Instruct/tree/main

手动下载所有文件
创建本地存放目录（推荐放在项目里，便于管理）：mkdir D:\apemad-framework\local_models\Qwen2.5-3B-Instruct
选择全部文件下载，保存到：D:\apemad-framework\local_models\Qwen2.5-3B-Instruct
修改 config.yaml 使用本地路径
打开 config/config.yaml，把 llm.model_name 改为本地绝对路径

llm:
  provider: "huggingface"
  model_name: "D:/apemad-framework/local_models/Qwen2.5-3B-Instruct"   # ← 改成这个
  temperature: 0.7
  max_tokens: 512   # 先调小，避免 OOM
  top_p: 0.9

修改 llm_wrapper.py 支持本地路径加载

# core/llm_wrapper.py
"""
统一的 LLM 调用封装，支持 HuggingFace 本地模型（含 4bit 量化）和 OpenAI/Groq API
支持返回 logits（用于不确定性计算）或仅文本输出
适配 6GB 显存电脑，使用 BitsAndBytes 4bit 量化
"""

import os
import yaml
from typing import Dict, Any, Optional
from dotenv import load_dotenv

# 加载 .env 文件（存放 API Key 或 HF_TOKEN）
load_dotenv()

# 读取 config.yaml（假设在项目根目录/config下）
CONFIG_PATH = os.path.join(os.path.dirname(os.path.dirname(__file__)), "config", "config.yaml")
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
    CONFIG = yaml.safe_load(f)

LLM_CONFIG = CONFIG["llm"]


class LLMWrapper:
    """统一的 LLM 调用类"""

    def __init__(self):
        self.provider = LLM_CONFIG["provider"].lower()
        self.model_name = LLM_CONFIG["model_name"]
        self.temperature = LLM_CONFIG.get("temperature", 0.7)
        self.max_tokens = LLM_CONFIG.get("max_tokens", 1024)
        self.top_p = LLM_CONFIG.get("top_p", 0.9)

        if self.provider == "huggingface":
            self._init_hf()
        elif self.provider in ["openai", "groq"]:
            self._init_api()
        else:
            raise ValueError(f"不支持的 LLM provider: {self.provider}")

    def _init_hf(self):
        from transformers import AutoTokenizer, AutoModelForCausalLM
        import torch
        from pathlib import Path

        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        print(f"加载 HuggingFace 模型: {self.model_name} on {self.device}")

        local_path = Path(self.model_name)
        load_from = str(local_path) if local_path.is_dir() else self.model_name

        self.tokenizer = AutoTokenizer.from_pretrained(load_from, trust_remote_code=True)

        self.model = AutoModelForCausalLM.from_pretrained(
            load_from,
            device_map="cuda:0",  # 强制全放 GPU 0
            trust_remote_code=True,
            torch_dtype=torch.bfloat16,  # bfloat16 更省显存
            low_cpu_mem_usage=True,
            offload_folder=None,  # 禁用 offload
            offload_state_dict=False,  # 禁用状态卸载
        )
        self.model.eval()
        print("模型加载完成，全 GPU 模式")

    def _init_api(self):
        """初始化 OpenAI 或 Groq API（不支持 logits）"""
        from openai import OpenAI

        api_key_env = "OPENAI_API_KEY" if self.provider == "openai" else "GROQ_API_KEY"
        api_key = os.getenv(api_key_env)
        if not api_key:
            raise ValueError(f"请在 .env 中设置 {api_key_env}")

        base_url = "https://api.groq.com/openai/v1" if self.provider == "groq" else None
        self.client = OpenAI(api_key=api_key, base_url=base_url)

    def generate(
        self,
        prompt: str,
        system_prompt: Optional[str] = None,
        return_logits: bool = False,
        **kwargs
    ) -> Dict[str, Any]:
        """
        统一生成接口
        Args:
            prompt: 用户输入 prompt
            system_prompt: 系统提示（可选）
            return_logits: 是否返回 logits（仅 HuggingFace 支持）
        Returns:
            dict: {"text": 生成文本, "logits": logits (可选), "usage": token 消耗}
        """
        temperature = kwargs.get("temperature", self.temperature)
        max_tokens = kwargs.get("max_tokens", self.max_tokens)
        top_p = kwargs.get("top_p", self.top_p)

        if self.provider == "huggingface":
            return self._generate_hf(prompt, system_prompt, return_logits, temperature, max_tokens, top_p)
        else:
            return self._generate_api(prompt, system_prompt, temperature, max_tokens, top_p)

    def _generate_hf(
        self,
        prompt: str,
        system_prompt: Optional[str],
        return_logits: bool,
        temperature: float,
        max_tokens: int,
        top_p: float
    ) -> Dict:
        """HuggingFace 本地生成（支持 logits 输出）"""
        from transformers import GenerationConfig
        import torch

        # 拼接完整 prompt
        full_prompt = f"{system_prompt}\n\n{prompt}" if system_prompt else prompt

        inputs = self.tokenizer(full_prompt, return_tensors="pt").to(self.device)

        generation_config = GenerationConfig(
            temperature=temperature,
            top_p=top_p,
            max_new_tokens=max_tokens,
            do_sample=True if temperature > 0 else False,
            pad_token_id=self.tokenizer.eos_token_id,
            output_scores=return_logits,                # 返回 logits
            return_dict_in_generate=True,
        )

        with torch.no_grad():
            outputs = self.model.generate(**inputs, generation_config=generation_config)

        generated_ids = outputs.sequences[0][inputs.input_ids.shape[1]:]
        text = self.tokenizer.decode(generated_ids, skip_special_tokens=True).strip()

        result = {
            "text": text,
            "usage": {
                "prompt_tokens": inputs.input_ids.shape[1],
                "completion_tokens": len(generated_ids)
            }
        }

        if return_logits and hasattr(outputs, "scores"):
            result["logits"] = outputs.scores  # tuple of tensors

        return result

    def _generate_api(
        self,
        prompt: str,
        system_prompt: Optional[str],
        temperature: float,
        max_tokens: int,
        top_p: float
    ) -> Dict:
        """OpenAI/Groq API 生成（不支持 logits）"""
        messages = []
        if system_prompt:
            messages.append({"role": "system", "content": system_prompt})
        messages.append({"role": "user", "content": prompt})

        response = self.client.chat.completions.create(
            model=self.model_name,
            messages=messages,
            temperature=temperature,
            max_tokens=max_tokens,
            top_p=top_p,
        )

        text = response.choices[0].message.content.strip()
        usage = response.usage.dict() if hasattr(response, "usage") else {}

        return {"text": text, "usage": usage}

    def get_logits(self, prompt: str) -> Optional[list]:
        """便捷方法：只返回 logits（用于不确定性计算）"""
        result = self.generate(prompt, return_logits=True)
        return result.get("logits", None)


# 测试入口（运行本文件时可直接测试）
if __name__ == "__main__":
    llm = LLMWrapper()
    response = llm.generate(
        prompt="你好，请用一句话介绍 Qwen2.5 模型。",
        system_prompt="你是一个友好的AI助手。"
    )
    print("生成文本：", response["text"])
    print("Token 使用：", response["usage"])

运行测试:python test_llm_wrapper.py

任务 2.1：任务复杂度评估模块（complexity_assessor.py）——这是框架的“入口大脑”，决定一切自适应行为。

进入 core 文件夹,创建文件：type nul > complexity_assessor.py # Windows

# core/complexity_assessor.py
"""
任务复杂度评估模块 - 优化版
输入：任务描述文本 + 可选多模态文件
输出：复杂度分数 C (0-1) + 等级 + 推荐代理数量
"""

import yaml
import os
import spacy
from typing import Optional, Tuple
from sentence_transformers import SentenceTransformer
import numpy as np

# 读取 config.yaml
CONFIG_PATH = os.path.join(os.path.dirname(os.path.dirname(__file__)), "config", "config.yaml")
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
    CONFIG = yaml.safe_load(f)

COMPLEXITY_CONFIG = CONFIG["complexity"]
WEIGHTS = COMPLEXITY_CONFIG["weights"]
THRESHOLDS = COMPLEXITY_CONFIG["thresholds"]
MAX_AGENTS = {
    "low": COMPLEXITY_CONFIG["max_agents_low"],
    "medium": COMPLEXITY_CONFIG["max_agents_medium"],
    "high": COMPLEXITY_CONFIG["max_agents_high"]
}

# 全局加载 spaCy
NLP = spacy.load("en_core_web_sm")

# 全局加载 sentence-transformers（用于 U 计算）
SEMANTIC_MODEL = SentenceTransformer('all-MiniLM-L6-v2')


class TaskComplexityAssessor:
    """任务复杂度评估器 - 优化版"""

    def __init__(self):
        self.max_length = 4096  # 最大 token 长度参考

    def compute_complexity(self, task_text: str, images: Optional[list] = None, tables: Optional[list] = None) -> Tuple[
        float, str, int]:
        """
        计算任务复杂度分数 C - 优化版
        """
        # 1. 输入长度归一化 (L) - 更敏感（短文本也容易复杂）
        tokens_approx = len(task_text.split()) + len(task_text) / 5  # 结合字符数
        L = min(tokens_approx / 300, 1.0)  # 分母调低，300 tokens 就接近 1

        # 2. 实体关系依赖深度 (D) - 优化版
        doc = NLP(task_text)
        entities = len(doc.ents)
        if len(doc) > 0:
            depths = []
            for token in doc:
                depth = 0
                current = token
                visited = set()
                while current.head != current and current.i not in visited:
                    depth += 1
                    visited.add(current.i)
                    current = current.head
                depths.append(depth)
            avg_depth = np.mean(depths) if depths else 0.0
            entity_density = entities / (len(doc) + 1)
            D = min(avg_depth / 5 + entity_density * 3, 1.0)  # 除数调低，实体密度 *3
        else:
            D = 0.0

        # 3. 初步不确定性 (U) - 语义多样性优化
        U = 0.0
        if len(task_text) > 20:
            try:
                sentences = [s.strip() for s in task_text.split('.') if s.strip()]
                if len(sentences) >= 2:
                    embeddings = SEMANTIC_MODEL.encode(sentences)
                    sims = []
                    for i in range(len(embeddings)):
                        for j in range(i + 1, len(embeddings)):
                            sim = np.dot(embeddings[i], embeddings[j]) / (
                                        np.linalg.norm(embeddings[i]) * np.linalg.norm(embeddings[j]) + 1e-8)
                            sims.append(sim)
                    avg_sim = np.mean(sims) if sims else 1.0
                    U = 1.0 - avg_sim  # 相似度越低，不确定性越高
                    U = U ** 1.5  # 指数放大，让差异更明显
                else:
                    U = 0.7  # 句子少，默认较高不确定性
            except Exception as e:
                print(f"语义不确定性计算失败: {e}, 默认 U=0.7")
                U = 0.7
        else:
            U = 0.0

        # 4. 多模态复杂度 (M)
        M = 0.0
        if images:
            M += len(images) * 0.6
        if tables:
            M += len(tables) * 0.6
        M = min(M, 1.0)

        # 加权计算 C
        C = (
                WEIGHTS["length"] * L +
                WEIGHTS["depth"] * D +
                WEIGHTS["uncertainty"] * U +
                WEIGHTS["modality"] * M
        )
        C = min(max(C, 0.0), 1.0)

        # 分类等级
        if C < THRESHOLDS["low"]:
            level = "low"
        elif C < THRESHOLDS["medium"]:
            level = "medium"
        else:
            level = "high"

        num_agents = MAX_AGENTS[level]

        print(f"复杂度评估结果 - C: {C:.2f}, 等级: {level}, 推荐代理数: {num_agents}")
        return C, level, num_agents


# 测试入口
if __name__ == "__main__":
    assessor = TaskComplexityAssessor()

    # 测试 1: 简单任务
    simple_task = "计算 2 + 2 = ?"
    c1, level1, agents1 = assessor.compute_complexity(simple_task)
    print(f"简单任务: C={c1:.2f}, {level1}, {agents1} 代理\n")

    # 测试 2: 中等复杂度任务
    medium_task = "写一篇 300 字的自我介绍，包括兴趣爱好和职业规划。"
    c2, level2, agents2 = assessor.compute_complexity(medium_task)
    print(f"中等任务: C={c2:.2f}, {level2}, {agents2} 代理\n")

    # 测试 3: 高复杂度任务（医疗诊断示例）
    complex_task = """
    患者35岁男性，症状：持续高热（39°C）、干咳、淋巴结肿大、疲劳、近期接触野生动物。
    化验单图像显示：IgM阳性、白细胞升高。诊断可能疾病，并评估概率和风险。
    """
    c3, level3, agents3 = assessor.compute_complexity(complex_task, images=["lab.jpg", "chart.png"], tables=["data.csv"])
    print(f"高复杂度任务: C={c3:.2f}, {level3}, {agents3} 代理")

任务 2.2：多智能体辩论池
任务目标

创建 core/debate_pool.py
实现动态代理创建函数 create_debate_pool(complexity_level, num_agents=None)
支持角色模板（从 config 或硬编码）
集成 LLMWrapper（调用 LLM 生成响应）
支持共享历史更新
测试：能创建 4/7/12 个代理，并模拟一轮发言

打开终端，确保在项目根目录,进入code，创建文件：type nul > debate_pool.py

# core/debate_pool.py
"""
多智能体辩论池模块 - 最终修复版
完全绕过 AutoGen 的 llm_config 和 OpenAI 校验
使用自定义 reply_func + 手动循环实现辩论
支持动态代理创建、角色分配、辩论运行
"""

import yaml
import os
from typing import List, Dict, Optional
from autogen.agentchat import AssistantAgent, UserProxyAgent
from autogen import GroupChat
from core.llm_wrapper import LLMWrapper

# 读取 config.yaml
CONFIG_PATH = os.path.join(os.path.dirname(os.path.dirname(__file__)), "config", "config.yaml")
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
    CONFIG = yaml.safe_load(f)

COMPLEXITY_CONFIG = CONFIG["complexity"]
MAX_AGENTS = {
    "low": COMPLEXITY_CONFIG["max_agents_low"],
    "medium": COMPLEXITY_CONFIG["max_agents_medium"],
    "high": COMPLEXITY_CONFIG["max_agents_high"]
}

# 角色模板
ROLE_TEMPLATES = {
    "supporter": "你是支持者（Supporter），用逻辑和证据支持主要观点，语气积极、建设性。只回答当前任务，不要跑题或扩展无关内容。",
    "critic": "你是批评者（Critic），从相反角度质疑观点，指出逻辑漏洞和风险，语气理性但尖锐。只回答当前任务，不要跑题或扩展无关内容。",
    "neutral_analyzer": "你是中立分析者（Neutral Analyzer），客观总结双方观点，提供平衡分析。只回答当前任务，不要跑题或扩展无关内容。",
    "summarizer": "你是总结者（Summarizer），在每轮结束时提炼共识和分歧，语言简洁。只回答当前任务，不要跑题或扩展无关内容。",
    "verifier": "你是验证者（Verifier），检查事实准确性、证据可靠性，指出潜在幻觉。只回答当前任务，不要跑题或扩展无关内容。",
    "domain_expert": "你是领域专家（Domain Expert），提供专业知识和背景，针对任务领域。只回答当前任务，不要跑题或扩展无关内容。"
}


class DebatePool:
    """多智能体辩论池管理器"""

    def __init__(self, complexity_level: str = "medium", custom_num_agents: Optional[int] = None):
        self.complexity_level = complexity_level
        self.num_agents = custom_num_agents or MAX_AGENTS.get(complexity_level, 7)
        self.agents = []
        self.llm = LLMWrapper()  # 统一本地 LLM
        self.conversation_history = []  # 共享历史
        self.current_speaker_index = 0  # 当前发言代理索引

        self._create_agents()

    def _create_agents(self):
        """创建代理 - 不使用 AutoGen llm_config"""
        roles = ["supporter", "critic", "neutral_analyzer", "summarizer"]
        if self.complexity_level == "high":
            roles += ["verifier", "domain_expert"] * (self.num_agents // 6 + 1)
        roles = roles[:self.num_agents]

        for i, role in enumerate(roles):
            name = f"{role.capitalize()}Agent_{i+1}"
            system_prompt = ROLE_TEMPLATES.get(role, ROLE_TEMPLATES["neutral_analyzer"])
            system_prompt += f"\n你是 {name}，请基于共享历史和任务进行辩论。"

            # 使用 AssistantAgent，但不传 llm_config（避免校验）
            agent = AssistantAgent(
                name=name,
                system_message=system_prompt,
                llm_config=None,  # 关键：不传 config
                human_input_mode="NEVER",
            )
            self.agents.append(agent)

        print(f"创建 {len(self.agents)} 个代理，复杂度等级: {self.complexity_level}")

    def _generate_reply(self, agent, messages):
        """代理生成回复 - 直接调用本地 LLMWrapper"""
        system_prompt = agent.system_message  # 原始角色提示
        # 每次都强化角色，避免模型忘记
        reinforced_system = f"{system_prompt}\n记住你的角色：你是 {agent.name}，请严格按照你的角色风格回复，不要模仿其他代理。"

        history_str = "\n".join([f"{msg['name']}: {msg['content']}" for msg in messages[-5:]])  # 取最近5轮历史
        full_prompt = f"{reinforced_system}\n严格只回答当前任务，不要跑题。\n\n共享历史:\n{history_str}\n\n你的回复："

        response = self.llm.generate(
            prompt=full_prompt,
            system_prompt=reinforced_system,
            return_logits=False
        )
        return response["text"].strip()

    def run_debate(self, task_description: str, max_rounds: Optional[int] = None) -> List[Dict]:
        """
        手动运行辩论循环（绕过 AutoGen 内置调用）
        """
        max_rounds = max_rounds or (5 if self.complexity_level == "low" else 8 if self.complexity_level == "medium" else 12)

        # 初始消息
        self.conversation_history = [{"name": "User", "content": task_description}]

        print("辩论开始...")
        print(f"任务: {task_description}")

        for round_num in range(max_rounds):
            speaker = self.agents[self.current_speaker_index]
            reply = self._generate_reply(speaker, self.conversation_history)

            self.conversation_history.append({"name": speaker.name, "content": reply})
            print(f"{speaker.name}: {reply}")

            self.current_speaker_index = (self.current_speaker_index + 1) % len(self.agents)

            # 简单终止条件：如果最后3轮回复相似度高，可早停（可选扩展）
            if round_num >= 2:
                last3 = self.conversation_history[-3:]
                if len(set([msg["content"][:50] for msg in last3])) <= 1:
                    print("共识达成，早停")
                    break

        print("辩论结束")
        return self.conversation_history

    def get_last_response(self) -> str:
        if self.conversation_history:
            return self.conversation_history[-1]["content"]
        return "暂无响应"


# 测试入口
if __name__ == "__main__":
    # 测试低复杂度
    pool_low = DebatePool(complexity_level="low")
    history_low = pool_low.run_debate("计算 2 + 2 = ?")
    print("\n低复杂度最后一轮响应：", pool_low.get_last_response())

    # 测试高复杂度
    pool_high = DebatePool(complexity_level="high")
    history_high = pool_high.run_debate(
        "患者35岁男性，症状：高热、干咳、淋巴结肿大、疲劳。近期接触野生动物。诊断可能疾病，并评估风险。"
    )
    print("\n高复杂度最后一轮响应：", pool_high.get_last_response())

运行：python core/debate_pool.py

2026年3月20日

完善辩论池（加任务输入、融合逻辑）