博客

🇬🇧 English Version

多 Agent 系统的信任危机

你有没有遇到过这种情况：让一个 LLM 审核另一个 LLM 的输出？

假设你搭建了一个三角色文章生成管道：Planner 负责规划结构，Specialist 负责撰写内容，Evaluator 负责审核质量。听起来很完美，对吧？但当你仔细审视 Evaluator 的工作时，会发现一个根本性的矛盾——Evaluator 本身也是一个 LLM。

它审核的是另一个 LLM 写的文章，就像让嫌疑人自己评判自己的证词。当 Specialist 编造了一个不存在的类名，或者引用了一个不存在的文件路径 /nonexistent/path.py 时，Evaluator 很可能"批准"这些内容。毕竟，对 LLM 来说，编造和事实之间的界限本就模糊。

这就是多 Agent 协作中的信任危机：我们依赖 LLM 来验证 LLM 的输出，但 LLM 本身就会幻觉。

crewai-pse 这个项目的核心价值，不在于它是"又一个 CrewAI 应用"，而在于它在架构层面做出了几个反直觉的设计决策——用程序化验证替代 LLM 自我评估、用独立修复循环避免 CrewAI 的开销、用沙箱限制防止越权访问。这些决策体现了对多 Agent 系统可靠性的深刻思考。

设计决策一：用程序化验证取代 LLM 自我评估

直觉做法

让 Evaluator Agent 用 grep 检查代码引用是否存在。毕竟，Evaluator 有 run_bash 工具，可以执行 shell 命令。

实际做法

在 run.py 中，作者实现了一个独立的 _verify_article() 函数，用正则表达式提取代码引用，然后用文件系统 grep 验证它们：

def _verify_article(article: str, source_dir: Path) -> tuple[list[str], list[str]]:
    """程序化验证：grep 检查代码引用是否存在。返回 (虚构列表, 正确列表)。"""
    refs = set(re.findall(r"`([A-Za-z_][\w._]*(?:/[A-Za-z_][\w._]*)*)`", article))
    for match in re.finditer(r"```(?:python)?\s*\n(.*?)```", article, re.DOTALL):
        for line in match.group(1).split("\n"):
            m = re.match(r"^\s*(?:def|class)\s+(\w+)", line)
            if m:
                refs.add(m.group(1))

这个函数做了两件关键事情：

1. 提取所有反引号包裹的代码标识符：包括文件路径、函数名、类名
2. 从代码块中提取 def 和 class 声明：确保代码示例中的关键标识符也被验证

然后，它对每个引用进行验证：

    for ref in sorted(refs):
        if len(ref) < 3 or ref.startswith("http"):
            continue
        if ref in PYTHON_KEYWORDS:
            continue
        # 文件路径 → 递归搜磁盘（搜 .py 和 .md 文件）
        if "/" in ref or ref.endswith(".py") or ref.endswith(".md"):
            found = any(
                f.name == ref.rsplit("/", 1)[-1]
                for ext in ("*.py", "*.md")
                for f in source_dir.rglob(ext)
                if ".venv" not in str(f) and "__pycache__" not in str(f)
            )

为什么更好

这里有几个精妙的设计细节：

1. 确定性检查 vs 概率性判断

正则匹配和文件搜索是确定性的。如果 _verify_article() 说某个类名不存在，那它真的不存在。这与 LLM 的"概率性判断"形成鲜明对比——LLM 可能会"觉得"某个类名存在，但实际上只是见过类似的命名模式。

2. 区分"代码引用"和"Python 关键字"

    PYTHON_KEYWORDS = {
        "def", "class", "import", "from", "return", "yield", "raise",
        "try", "except", "finally", "with", "as", "if", "elif", "else",
        "for", "while", "break", "continue", "pass", "lambda", "and",
        "or", "not", "in", "is", "True", "False", "None", "self", "cls",
        # ... 更多关键字和内置函数
    }

如果不排除这些关键字，文章中出现 def 或 class 这样的词会被误报为虚构代码引用。

3. 递归搜索排除虚拟环境

if ".venv" not in str(f) and "__pycache__" not in str(f)

这是非常实用的细节。虚拟环境和缓存目录不应该被纳入验证范围，否则会产生大量误报。

设计决策二：独立的修复循环，绕过 CrewAI 编排

直觉做法

当 Evaluator 发现问题后，Planner 重新规划，Specialist 重写。让 CrewAI 的编排引擎处理整个修复流程。

实际做法

