多智能体系统驱动可控文本分类：从原理到工程实践

1. 项目概述：当文本分类遇上“多智能体委员会”

最近在折腾文本分类和知识图谱构建的项目，发现一个挺有意思的痛点：传统的文本聚类或者分类方法，无论是基于规则、传统机器学习还是预训练模型微调，一旦分类体系（Taxonomy）需要调整，或者希望分类结果能带上一些特定的“风格”和“约束”，整个流程就得推倒重来，费时费力。比如，我想把一堆技术博客按“技术栈”和“受众水平”两个维度交叉分类，或者要求生成的类别标签必须包含“成本评估”和“部署复杂度”信息，传统方法就显得很笨拙。

这让我开始关注一个新兴的方向：Agentic Clustering（智能体驱动的聚类）。这个概念的核心，是把大语言模型（LLM）当作一个个具有特定“专长”和“角色”的智能体（Agent），让它们组成一个“专家委员会”，通过协作、辩论、迭代的方式，对文本进行可控、可解释的聚类和分类。这不仅仅是“用LLM做聚类”，而是构建一个多智能体系统（Multi-Agent System），来实现对分类结果的精细化控制（Controllable）。

简单来说，它想解决的是：我们不再满足于“机器告诉我这些文本大概分几类”，而是希望“指挥”机器，按照我们设定的维度、风格、约束，去生成一个结构清晰、符合业务需求的文本分类体系。这背后的驱动力，正是当前LLM Agent研究的热潮，比如Lilian Weng那篇经典的《LLM Powered Autonomous Agents》里探讨的规划、反思、协作等能力，在这里被具体应用到了文本组织这个经典任务上。

2. 核心思路拆解：从单打独斗到委员会评审

传统的文本聚类（如K-means, DBSCAN）或主题模型（如LDA），本质上是基于统计特征（词频、向量距离）的无监督学习。它们的“可控性”很差，你很难告诉算法：“请考虑一下作者的情感倾向”或者“忽略掉所有的时间状语”。而基于Prompt的单一LLM分类，虽然灵活性高，但存在输出不稳定、思维链不可控、复杂约束难以一次性满足等问题。

Agentic Clustering的基本思路，是将复杂的分类任务分解，交由多个各司其职的LLM智能体来完成。整个流程可以类比为一个项目评审会：

1. 任务分解与智能体角色定义：首先，根据你的分类目标，设计不同的“专家”角色。例如：

   • 维度分析智能体：负责从文本中提取指定的分类维度（如“技术领域”、“难度等级”、“商业价值”）。
   • 特征提取与摘要智能体：负责为每段文本生成关键特征描述或摘要，作为后续讨论的基础。
   • 聚类提议智能体：基于特征，初步提议可能的类别和归属。
   • 约束检查智能体：专门审核初步结果是否符合预设的硬性约束（如“类别不能超过10个”、“每个类别必须包含实践案例”）。
   • 仲裁与整合智能体：当不同智能体意见冲突时，进行裁决，并整合最终分类体系。

2. 多轮迭代与精炼：智能体之间不是一次性传递结果，而是进行多轮交互。例如，聚类提议智能体给出初版分类后，约束检查智能体会提出异议，维度分析智能体可能补充新的视角，仲裁智能体组织讨论并修改提案。这个过程循环进行，直到达成共识或满足终止条件。

3. 可控性的实现：可控性就体现在你对这些智能体的“角色设定”（System Prompt）和它们之间的交互规则（Orchestration）上。你想控制分类的维度？那就强化维度分析智能体。你想控制标签的表述风格？那就为生成标签的智能体设定具体的文风要求。你想确保某些文本不被分在一起？可以设计一个专门的“隔离约束”智能体。

这种方法的优势很明显：模块化、可解释、强可控。每个智能体的失败或偏见可以被隔离和调试，整个决策过程有“会议纪要”（智能体间的对话历史）可追溯，你可以通过调整智能体阵容和议事规则来精确控制输出。