在 run.py 的主循环中，作者用独立的 OpenAI 客户端直接调用 LLM 进行修复，完全不经过 Crew 编排：

    # 程序化验证 + 自动修正
    max_retries = 3
    for attempt in range(1, max_retries + 1):
        fictitious, verified = _verify_article(article, source_dir)
        print(f"\n{'='*60}")
        print(f"  核查 (第{attempt}次) — 已验证 {len(verified)} 项")
        if fictitious:
            # 分离代码引用和夸大词
            code_refs = [f for f in fictitious if not f.startswith("[夸大]")]
            exaggerations = [f for f in fictitious if f.startswith("[夸大]")]

            print(f"  ❌ 虚构内容 {len(fictitious)} 项: {', '.join(fictitious)}")
            if attempt < max_retries:
                print("  🔄 自动修正中...")
                fix_parts = []
                if code_refs:
                    fix_parts.append(f"**虚构代码引用（在源码中不存在，必须删除）**: {', '.join(code_refs)}")
                if exaggerations:
                    exagg_words = [f.split("—")[0].replace("[夸大]", "").strip() for f in exaggerations]
                    fix_parts.append(f"**禁止使用的夸大词汇（必须从文章中彻底删除这些词）**: {', '.join(exagg_words)}")

                fix_prompt = f"""以下文章被核查发现问题，请修正。

{'\n'.join(fix_parts)}

**规则**:
1. 虚构代码引用：删除包含该引用的句子或代码示例，不要创造性替换
2. 夸大词汇：删除包含该词汇的整句话，不要尝试改写
3. 保持文章其余部分不变
4. 输出修正后的完整文章（从 Front Matter 开始），不输出解释

## 当前文章
{article}"""
                try:
                    resp = fix_client.chat.completions.create(
                        model=fix_model,
                        messages=[{"role": "user", "content": fix_prompt}],
                        max_tokens=8192,
                        temperature=0.7,
                    )
                    article = resp.choices[0].message.content

为什么更好

1. 性能优势

每次修复都重新运行完整的三 Agent 管道（Planner → Specialist → Evaluator）是非常昂贵的。直接调用 LLM API 进行修复，只需要一次 API 调用，而不是三次。

2. 精确控制

修复提示可以直接指定"删除虚构内容"而非"创造性替换"。这避免了 LLM 在修复过程中引入新的幻觉。

3. 程序化兜底

    EXAGGERATED_TERMS = {
        # 夸大词汇已按要求删除
    }


def _strip_exaggerated(text: str) -> str:
    """程序化删除包含夸大词汇的句子（按句号/换行分割）。"""
    for keyword in EXAGGERATED_TERMS:
        # 按句子边界（中文句号、换行、分号）逐句清理
        lines = text.split("\n")
        cleaned = []
        for line in lines:
            if keyword in line:
                # 尝试只删除包含关键词的子句（按中文标点分割）
                parts = re.split(r'([。；;])', line)
                filtered = []
                for i in range(0, len(parts) - 1, 2):
                    sentence = parts[i]
                    punct = parts[i + 1] if i + 1 < len(parts) else ""
                    if keyword not in sentence:
                        filtered.append(sentence + punct)
                # 处理最后一段（无标点结尾）
                if len(parts) % 2 == 1 and parts[-1]:
                    if keyword not in parts[-1]:
                        filtered.append(parts[-1])
                cleaned_line = "".join(filtered).strip()
                if cleaned_line:
                    cleaned.append(cleaned_line)
            else:
                cleaned.append(line)
        text = "\n".join(cleaned)
    return text

当 LLM 修复也失败时（最多 3 轮），程序化兜底函数 _strip_exaggerated() 会直接删除包含夸大词汇的句子。这是最后一道防线，确保输出不会出现未经证实的功能描述。

设计决策三：沙箱化的文件访问

直觉做法

给 Agent unrestricted 的文件系统权限，让它们可以自由读取任何文件。

实际做法

在 tools.py 中，read_file 工具强制路径必须在 PSE_ROOT 之下：

# 项目根目录，限制文件访问范围
_PROJECT_ROOT = Path(os.getenv("PSE_ROOT", Path.cwd())).resolve()


@tool("read_file")
def read_file(path: str) -> str:
    """读取文件内容。参数 path 为文件路径（限定在项目目录内）。"""
    p = Path(path).resolve()
    if not str(p).startswith(str(_PROJECT_ROOT)):
        return f"[错误] 路径超出项目范围: {path}"
    if not p.exists():
        return f"[错误] 文件不存在: {path}"
    if not p.is_file():
        return f"[错误] 不是文件: {path}"
    return p.read_text(encoding="utf-8")

为什么更好

1. 安全

防止 Agent 读取项目外的敏感文件。即使 Agent 被诱导去读取 /etc/passwd 或用户的私钥，read_file 也会拒绝访问。

2. 可控

通过环境变量 PSE_ROOT 灵活配置根目录。这使得同一个工具可以在不同项目间复用。

3. 处理符号链接

_PROJECT_ROOT = Path(os.getenv("PSE_ROOT", Path.cwd())).resolve()

使用 Path.resolve() 处理符号链接，防止路径逃逸。如果 _PROJECT_ROOT 是一个符号链接，resolve() 会解析到实际路径，确保安全检查有效。

完整流水线：从源码到双语文章

现在让我们看看整个流水线是如何运作的：

源码 → Planner 提纲 → Specialist 初稿 → 验证/修复 → 中文定稿 → 英文翻译

CrewAI 阶段

在 agents.py 中，三个 Agent 被创建并组装成 Crew：

def create_crew(task: str | None = None) -> Crew:
    """创建 PSE 三角色 Crew（Sequential 流程）。"""
    return Crew(
        agents=[create_planner(task), create_specialist(task), create_evaluator(task)],
        process=Process.sequential,
        verbose=True,
    )

注意这里使用的是 Process.sequential，意味着 Agent 按顺序执行：Planner 先规划，Specialist 再写作。Evaluator 虽然被创建，但在主流程中实际上没有被使用——验证工作由程序化的 _verify_article() 函数承担。

程序化验证阶段

Specialist 输出初稿后，run.py 中的主循环开始验证：

    crew_output = crew.kickoff()

    # 从 CrewOutput 中提取 Specialist 的输出
    if crew_output.tasks_output:
        article = crew_output.tasks_output[-1].raw
    elif specialist_task.output:
        article = specialist_task.output.raw
    else:
        article = ""

然后进入验证循环，直到所有虚构内容被清除或达到最大重试次数。

翻译阶段

中文定稿后，自动翻译成英文：

    translate_prompt = (
        "Translate the following Chinese technical article to English. "
        "Keep ALL code examples, file paths, class names, and function names unchanged. "
        f"Output ONLY the translated article:\n\n{article}"
    )

翻译提示特别强调"保持所有代码示例、文件路径、类名和函数名不变"，确保技术准确性在翻译过程中不被破坏。

设计哲学总结

crewai-pse 的核心观点很明确：多 Agent 系统的可靠性不来自更多的 Agent，而来自更聪明的验证机制。

这个项目展示了当 LLM 的输出需要事实准确性保证时，程序化验证比 LLM 自评更可靠。正则表达式和文件搜索是确定性的，不会因为 LLM 的"信心水平"而产生偏差。

适用场景

任何需要"生成 + 验证"闭环的 AI 应用都可以借鉴这些设计决策：

• 代码生成：验证生成的代码是否引用了真实存在的 API
• 技术文档：确保文档中的示例和引用与实际代码库一致
• 研究报告：防止模型编造不存在的实验结果或参考文献

启示

1. 不要过度依赖 LLM 的自我评估：让 LLM 检查 LLM 的输出是不可靠的
2. 程序化验证是关键：用确定性的检查替代概率性的判断
3. 独立于编排的修复机制：绕过复杂的 Agent 流程，直接用 LLM API 进行精确修复
4. 安全沙箱必不可少：即使是对"助手"类型的 Agent，也要限制其文件系统访问权限

crewai-pse 不是要取代 CrewAI 或其他多 Agent 框架，而是展示了如何在这些框架之上添加一层可靠的验证机制。这是一种务实的方法：承认 LLM 的局限性，用工程手段弥补它们，而不是指望模型本身变得"完美"。

源码导航

快速开始

安装

pip install crewai crewai-tools python-dotenv openai

配置

1. 复制 .env.example 为 .env，填入你的 OpenAI API Key
2. 设置 PSE_ROOT 环境变量指向你的项目根目录
3. 在 tasks/project-articles/projects.json 中添加项目配置

运行

cd tasks/project-articles
python run.py <项目名>

例如：

python run.py my-project

这将自动生成该项目的中文技术文章，并附带英文翻译。

这篇文章是通过 crewai-pse 框架本身生成的——Planner 规划了结构，Specialist 撰写了内容，程序化验证确保了所有代码引用和类名的准确性。

🇬🇧 English Version

从单一检索到混合策略

在传统 RAG 系统中，检索环节通常依赖向量相似度搜索。这种方式擅长捕捉语义相似性，但当用户查询包含精确术语、专有名词或特定格式时，纯语义检索往往会失效。

你有没有遇到过这样的场景：查询"IRR 计算公式"，系统却返回了大量讨论"投资回报率概念"的文档，而真正包含公式细节的文档反而被排在后面？这是因为向量嵌入在编码时丢失了精确的关键词匹配能力。

解决这个问题需要一种更智能的检索策略——混合检索（Hybrid Retrieval），它同时利用关键词检索和语义检索的优势。

BM25 关键词检索的实现

BM25（Best Matching 25）是一种经典的关键词排序算法，它在信息检索领域被广泛使用。与简单的 TF-IDF 相比，BM25 引入了两个关键改进：

1. 词频饱和：通过参数 k1 控制词频的影响上限，避免高频词过度主导
2. 文档长度归一化：通过参数 b 平衡长短文档的偏好

class BM25:
    """BM25 关键词检索算法

    Okapi BM25 算法实现，用于关键词检索。
    """

    def __init__(
        self,
        k1: float = 1.5,
        b: float = 0.75,
        epsilon: float = 0.25,
    ):
        self.k1 = k1
        self.b = b
        self.epsilon = epsilon
        self.doc_freqs: dict[str, int] = {}
        self.doc_len: list[int] = []
        self.avgdl: float = 0
        self.doc_count: int = 0
        self.doc_term_freqs: list[dict[str, int]] = []
        self.idf: dict[str, float] = {}
        self.documents: list[Document] = []

    def _tokenize(self, text: str) -> list[str]:
        """分词"""
        text = text.lower()
        tokens = re.findall(r"\w+", text)
        return tokens

    def fit(self, documents: list[Document]):
        """训练 BM25 模型"""
        self.documents = documents
        self.doc_count = len(documents)
        self.doc_len = []
        self.doc_term_freqs = []
        self.doc_freqs = {}

        for doc in documents:
            tokens = self._tokenize(doc.page_content)
            self.doc_len.append(len(tokens))
            term_freqs = Counter(tokens)
            self.doc_term_freqs.append(dict(term_freqs))

            for term in term_freqs:
                if term not in self.doc_freqs:
                    self.doc_freqs[term] = 0
                self.doc_freqs[term] += 1

        self.avgdl = sum(self.doc_len) / self.doc_count if self.doc_count > 0 else 0
        self._calc_idf()