3. 系统架构设计与智能体分工

要实现一个可控的文本分类多智能体系统，我们需要设计一个清晰的架构。这里我结合实践，分享一个可落地的四层架构设计。

3.1 控制层：任务规划与流程编排

这是系统的大脑，通常由一个管理器智能体（Manager Agent）或一个固定的编排脚本（Orchestrator）担任。它的核心职责是：

• 解析用户指令：将“构建一个关注安全性和性能的微服务架构文章分类”这样的自然语言指令，分解为具体的智能体任务序列。
• 初始化与调度：根据任务序列，实例化所需的各个智能体，并控制它们的执行顺序。是串行、并行还是基于条件的触发？
• 管理对话上下文：维护一个共享的“工作区”，存储原始文本、中间结果（如特征、提案）、智能体间的讨论记录。确保每个智能体在响应时，能获取到必要的上下文。
• 判断终止条件：设定收敛标准，例如“连续三轮分类结果无变化”或“所有约束智能体均通过”，并在此条件满足时结束迭代，输出最终结果。

实操心得：在初期，可以用一个简单的Python脚本配合状态机来实现编排器，这样调试起来更直观。后期可以考虑使用LangGraph或AutoGen这类专门的多智能体编排框架，它们提供了更强大的流程控制和工具调用能力。

3.2 感知与理解层：文本特征工程智能体

这一层的智能体负责将原始文本转化为结构化、语义化的信息，供后续决策使用。通常包括：

• 通用特征提取智能体：它的Prompt会要求它从文本中提取关键实体、主题、情感、写作风格等通用特征。例如：“请从以下技术文章中提取：1. 核心提到的技术产品/工具；2. 解决的问题类型（性能、安全、成本）；3. 文章基调（教程、综述、批判）。”
• 定制化维度提取智能体：这是实现“可控性”的关键。如果你关心“部署成本”，就需要一个专门针对“成本”维度进行扫描的智能体。它的Prompt可能是：“请仅关注文本中与软件部署、运维相关的直接与间接成本描述，包括云服务费用、人力成本、时间成本，并量化其表述（如‘高昂’、‘低廉’、‘需要三名工程师’）。”
• 文本摘要智能体：为长文本生成简洁、包含核心论点的摘要。这个摘要将成为后续智能体快速理解文本内容的主要依据，避免每次都传输全文，节省Token消耗。

3.3 决策与生成层：聚类与分类智能体

这是产生分类结果的核心层，智能体们在此“开会讨论”。

• 聚类提议智能体：它接收来自感知层的文本特征和摘要，尝试提出一个聚类方案。Prompt示例：“基于以下文章的摘要和特征，请将它们分成3-5个逻辑类别，并为每个类别命名。命名需体现技术领域和内容深度。”
• 类别标签润色智能体：负责对提议智能体生成的、可能比较粗糙的类别标签进行优化，使其更符合业务术语、更吸引人或更精确。例如，将“快的方法”润色为“高性能优化方案”。
• 约束验证智能体：这是一个“挑刺者”角色。它根据预设的硬性约束检查提案。约束可能包括：
  • 数量约束：“总类别数必须在4到6个之间。”
  • 内容约束：“每个类别下至少包含一篇涉及‘源码分析’的文章。”
  • 语义约束：“类别名称不能是单个技术名词，必须是短语。”
  • 分布约束：“单个类别下的文章数不能超过总数的40%。” 该智能体输出验证结果和具体的修改建议。

3.4 仲裁与反馈层：共识达成与输出

当决策层出现分歧（如提议与约束冲突）时，需要这一层来拍板。

• 仲裁智能体：它听取所有智能体的输出和争论点，做出最终决定。它的Prompt需要赋予较高的“权威性”和综合判断能力，例如：“你是一个首席架构师。以下是关于文章分类的多个提案和反对意见。请评估每个方案的合理性、是否符合所有约束，并最终确定一个最优的分类体系，阐述理由。”
• 结果格式化智能体：将最终达成的共识，格式化成用户需要的输出形式，如JSON、Markdown表格或可视化图谱所需的节点-边数据。

这个架构中，信息流是循环的。约束验证不通过，提案可能被打回给聚类提议智能体重新思考，并附带修改意见。这种设计使得系统具备了自我修正的能力。

4. 关键技术实现与Prompt工程细节

有了架构，每个智能体的“能力”和“性格”就靠Prompt来塑造了。这是最体现工程技巧的部分。

4.1 智能体角色Prompt设计模板

一个有效的角色Prompt通常包含以下几个部分：

你是一个[具体领域]的[专家角色]。 你的核心任务是[清晰的任务描述]。 在完成任务时，请务必遵循以下原则： 1. [原则一， 如：始终从技术实现角度思考] 2. [原则二， 如：你的输出必须是JSON格式] 3. [原则三， 如：如果信息不足，请明确输出“信息不足”而非猜测] 请按以下步骤思考（可选项，用于引导思维链）： 步骤1: 首先，分析输入中的... 步骤2: 然后，重点关注... 步骤3: 最后，综合得出... 你的输出格式必须是： [明确的格式示例，如：{"category_name": "string", "articles": [id_list], "reason": "string"}] 现在，开始处理以下任务： [此处插入具体的任务输入，如文本、特征列表等]

示例：约束验证智能体Prompt

你是一个严格的质量保证专家。你的任务是审核文本分类方案是否违反既定规则。 规则列表： - 规则1: 分类数量必须为5个，不能多也不能少。 - 规则2: 任何分类的名称不能超过5个汉字或10个英文字符。 - 规则3: 每个分类下至少包含2篇文档。 请审核以下分类方案： {classification_scheme} 你的输出必须是JSON格式： { "all_rules_passed": true/false, "violations": [ {"rule_number": 1, "description": "分类数量为4，不符合要求为5。"}, ... ], "suggested_fixes": ["建议合并‘X’和‘Y’类别以达到5个。”] }

4.2 智能体间的通信与上下文管理

智能体不能孤立工作，它们需要交换信息。这里的关键是设计好通信消息格式和上下文窗口管理。

1. 标准化消息格式：建议所有智能体的输入输出都采用结构化的数据格式（如JSON）。这样，编排器可以轻松地从一个智能体的输出中提取特定字段，作为另一个智能体的输入。例如，特征提取智能体输出{"entities": [...], "topics": [...], "sentiment": "neutral"}，聚类智能体直接读取topics字段。

2. 上下文压缩与摘要：多轮对话会消耗大量Token。我们需要一个上下文总结智能体，在每一轮或关键轮次后，对之前的讨论进行精简摘要，保留核心论点和未决争议，替换掉冗长的原始对话历史。这能有效控制成本并让后续讨论聚焦。

3. 工具调用能力：让智能体具备使用工具的能力，可以极大增强系统实用性。例如：

   • 一个智能体可以调用外部API验证某个技术名词的准确性。
   • 另一个智能体可以调用一个本地的聚类算法（如sklearn的K-means）对文本向量进行初步分组，然后将结果作为LLM智能体进一步精炼的起点。这就是传统算法与LLM智能体的混合模式，兼具效率与语义理解。

4.3 迭代循环与终止条件设计

系统不能无限讨论下去，必须设计合理的停止机制。

• 最大迭代轮次：设置一个安全上限，如10轮，防止死循环。
• 共识度度量：比较连续两轮产生的分类结果（如类别名称、文档归属）的相似度。可以使用Jaccard相似度或基于嵌入向量的余弦相似度来计算。当相似度超过一个阈值（如0.95）时，认为已收敛。
• 约束满足度：当约束验证智能体连续报告“所有规则通过”时，即可终止。
• 仲裁裁决：当仲裁智能体做出最终决定后，流程终止。

通常，会采用组合条件，例如：“当连续三轮共识度 > 0.9或所有约束通过或达到最大轮次”时停止。

5. 实战演练：构建一个技术博客分类器

假设我们有一个技术博客文章数据集，我们的目标是建立一个分类体系，要求：1) 反映核心技术栈；2) 区分内容深度（入门/进阶/原理）；3) 每个类别至少包含3篇文章；4) 类别名称直观且包含技术关键词。