这个实现的核心在于 fit 方法：它遍历所有文档，统计词频（term frequency），同时计算逆文档频率（IDF）。IDF 的计算采用了 Okapi 公式的标准形式：

def _calc_idf(self):
    """计算 IDF"""
    idf_sum = 0
    negative_idfs = []

    for term, freq in self.doc_freqs.items():
        idf = math.log(self.doc_count - freq + 0.5) - math.log(freq + 0.5)
        self.idf[term] = idf
        idf_sum += idf
        if idf < 0:
            negative_idfs.append(term)

    avg_idf = idf_sum / len(self.idf) if self.idf else 0
    eps = self.epsilon * avg_idf

    for term in negative_idfs:
        self.idf[term] = eps

注意这里对负 IDF 值的处理：当某个词出现在几乎所有文档中时，其 IDF 可能为负。作者使用了 epsilon 参数作为下限，避免这些通用词对评分产生负面影响。

三种融合策略

混合检索的关键挑战在于如何融合关键词检索和语义检索的结果。HybridRetriever 类提供了三种不同的融合策略：

加权融合（Weighted Fusion）

这是最直观的方法：分别对两种检索结果进行归一化，然后按权重相加。

def _weighted_fusion(
    self,
    keyword_results: list[tuple[Document, float]],
    semantic_results: list[tuple[Document, float]],
    k: int,
    keyword_weight: float,
    semantic_weight: float,
) -> list[tuple[Document, float]]:
    """加权融合"""
    doc_scores: dict[str, tuple[Document, float]] = {}

    max_kw_score = max((score for _, score in keyword_results), default=1.0) or 1.0
    for doc, score in keyword_results:
        doc_id = self._get_doc_id(doc)
        normalized_score = score / max_kw_score
        doc_scores[doc_id] = (doc, normalized_score * keyword_weight)

    max_sem_score = max((score for _, score in semantic_results), default=1.0) or 1.0
    for doc, score in semantic_results:
        doc_id = self._get_doc_id(doc)
        normalized_score = 1 - (score / max_sem_score)
        if doc_id in doc_scores:
            existing_doc, existing_score = doc_scores[doc_id]
            doc_scores[doc_id] = (
                existing_doc,
                existing_score + normalized_score * semantic_weight,
            )
        else:
            doc_scores[doc_id] = (doc, normalized_score * semantic_weight)

    results = sorted(doc_scores.values(), key=lambda x: x[1], reverse=True)
    return results[:k]

关键点在于语义检索返回的是距离分数（越小越相似），因此需要转换为相似度分数（1 - score/max）。

倒数排名融合（RRF）

RRF 是一种不依赖于具体分数的融合方法，它只关注文档在两个列表中的排名位置。这种方法更加稳健，因为不同检索器的分数尺度可能完全不同。

def _rrf_fusion(
    self,
    keyword_results: list[tuple[Document, float]],
    semantic_results: list[tuple[Document, float]],
    k: int,
) -> list[tuple[Document, float]]:
    """Reciprocal Rank Fusion"""
    doc_scores: dict[str, tuple[Document, float]] = {}

    for rank, (doc, _) in enumerate(keyword_results):
        doc_id = self._get_doc_id(doc)
        rrf_score = 1 / (self.rrf_k + rank + 1)
        doc_scores[doc_id] = (doc, rrf_score)

    for rank, (doc, _) in enumerate(semantic_results):
        doc_id = self._get_doc_id(doc)
        rrf_score = 1 / (self.rrf_k + rank + 1)
        if doc_id in doc_scores:
            existing_doc, existing_score = doc_scores[doc_id]
            doc_scores[doc_id] = (existing_doc, existing_score + rrf_score)
        else:
            doc_scores[doc_id] = (doc, rrf_score)

    results = sorted(doc_scores.values(), key=lambda x: x[1], reverse=True)
    return results[:k]

RRF 的核心公式是 1 / (k + rank)，其中 k 是一个调节参数（默认为 60）。这个公式的特点是：

• 排名越高（rank 越小），得分越高
• 两个列表中都出现的高排名文档会得到显著更高的综合得分
• 不需要对原始分数做任何归一化处理

分数归一化融合

这是加权融合的变体，但它使用 Min-Max 归一化来处理分数分布不一致的问题。

def _score_fusion(
    self,
    keyword_results: list[tuple[Document, float]],
    semantic_results: list[tuple[Document, float]],
    k: int,
    keyword_weight: float,
    semantic_weight: float,
) -> list[tuple[Document, float]]:
    """分数归一化融合"""
    doc_scores: dict[str, tuple[Document, float]] = {}

    kw_scores = [score for _, score in keyword_results]
    kw_min = min(kw_scores) if kw_scores else 0
    kw_max = max(kw_scores) if kw_scores else 1
    kw_range = kw_max - kw_min if kw_max != kw_min else 1

    for doc, score in keyword_results:
        doc_id = self._get_doc_id(doc)
        normalized = (score - kw_min) / kw_range if kw_range > 0 else 0.5
        doc_scores[doc_id] = (doc, normalized * keyword_weight)

    sem_scores = [score for _, score in semantic_results]
    sem_min = min(sem_scores) if sem_scores else 0
    sem_max = max(sem_scores) if sem_scores else 1
    sem_range = sem_max - sem_min if sem_max != sem_min else 1

    for doc, score in semantic_results:
        doc_id = self._get_doc_id(doc)
        normalized = 1 - (score - sem_min) / sem_range if sem_range > 0 else 0.5
        if doc_id in doc_scores:
            existing_doc, existing_score = doc_scores[doc_id]
            doc_scores[doc_id] = (
                existing_doc,
                existing_score + normalized * semantic_weight,
            )
        else:
            doc_scores[doc_id] = (doc, normalized * semantic_weight)

    results = sorted(doc_scores.values(), key=lambda x: x[1], reverse=True)
    return results[:k]

这种方法的优点是对异常值不那么敏感，但缺点是如果某个检索结果中没有返回任何文档，归一化会退化为默认值 0.5。

混合 RAG 系统集成

HybridRAGSystem 类将混合检索集成到 RAG 流程中，提供了统一的接口：

class HybridRAGSystem:
    """混合 RAG 系统

    结合关键词检索和语义检索的 RAG 系统。
    """

    def __init__(
        self,
        rag_system,
        fusion_method: str = "rrf",
        keyword_weight: float = 0.3,
    ):
        self.rag = rag_system
        self.retriever = HybridRetriever(
            fusion_method=fusion_method,
            keyword_weight=keyword_weight,
        )
        self._fitted = False

    def fit(self):
        """训练混合检索模型"""
        if not self.rag.vector_store:
            raise ValueError("RAG 系统的向量存储未初始化")

        if hasattr(self.rag.vector_store, "similarity_search"):
            docs = self.rag.vector_store.similarity_search("", k=1000)
            self.retriever.fit(docs)
            self._fitted = True

    def generate_answer(
        self,
        query: str,
        k: int = 3,
        alpha: float | None = None,
        max_context_length: int = 4000,
    ):
        """生成回答"""
        results = self.retrieve(query, k, alpha)
        relevant_docs = [doc for doc, _ in results]

        if not relevant_docs:
            return "抱歉，没有找到相关的文档信息。", []

        prompt = self.rag.prompt_builder.build_qa_prompt(
            query=query,
            documents=relevant_docs,
            max_context_length=max_context_length,
        )

        answer = self.rag.llm_integration.generate(prompt)
        return answer, relevant_docs

fit 方法的设计很巧妙：它从向量存储中抽取一批文档（默认 1000 条）来训练 BM25 索引。这避免了需要对整个文档库进行全量索引的性能开销，同时保证了召回率。

源码导航

快速开始

安装依赖：

pip install langchain-llm-toolkit

使用混合检索 RAG 系统：

from langchain_llm_toolkit.hybrid_retriever import HybridRAGSystem
from langchain_llm_toolkit.rag import RAGSystem

# 初始化基础 RAG 系统
rag = RAGSystem(vector_store_type="qdrant")

# 加载文档并创建向量存储
documents = rag.load_and_process_documents(["document.pdf"])
rag.create_vector_store(documents)

# 初始化混合 RAG 系统
hybrid_rag = HybridRAGSystem(rag_system=rag, fusion_method="rrf")

# 生成回答
answer, docs = hybrid_rag.generate_answer("什么是 IRR？")
print(answer)

你也可以直接在 RAGSystem 中使用混合检索方法：

# 使用内置的混合检索
relevant_docs = rag.retrieve_hybrid(query="IRR 计算公式", k=5, bm25_weight=0.3)

这种方法默认使用 30% 的 BM25 权重和 70% 的语义检索权重，你可以根据实际需求调整 bm25_weight 参数。

🇬🇧 English Version

一、引言

Agentic Souls 是一个基于文档驱动的 AI 软件开发工作流系统，它通过定义 Planner（规划者）、Evaluator（评审官）和 Specialist（实施者）三个核心角色，利用结构化的 Markdown 文件作为状态机和控制流，实现高质量的软件交付。

二、核心设计

2.1 设计决策 1：Markdown 作为状态机与控制流

直觉会怎么做：使用数据库或复杂的 JSON 结构来管理任务状态，在代码中硬编码流程逻辑。

实际怎么做的：Agentic Souls 使用一系列结构化的 Markdown 文件（task.md, plan.md, execution.md, evidence.md, verdict.md）作为唯一的真相来源。每个文件代表工作流的一个阶段，文件的创建和更新即为状态迁移。

根据 docs/campaign.md 的定义，一个完整的 Campaign 包含以下结构：

campaign:
  id: CAMP-XXX
  name: [Campaign 名称]
  status: pending | in_progress | completed | failed

  task:           # 任务定义
    ...

  plan:           # 执行计划
    ...

  execution:      # 执行记录
    ...

  evidence:       # 验证证据
    ...

  verdict:        # 最终判决
    ...

文件按阶段逐步创建，而不是一次性生成：

为什么更好：