5.1 环境准备与智能体初始化

我们使用Python和OpenAI API（或其他兼容API的LLM服务）来构建。为了简化，我们用同一个LLM模型（如GPT-4）通过不同的Prompt来扮演不同角色。

import openai import json from typing import List, Dict, Any # 初始化客户端 client = openai.OpenAI(api_key="your-api-key") model = "gpt-4-turbo-preview" class TextAgent: def __init__(self, name, system_prompt): self.name = name self.system_prompt = system_prompt self.conversation_history = [] def query(self, user_input): messages = [ {"role": "system", "content": self.system_prompt}, *self.conversation_history[-6:], # 保留最近3轮对话作为历史（假设每轮一问一答） {"role": "user", "content": user_input} ] try: response = client.chat.completions.create( model=model, messages=messages, temperature=0.2, # 降低随机性，使输出更稳定 response_format={"type": "json_object"} # 强制JSON输出 ) reply = response.choices[0].message.content self.conversation_history.append({"role": "user", "content": user_input}) self.conversation_history.append({"role": "assistant", "content": reply}) return json.loads(reply) except Exception as e: print(f"智能体 {self.name} 调用出错: {e}") return None # 初始化智能体 feature_agent = TextAgent( name="特征分析员", system_prompt="你是一个技术文章分析专家。请从给定的文章中提取：1. 主要涉及的技术栈（如Python, Kubernetes）；2. 内容深度（入门、进阶、原理剖析、源码解读）。以JSON格式输出：{\"technologies\": [list], \"depth\": \"string\"}" ) cluster_agent = TextAgent( name="分类架构师", system_prompt="你是一个技术内容分类架构师。根据一批文章的技术栈和深度特征，设计一个分类体系。要求类别能综合反映技术和深度，类别数在4-6个之间。输出JSON: {\"categories\": [{\"name\": \"string\", \"criteria\": {\"tech\": [], \"depth\": \"string\"}}]}" ) constraint_agent = TextAgent( name="规则审计员", system_prompt="你是一个严格的规则审计员。检查分类方案是否满足：1. 每个类别对应的文章数>=3；2. 类别名称必须包含技术关键词。输出JSON: {\"passed\": bool, \"issues\": [\"string\"]}" ) arbitration_agent = TextAgent( name="首席仲裁官", system_prompt="你是一个首席技术官，负责最终决策。你会收到分类方案和审计问题。请给出修改后的最终方案，并确保解决所有问题。输出最终的分类体系JSON。" )

5.2 单轮分类流程模拟

假设我们已经有了一批文章的特征列表article_features。

def run_one_round(article_features): """运行一轮分类流程""" # 1. 特征智能体分析每篇文章 all_features = [] for article in article_features: # 这里简化了，实际应该传入文章内容 result = feature_agent.query(f"分析以下文章特征：{article}") if result: all_features.append(result) # 2. 聚类智能体提出方案 cluster_input = {"articles_features": all_features} proposal = cluster_agent.query(json.dumps(cluster_input, ensure_ascii=False)) # 3. 约束智能体验证 # 这里需要模拟将文章分配给类别，实际中可能需要一个额外的“分配智能体”或简单规则 # 假设我们有一个初步分配结果 `assignment` assignment = simulate_assignment(proposal, all_features) audit_input = {"proposal": proposal, "assignment": assignment} audit_result = constraint_agent.query(json.dumps(audit_input, ensure_ascii=False)) # 4. 如果有问题，提交仲裁 if not audit_result.get("passed", True): arbitration_input = { "original_proposal": proposal, "audit_issues": audit_result.get("issues", []) } final_scheme = arbitration_agent.query(json.dumps(arbitration_input, ensure_ascii=False)) return final_scheme else: return proposal def simulate_assignment(proposal, features): """模拟文章分配到类别的过程（简化版）""" # 这是一个占位函数。实际中，可以基于特征与类别标准的相似度来分配。 # 例如，计算每篇文章的特征与每个类别criteria的匹配度，取最高分。 assignment = {} # ... 分配逻辑 ... return assignment