1. 人类可读性：开发者可以直接阅读和理解任务进展，无需解析代码逻辑。
2. 版本控制友好：Git 天然支持 Markdown 的差异对比，便于审计和回溯。
3. 解耦：AI Agent 只是文件的消费者和生产者，流程逻辑完全由文件结构定义，易于扩展新的工作流。

2.2 设计决策 2：基于证据的独立评估（Evaluator）

直觉会怎么做：Planner 声称任务完成，直接标记为通过；或者 Evaluator 依赖 Specialist 的自述。

实际怎么做的：Evaluator 被设计为一个独立的 sub_agent，它不信任任何陈述。它必须通过读取文件、运行命令（如 pytest, ruff）来获取"一手证据"，并将这些证据映射到具体的 Acceptance Criteria (AC)。

从 souls/evaluator.md 中可以看到 Evaluator 的核心约束：

规则_1:
  描述: 不信任 Planner 的陈述，自己动手验证
  原因: 独立验证，确保客观公正

规则_2:
  描述: 不发现证据不给 PASS，去读文件、跑命令
  原因: 证据驱动，避免主观判断

规则_3:
  描述: 不在职责范围内给建议，只输出判决
  原因: 保持角色边界，避免越权

实际的 evidence.md 展示了这种证据驱动的验证方式。以 campaigns/<your-campaign>/evidence.md 为例：

## Phase 1 (P0)

| AC | Status | Verification | Proof |
|----|--------|-------------|-------|
| AC-1 | PASS | 代码审查 | `SimilarityDetector(min_block_size=min_lines)` + `add_code_block` |
| AC-2 | PASS | 读取 pyproject.toml | `ai-analyze = "src.cli:main"` |
| AC-3 | PASS | grep 搜索 | `grep -r "from src\." src/` 返回 0 结果 |
| AC-4 | PASS | 代码审查 | serena_stdio_client.py 无局部 Path 导入 |
| AC-5 | PASS | 测试通过 | `assert len(h) == 64` |

Evaluator 的判决结果只能是以下四种之一：

• PASS：所有 AC 达成
• PARTIAL：核心 AC 达成，但存在非阻塞问题
• FAIL：核心 AC 未达成
• BLOCKED：存在阻塞，需要外部干预

为什么更好：

1. 客观性：消除了主观判断和"幻觉"，所有判决都基于可验证的证据。
2. 可追溯性：evidence.md 提供了完整的审计轨迹，方便排查问题。
3. 角色边界：Evaluator 被严格限制只输出判决（PASS/FAIL），不提供建议，确保了制衡。

2.3 设计决策 3：通过约束性 Soul 文档实现角色隔离

直觉会怎么做：在代码中定义不同的 Agent 类，通过参数传递角色信息。

实际怎么做的：每个角色（Planner, Specialist, Evaluator）对应一个 Markdown 文件（位于 souls/ 目录）。这些文件不仅定义了职责，更包含了严格的"行为准则"和"禁止行为"。

以 souls/planner.md 为例，Planner 的核心约束是：

规则_1:
  描述: 永远不自己写超过20行的实现代码
  原因: 保持角色边界，避免越权

规则_2:
  描述: 所有实现通过 Task 工具委托给 Specialist
  原因: 职责分离，确保执行质量

规则_3:
  描述: 完成前必须调用 Evaluator，不允许自我评判
  原因: 独立验证，确保交付质量

souls/specialist.md 定义了 Specialist 的执行规范：

规则_1:
  描述: 只做 Planner 分配的子任务，不自行扩大范围
  原因: 保持职责边界，避免越权

规则_2:
  描述: 不做整体规划，不判断整体是否完成
  原因: 这是 Planner 的职责，不是 Specialist 的职责

规则_3:
  描述: 完成后立即汇报，说清楚做了什么、产物在哪、是否需要 Planner 关注
  原因: 及时反馈，便于 Planner 决策

souls/evaluator.md 则强调了 Evaluator 的独立性：

隔离机制:
  - Evaluator 是独立的 sub_agent
  - 不继承 Planner 的判断
  - 不接受 Planner 的建议
  - 只基于自己的验证输出判决

为什么更好：

1. 灵活性：可以轻松创建新角色或修改现有角色的行为，只需编辑 Markdown 文件。
2. 显式约束：将业务逻辑中的约束（如代码行数限制、角色边界）显式化，防止 AI 越权。
3. 可组合性：不同的工作流（Workflow）可以引用不同的 Soul 组合，适应不同场景。

三、源码导航

四、与 PSE 框架的关系

Agentic Souls 与 autogen-pse、crewai-pse 共享 Planner / Specialist / Evaluator 三角色协作的设计理念，但侧重不同：

Agentic Souls 可视为 PSE 模式的"文档层抽象"——用 Markdown 文件替代代码编排，适合需要强审计、人类可读流程的场景。

五、快速开始

git clone https://github.com/erishen/agentic-souls.git
cd agentic-souls

Agentic Souls 是一个文档驱动的系统，不需要安装依赖即可使用。你可以将其集成到任何支持 Markdown 上下文加载的 Agent 框架中（如 AutoGen、CrewAI、Claude Code 等）：