5.3 多轮迭代与结果精炼

我们需要在外面包裹一个循环，实现多轮迭代。

def agentic_clustering(articles, max_rounds=5): """主流程：多智能体迭代聚类""" previous_result = None for round in range(max_rounds): print(f"\n=== 第 {round+1} 轮迭代 ===") current_result = run_one_round(articles) # 这里需要适配，实际输入是文章内容 if previous_result is not None and results_converge(previous_result, current_result): print(f"方案在{round+1}轮后收敛。") return current_result previous_result = current_result # 将本轮结果作为下一轮的部分上下文或初始状态（可通过修改智能体的历史实现） # 例如，让聚类智能体知道上一轮的方案和问题 cluster_agent.conversation_history.append({ "role": "user", "content": f"上一轮的方案和遇到的问题如下，请在此基础上优化：{json.dumps(current_result)}" }) print(f"达到最大迭代轮次{max_rounds}。") return previous_result def results_converge(result_a, result_b, threshold=0.9): """判断两轮结果是否收敛（简化版）""" # 实际中需要比较类别结构、文档归属等。这里简单比较JSON字符串的相似度。 # 更严谨的做法是比较类别名称的集合相似度或文档分配的一致性。 str_a = json.dumps(result_a, sort_keys=True) str_b = json.dumps(result_b, sort_keys=True) # 可以使用更复杂的相似度计算，这里仅示意 return str_a == str_b # 简单判断是否完全相同

通过这样的多轮迭代，智能体们可以不断修正分类方案，直到满足约束并趋于稳定。

6. 性能优化、成本控制与常见问题

将多个LLM智能体投入生产，必须考虑性能和成本。

6.1 延迟与性能优化策略

多轮调用意味着延迟累积。优化策略包括：

• 智能体并行化：当智能体间没有严格依赖时，让它们并行工作。例如，特征提取智能体可以同时分析所有文档，而不是串行。
• 缓存机制：对相同的输入，缓存智能体的输出。例如，相同的文本摘要可以被多次使用，无需重复生成。
• 轻量级模型混合使用：并非所有智能体都需要最强的模型。特征提取、约束检查等相对简单的任务，可以使用更小、更快的模型（如GPT-3.5-Turbo），而仲裁、创意生成等复杂任务再用大模型。
• 异步与非阻塞调用：使用异步编程（如Python的asyncio）来发起LLM API调用，避免等待时间阻塞整个流程。

6.2 Token消耗与成本控制

这是商业应用的核心关切。

• 精简Prompt和上下文：不断优化Prompt，去除冗余指令。严格管理对话历史，只保留最关键的信息。
• 分层处理：先让一个“路由智能体”判断文本的复杂程度。简单文本走快速、低成本的分类路径；复杂、有争议的文本才进入完整的多智能体精炼流程。
• 结果复用与归档：对已稳定分类的文本或类别体系进行归档。当新文本到来时，先尝试匹配已有类别，匹配失败再启动智能体流程。
• 设置预算与熔断：监控每轮、每个智能体的Token消耗，设置每日或每任务预算，超限则触发降级策略（如 fallback 到规则分类）。

6.3 常见问题与调试技巧

在实际操作中，你会遇到各种“诡异”的情况。