1. 克隆仓库：获取 Soul 文档、工作流定义和 Campaign 示例
2. 选择工作流：从 workflows/ 目录选取适合的工作流（如 feature-development.md）
3. 加载角色：将对应的 souls/planner.md、souls/specialist.md、souls/evaluator.md 作为 Agent 的系统提示
4. 执行 Campaign：Planner 创建 task.md → Specialist 执行并追加 execution.md → Evaluator 验证并写入 evidence.md / verdict.md

典型流程：

1. 加载 Planner Soul: "请读取 souls/planner.md"
2. 选择工作流: "请按照 workflows/feature-development.md 执行"
3. 切换到 Specialist: "请读取 souls/specialist.md"
4. 切换到 Evaluator: "请读取 souls/evaluator.md"

🇬🇧 English Version

一、引言

autogen-pse 是一个基于 Microsoft AutoGen 构建的 Planner-Specialist-Evaluator（PSE） 三角色 Agent 协作框架。它通过顺序发言的 RoundRobinGroupChat 实现状态流转，配合 cycle 级别的循环控制与 token 优化策略，形成一个可重试、可追溯、可审计的自动化交付管线。

二、核心设计

2.1 设计决策 1：用 RoundRobinGroupChat 替代 Function Calling 实现 PSE 状态机

直觉会怎么做：为 Planner、Specialist、Evaluator 分别定义工具函数（如 plan(), execute(), evaluate()），通过 LLM 的 Function Calling 来选择下一步动作，手动维护状态转移图。

实际怎么做的：直接使用 AutoGen 的 RoundRobinGroupChat（顺序发言模式），配合自定义的 termination_condition 和 max_turns 控制每轮对话长度。外层再套一个 while True 的 cycle 循环，在 run_task() 中根据 _detect_outcome() 的检测结果动态修改下一轮的 current_task 文本注入到对话中。

# src/autogen_pse/orchestrator.py
def create_pse_team(
    model_client: Optional[OpenAIChatCompletionClient] = None,
    task: Optional[str] = None,
) -> RoundRobinGroupChat:
    if model_client is None:
        model_client = _create_model_client()

    planner = create_planner(model_client, task)
    specialist = create_specialist(model_client, task)
    evaluator = create_evaluator(model_client, task)

    text_term = TextMentionTermination("交付完成") | TextMentionTermination("BLOCKED")
    return RoundRobinGroupChat(
        participants=[planner, specialist, evaluator],
        termination_condition=text_term | ExternalTermination(),
        max_turns=TURNS_PER_CYCLE,
    )

为什么更好：避免了手写状态机的复杂性——LLM 天然理解"谁该说话"（RoundRobin 顺序），而 cycle 级别的控制逻辑集中在 orchestrator 中，通过改写 task 文本实现"修复重试" vs "重新规划"的分歧路径，代码更简洁且易于扩展。

2.2 设计决策 2：双轨重试与上下文控制——PARTIAL 传摘要、FAIL 清上下文

直觉会怎么做：无论 PARTIAL 还是 FAIL，都将完整的对话历史传递给下一轮循环，让 LLM 看到全部上下文。

实际怎么做的：在 _run_one_cycle() 中，messages 列表被收集用于 trace 记录，但传给下一轮的是改写后的 current_task 文本（而不是原始对话历史）。对于 PARTIAL，只传递 result.summary（判决摘要，截断至 2000 字）；对于 FAIL，传递"基于原始任务重新制定计划"的指令，清空上轮讨论细节。

# src/autogen_pse/orchestrator.py
async def _run_one_cycle(
    team: RoundRobinGroupChat,
    task: str,
    verbose: bool,
) -> CycleResult:
    tracker = TokenTracker()
    messages = []

    async for msg in team.run_stream(task=task):
        tracker.feed(msg)
        if isinstance(msg, TaskResult):
            continue
        messages.append(msg)
        if verbose:
            _print_message(msg)

    outcome, reason, summary = _detect_outcome(messages)
    chat_log = []
    for m in messages:
        if isinstance(m, TextMessage):
            chat_log.append({"source": m.source, "content": m.content})
    return CycleResult(outcome, summary, tracker.report, chat_log, reason=reason)

而在 run_task() 中，PARTIAL 和 FAIL 的处理逻辑如下：

if result.outcome == "PARTIAL":
    partial_count += 1
    if partial_count > MAX_PARTIAL_RETRIES:
        current_task = (
            f"连续 PARTIAL {partial_count} 次，已达上限。"
            f"最近原因: {result.reason}。"
            f"请评估是否仍有可交付内容，或宣布 BLOCKED。\n\n"
            f"原始任务: {task}\n\n"
            f"最近判决: {result.summary}"
        )
        continue
    current_task = (
        f"上一轮 Evaluator 判决 PARTIAL (原因: {result.reason})。"
        f"请修复以下问题后重新提交：\n\n{result.summary}"
    )
    continue

if result.outcome == "FAIL":
    fail_count += 1
    fail_rem = MAX_FAIL_RETRIES - fail_count
    if fail_rem < 0:
        if verbose:
            print(f"\n❌ 连续 FAIL {MAX_FAIL_RETRIES}+ 次，强制 BLOCKED")
        break
    current_task = (
        f"上一轮被判 FAIL (原因: {result.reason})。请基于原始任务重新制定计划。"
        f"（剩余重试次数: {fail_rem}）\n\n"
        f"原始任务: {task}\n\n"
        f"失败原因: {result.summary}"
    )
    continue