问题1：智能体陷入循环争论，无法收敛。

• 排查：检查仲裁智能体的Prompt是否具有足够的权威性和决策逻辑。查看约束是否自相矛盾或过于严苛。
• 解决：增强仲裁智能体的指令，如“你必须从给出的选项中做出明确选择，并停止无休止的讨论”。或者引入“投票机制”，让其他智能体对多个方案投票，仲裁者采纳票数最高的。

问题2：分类结果不稳定，每次运行差异大。

• 排查：首先检查所有智能体的temperature参数是否设置过高（建议决策类任务设在0.1-0.3）。其次，检查输入（如文本特征）的提取是否本身波动很大。
• 解决：降低temperature。为特征提取智能体提供更明确的提取规则和示例（few-shot learning）。在最终输出前，可以增加一轮“一致性校验”，让同一个智能体对结果进行二次确认。

问题3：处理长文档时性能急剧下降。

• 排查：是否将整篇长文档反复发送给每个智能体？
• 解决：实施“摘要先行”策略。第一件事就是用摘要智能体生成一个固定长度的、高质量的摘要。后续所有智能体主要基于摘要工作，仅在需要时（如仲裁者需要查看细节）才请求原文的特定片段。

问题4：某个智能体频繁输出格式错误，导致流程中断。

• 排查：该智能体的Prompt中关于输出格式的指令是否清晰？是否提供了正确的示例？
• 解决：在Prompt中使用结构化输出描述（如JSON Schema）并强制LLM使用JSON模式（如果API支持）。在代码层面对智能体的输出进行健壮性解析，如果解析失败，则自动重试或使用一个默认的、简单的解析器进行修复。

踩坑实录：在一次实践中，约束智能体总是报告“类别名称过长”，但肉眼看起来并不长。后来发现，是因为LLM将中文标点也计入了字符，而我们的规则是基于英文字符定义的。解决方案是在Prompt中明确说明“一个汉字计为2个字符单位”，或者在后续代码中做统一转换。教训：给智能体的规则必须极度精确、无歧义，最好能用代码实现的逻辑就不要完全依赖LLM的理解。

7. 进阶应用与未来展望

Agentic Clustering 的思路可以扩展到许多有趣的方向。

动态与增量聚类：当有新文档不断加入时，不需要对整个数据集重新聚类。可以设计一个“新文档处理智能体”，它判断新文档与现有类别的匹配度。如果匹配度高，则直接归入；如果匹配度低或引起冲突，则触发一个局部的、小范围的多智能体讨论，决定是创建新类别还是调整现有类别边界。这非常适合流式数据或知识库的持续维护。

与向量数据库结合：将文本的向量嵌入（Embedding）存储到向量数据库（如Chroma, Weaviate）中。聚类提议智能体可以首先基于向量相似度进行快速、粗粒度的分组，然后将分组结果和原始文本交给LLM智能体进行语义精炼和命名。这样结合了传统方法的效率与LLM的语义理解能力。

可解释性与可视化：整个多智能体的讨论过程本身就是绝佳的可解释性材料。可以将其自动整理成一份“分类决策报告”，说明每个类别为什么成立，哪些文章是关键，争议点是如何解决的。这比黑箱模型的结果可信度高得多。

领域自适应与微调：如果某个垂直领域（如法律、医疗）有大量标注数据或领域知识，可以对其中某些关键智能体（如特征提取、仲裁）进行LoRA等方式的微调，让它们更精通领域术语和逻辑，从而获得更精准的分类效果。

Agentic Clustering 代表了一种趋势：AI应用正从单一模型调用，走向由多个专业化、可对话的智能体组成的协同系统。它把控制权更灵活地交还给了人类设计者——我们通过设计智能体的角色和互动规则，来间接但精确地塑造我们想要的AI行为。这个过程虽然比调一个API复杂，但带来的可控性、可解释性和灵活性，对于许多严肃的企业应用来说，是至关重要的。