为什么更好：防止 token 爆炸——随着循环次数增加，完整对话历史的 token 量会线性增长，既浪费成本又可能超出模型上下文窗口。通过只传递结构化摘要/改写任务，每轮保持恒定 token 开销，同时确保 Evaluator 的反馈精准传递（PARTIAL 保留问题细节以便修复，FAIL 要求彻底重来）。

2.3 设计决策 3：可扩展的任务架构——注册、规则注入与后处理修正

直觉会怎么做：每个任务硬编码一套 prompt，或通过命令行参数切换不同任务的配置。

实际怎么做的：三层设计——

1. 任务注册表 _registry.json 统一管理所有任务，CLI 通过 cli.py 的子命令调度
2. prompt 加载器 prompts.py 支持按 task 参数动态加载 tasks/<name>/prompts/*.md，不存在时回退到 demo 通用 prompt，并通过 _inject_rules() 将 {INVESTMENT_RULES} / {STOP_LOSS_RULES} 占位符替换为任务私密的规则文件内容
3. 后处理管道 在 portfolio-review/run.py 中，extract_review_from_trace() 从 trace JSON 中逆向提取 Specialist 的最终结论并保存为 Markdown 文件，sanitize_review() 用规则引擎强制修正 LLM 的违规输出（如防御底仓误卖、小额持仓乱建议）

Prompt 加载与规则注入（src/autogen_pse/prompts.py）：

def _load_private_rules(task_dir: Path, name: str) -> str:
    """加载任务私密规则文件，不存在时返回空字符串。"""
    rules_path = task_dir / f"{name}_rules.md"
    if rules_path.exists():
        return rules_path.read_text(encoding="utf-8").strip()
    return ""


def _inject_rules(template: str, name: str, task_dir: Path) -> str:
    """将 {INVESTMENT_RULES} / {STOP_LOSS_RULES} 占位符替换为私密规则文件内容。"""
    rules = _load_private_rules(task_dir, name)
    if not rules:
        return template

    # 替换可能的占位符
    for placeholder in ("INVESTMENT_RULES", "STOP_LOSS_RULES"):
        key = "{" + placeholder + "}"
        template = template.replace(key, rules)
    return template


def load_prompt(name: str, task: str | None = None) -> str:
    """加载指定角色的系统提示词。"""
    if task:
        task_dir = PROMPTS_DIR / task
        prompt_path = task_dir / "prompts" / f"{name}.md"
        if prompt_path.exists():
            template = prompt_path.read_text(encoding="utf-8")
            return _inject_rules(template, name, task_dir / "prompts")

    # 回退到通用提示词
    prompt_path = PROMPTS_DIR / "demo" / "prompts" / f"{name}.md"
    if not prompt_path.exists():
        raise FileNotFoundError(f"提示词文件不存在: {prompt_path}")
    return prompt_path.read_text(encoding="utf-8")

后处理管道（tasks/portfolio-review/run.py）：

extract_review_from_trace() 从 trace JSON 中逆向提取 Specialist 的最终结论并保存为 Markdown 文件；sanitize_review() 用规则引擎强制修正 LLM 的违规输出，例如：

• 某类核心持仓的卖出建议 → 改为持有观察（除非触发止损线）
• 小额持仓的操作建议 → 移除（噪音过多，无实际意义）

后处理的核心思路是：不信任 LLM 的最终输出，用确定性规则兜底修正，确保交付物符合预设约束。

为什么更好：新增任务只需三步（建目录、写三个 prompt、写 run.py），无需改动核心引擎。私密规则文件（如止损规则）不进入 prompt 模板，保持模板通用性同时允许任务级定制。后处理修正弥补了 LLM 不可靠性的最后一环，形成"引擎通用化 + 任务私有化 + 输出规则化"的完整闭环。

三、Web Dashboard

除了 CLI 和 Makefile 入口，autogen-pse 还提供了一个基于 FastAPI + SSE 的 Web Dashboard，支持实时查看 PSE 执行过程和历史记录。

核心功能：

• 一键执行：通过 Web 界面触发任务，SSE 流式输出让 Agent 的每一步思考过程实时可见
• 执行历史：自动记录每次运行的 trace，支持按时间、判决结果筛选查看
• Trace 详情：展开每一轮的完整对话，按 Agent 维度统计 Token 消耗
• 自动 detect prepare：Web 层自动检测任务是否需要数据准备，无需手动运行 prepare

技术实现上，web_server.py 位于项目根目录，使用 FastAPI 提供 REST API + SSE 端点，前端通过 Vite + React 构建。启动方式：

make serve    # 启动 Web Dashboard (http://localhost:8080)

四、源码导航

五、快速开始

git clone https://github.com/erishen/autogen-pse.git
cd autogen-pse
cp .env.example .env          # 配置 API Key 和外部路径
uv sync                        # 安装依赖
make demo                      # 运行 quicksort demo

如需运行投资周报任务：

make review                    # 运行 PSE 投资分析
make serve                     # 启动 Web Dashboard (http://localhost:8080)

CLI 交互示例：

python cli.py list             # 列出所有可用任务
python cli.py run portfolio-review   # 运行投资分析任务
python cli.py trace -n 5       # 查看最近 5 次执行 trace