Council框架：构建可编排的智能决策委员会系统

1. 项目概述：从单体应用到分布式决策的演进

在软件架构的演进历程中，我们常常面临一个核心挑战：如何将复杂的业务逻辑从臃肿的单体应用中剥离出来，构建出清晰、可维护且具备高内聚、低耦合特性的系统。传统的做法是引入微服务架构，将不同的业务能力拆分为独立的服务。然而，当业务决策本身变得极其复杂，涉及多步骤、多条件、多数据源的评估与裁决时，仅仅依靠服务拆分往往不够。决策逻辑本身就可能成为一个新的“单体”，难以测试、复用和演进。这正是我最近深度研究并实践的一个开源项目infektyd/council试图解决的问题。

Council这个名字本身就很有意思，它直译为“委员会”或“议会”。想象一下，在一个复杂的决策场景中，比如评估一笔贷款申请，我们不会只依赖一个简单的规则引擎，而是会组建一个“委员会”：信用评分模型、反欺诈系统、收入核实模块、合规性检查服务等“专家”成员各司其职，分别给出自己的“意见”和“评分”，最终由一个“主席”或“议事规则”来汇总这些意见，形成最终的决策。Council项目就是将这一现实世界的决策模式抽象为代码框架，它旨在帮助你构建一个由多个可插拔的“智能体”（Agents）组成的“委员会”，协同工作以完成复杂的任务。

这个框架的核心价值在于，它将决策过程从硬编码的业务逻辑中解放出来，转变为一种可编排、可观测、可扩展的管道（Pipeline）。无论是处理一份文档、分析一段用户对话，还是运行一个自动化工作流，你都可以通过组合不同的“专家”能力来灵活应对。在我过去参与的几个风控和内容审核系统中，决策逻辑的每一次变更都伴随着漫长的开发、测试和上线周期。而Council提供的范式，使得我们可以像搭积木一样，动态调整决策委员会的成员和议事规则，极大地提升了系统的适应性和迭代速度。接下来，我将深入拆解这个项目的设计思路、核心组件以及如何在实际项目中落地。

2. 核心架构与设计哲学解析

2.1 委员会模式：超越简单的链式调用

初看Council，你可能会联想到诸如 LangChain 这类用于构建大语言模型（LLM）应用的工作流框架。它们确实有相似之处，都涉及链（Chain）或代理（Agent）的概念。但Council的设计哲学有更明确的侧重点：专注于复杂决策的编排与执行，而非仅仅是 LLM 的调用封装。

它的核心抽象是CouncilContext,Agent,Skill,Controller和Evaluator。我们可以这样理解它们的关系：

• Agent（代理/委员）：委员会中的基本成员单位。每个Agent封装了解决特定子问题的能力。一个Agent必须至少包含一个Skill（技能）和一个Controller（控制器）。
• Skill（技能）：这是Agent真正干活的部分。它接收输入（CouncilContext），执行具体的逻辑（可以是调用一个 LLM、执行一段代码、查询数据库等），并产生输出。例如，“情感分析技能”、“实体提取技能”、“SQL查询技能”。
• Controller（控制器）：这是Agent的“大脑”。它决定在某个执行周期中，调用哪一个（或哪几个）Skill。最简单的控制器可能总是调用同一个技能，而复杂的控制器可以根据上下文动态选择最合适的技能。
• Evaluator（评估器）：这是委员会“主席”角色的核心。当多个Agent针对同一任务给出了自己的响应（可能是一个答案，也可能是一段分析）后，Evaluator负责对这些响应进行评估、打分或排序，最终选出最优结果，或者综合所有结果生成最终输出。
• CouncilContext（上下文）：在整个委员会执行过程中流转的数据总线。它包含了原始的用户输入、每个处理环节产生的中间数据、执行状态以及最终的输出结果。所有组件都通过它来读取和写入数据。

这种设计的精妙之处在于关注点分离和责任链的显式化。在传统代码中，一个复杂的if-else或switch-case语句块可能隐含了控制器逻辑；分散在各处的函数调用可能就是技能；最后的结果合并可能就是评估器。Council迫使你将它们清晰地定义出来，这使得每个部分都可以被独立地开发、测试、替换和监控。

2.2 管道执行流程：一次请求的生命周期

理解数据流是掌握Council的关键。假设我们构建一个“内容安全委员会”来处理用户提交的评论：

1. 初始化与输入：用户评论“这个产品太棒了，我一定要买！”被放入一个新的CouncilContext对象。
2. 委员会执行：框架开始执行你预先定义好的“委员会”。这个委员会可能由三个Agent组成：
   • AgentA（情感分析委员）：它的控制器决定调用“正面情感检测”技能。技能调用一个轻量级的情感分析模型，在上下文中写入结果：{“sentiment”: “positive”, “score”: 0.95}。
   • AgentB（垃圾广告检测委员）：它的控制器调用“广告关键词匹配”技能。技能检查评论中是否包含“购买”、“点击链接”等短语，写入结果：{“is_spam”: false, “matched_keywords”: []}。
   • AgentC（违规词检测委员）：它的控制器调用“违禁词过滤”技能。技能比对内部词库，写入结果：{“has_banned_words”: false}。
3. 评估与裁决：三个Agent执行完毕后，它们产生的响应（即技能输出的结果对象）被收集起来，交给一个预设的Evaluator，比如MajorityEvaluator（多数决评估器）或WeightedEvaluator（加权评估器）。评估器读取每个响应中的关键字段（如is_spam,has_banned_words），按照规则进行裁决。例如，规则可以是：只要has_banned_words为true或is_spam为true，则拒绝评论；否则，结合情感分数，给予通过。
4. 输出与结束：评估器将最终裁决结果（如{“status”: “approved”, “reason”: “positive sentiment, no spam or banned words”}）写回CouncilContext。框架将最终上下文返回给调用方。

整个过程中，每个Agent都是独立、可重用的。如果你想增加一个“图片OCR识别委员”来处理带图的评论，只需定义一个新的Agent并将其加入委员会即可，无需修改现有任何代码。这种可插拔性是Council最大的优势之一。

3. 核心组件深度拆解与实操

3.1 技能（Skill）的设计：不止于LLM封装

很多人会误以为Skill就是包装一个 LLM 的 API 调用。这大大低估了它的能力。Skill本质是一个可执行单元，任何可以接收输入、产生输出的逻辑都可以封装成Skill。

1. 自定义代码技能：这是最灵活的方式。你可以将任何现有的业务函数包装成Skill。

from council.skills import SkillBase from council.context import CouncilContext class DatabaseQuerySkill(SkillBase): def __init__(self): super().__init__("database_query") # 初始化数据库连接等 self.db_client = ... def execute(self, context: CouncilContext) -> str: # 从上下文中获取用户查询意图 query_intent = context.current.get(“user_query”) # 执行复杂的数据库查询逻辑 result = self._run_complex_sql(query_intent) # 将结果以结构化或文本形式返回 return f“查询结果: {result}”

2. LLM技能：这是自然语言处理任务的核心。Council通常与LLMEngine结合，后者统一管理不同模型提供商（如 OpenAI, Anthropic, 本地模型）的调用。一个 LLM Skill 的关键在于设计高质量的提示词（Prompt），并将上下文中的信息有效地填充进去。

from council.skills import LLMSkill from council.llm import OpenAILLM llm = OpenAILLM(api_key=“your_key”, model=“gpt-4”) prompt_template = “““ 你是一个资深产品评论分析师。请分析以下用户评论的情感倾向和主要观点。 评论：{user_input} 请以JSON格式输出，包含‘sentiment’（positive/neutral/negative）和‘key_points’（列表）字段。 “““ sentiment_skill = LLMSkill(llm, prompt_template, name=“sentiment_analysis”)

3. 工具调用技能（进阶）：对于需要执行具体动作的场景，如发送邮件、调用外部API、操作文件，可以设计工具调用技能。这类技能的重点在于错误处理与状态回滚。例如，一个“发送邮件技能”在执行失败时，应该在上下文中留下明确的错误标志，以便后续的评估器或控制器能据此做出反应（如触发重试或转人工）。

实操心得：技能设计的单一职责原则一个常见的错误是把太多逻辑塞进一个Skill。比如，设计一个“分析并存储评论技能”，既做情感分析，又把结果写入数据库。这违反了单一职责原则，不利于测试和复用。正确的做法是拆分成“情感分析Skill”和“数据持久化Skill”，然后通过控制器的编排让它们顺序执行。这样，当数据库 schema 变更时，你只需要修改持久化技能，情感分析部分完全不受影响。

3.2 控制器（Controller）的策略：动态路由的艺术

控制器决定了Agent在本次执行中要做什么。最简单的BasicController总是执行同一个技能。但Council的威力在于其动态控制器。

1.LLMController：这是非常强大的一个组件。它利用 LLM 的推理能力，根据当前上下文，动态决定调用哪个技能。你需要为它提供一个技能列表和相应的描述。

from council.controllers import LLMController controller = LLMController( llm=llm, response_threshold=0.7, # 置信度阈值 skills=[sentiment_skill, query_skill, translation_skill], skills_descriptions={ “sentiment_analysis”: “当需要判断文本情感时使用此技能。”, “database_query”: “当问题涉及查询产品信息或订单状态时使用此技能。”, “translation”: “当需要将文本翻译成其他语言时使用此技能。” } )

当请求“告诉我用户‘张三’对最新产品的评价怎么样？”时，LLMController可能会分析出需要先调用database_query技能获取评论内容，再调用sentiment_analysis技能进行分析。它自己会生成一个执行计划。

2. 自定义规则控制器：对于业务规则明确的场景，可以基于上下文中的数据进行逻辑判断。

from council.controllers import ControllerBase class RuleBasedController(ControllerBase): def __init__(self, skills): super().__init__() self.skills = skills def select_skill(self, context): user_tier = context.current.get(“user_tier”, “standard”) if user_tier == “premium”: # 为高级用户使用更精准但昂贵的技能 return self.skills[“premium_analysis”] else: return self.skills[“standard_analysis”]

注意事项：控制器的性能与稳定性使用LLMController虽然灵活，但会引入额外的 LLM 调用延迟和成本。在生产环境中，需要谨慎评估。一种最佳实践是分层控制：第一层使用快速的、基于规则的控制器处理大部分明确请求；将规则无法处理的、模糊的请求，路由给第二层的LLMController。同时，务必为LLMController设置超时和重试机制，并对其决策进行日志记录和监控，以便分析和优化技能描述。

3.3 评估器（Evaluator）的抉择：从投票到强化学习

委员会的所有Agent完成任务后，评估器负责做最终决定。这是将多个“专家意见”融合成“集体智慧”的关键步骤。

1. 基础评估器：

• MajorityEvaluator：适用于分类任务，选择得票最多的类别。
• WeightedEvaluator：为每个Agent分配权重（基于历史准确率、成本或优先级），计算加权得分。
• LLMEvaluator：利用 LLM 作为“超级主席”，阅读所有Agent的响应和推理过程，给出最终判断和理由。这能力最强，但成本也最高。

2. 实现一个自定义业务评估器：在风控场景中，决策规则可能很复杂。例如，三个 Agent 分别输出：欺诈概率、信用评分、行为异常度。最终决策可能是一个分段函数。

from council.evaluators import EvaluatorBase class RiskEvaluator(EvaluatorBase): def execute(self, context): responses = context.last_responses # 获取所有Agent的响应 fraud_score = responses[“fraud_detector”].get(“score”, 0) credit_score = responses[“credit_scorer”].get(“score”, 650) anomaly_flag = responses[“behavior_analyzer”].get(“is_anomalous”, False) final_decision = “approve” if fraud_score > 0.8: final_decision = “reject” elif fraud_score > 0.5 and credit_score < 600: final_decision = “review” elif anomaly_flag and credit_score < 700: final_decision = “review” # ... 更复杂的规则 context.set_evaluation_result({“decision”: final_decision, “scores”: {“fraud”: fraud_score, “credit”: credit_score}}) return context

3. 评估器的训练与迭代（高级话题）：在长期运行中，你可以收集大量的决策案例（包括所有 Agent 的中间结果和最终业务结果）。利用这些数据，可以训练一个机器学习模型（如梯度提升树或神经网络）作为评估器，自动学习最优的决策边界，这比手动编写规则更精准，并能持续优化。Council的架构为这种迭代提供了完美的数据基础。

4. 构建一个真实项目：智能客服路由系统

让我们通过一个完整的例子，将上述概念串联起来。假设我们要构建一个智能客服路由系统，目标是根据用户输入的工单描述，自动将其分配给最合适的处理团队（技术组、账单组、售后组、普通咨询组）。

4.1 系统设计与组件定义

第一步：定义技能（Skills）我们需要四个分类技能，每个技能专注于识别问题是否属于自己负责的领域。

• TechSkill: 识别技术问题（如“无法登录”、“软件崩溃”）。
• BillingSkill: 识别账单问题（如“扣费错误”、“订阅续费”）。
• AfterSalesSkill: 识别售后问题（如“退货申请”、“维修进度”）。
• GeneralSkill: 识别一般咨询（如“如何使用某功能”、“营业时间”）。

此外，还需要一个EmergencySkill用于识别紧急问题（如“数据丢失”、“安全漏洞”），这类问题需要优先路由。

每个技能都是一个 LLM Skill，使用精心设计的少量示例（Few-shot）提示词来保证分类准确性。

第二步：定义代理（Agents）我们创建五个 Agent，每个对应一个技能，并使用BasicController。

• TechAgent: 控制器绑定TechSkill。
• BillingAgent: 控制器绑定BillingSkill。
• AfterSalesAgent: 控制器绑定AfterSalesSkill。
• GeneralAgent: 控制器绑定GeneralSkill。
• EmergencyAgent: 控制器绑定EmergencySkill。

第三步：设计委员会流程我们采用两阶段委员会设计，这是处理优先级和复杂分类的常见模式。

1. 第一阶段委员会（优先级委员会）：仅包含EmergencyAgent。它的评估器规则很简单：如果EmergencySkill输出的置信度超过 90%，则直接判定为紧急问题，路由到“紧急处理队列”，流程结束。否则，将上下文（包含原始用户问题）传递给第二阶段委员会。
2. 第二阶段委员会（分类委员会）：包含TechAgent,BillingAgent,AfterSalesAgent,GeneralAgent。四个 Agent 并行执行各自的分类技能。

第四步：定义评估器对于第二阶段委员会，我们使用WeightedEvaluator。因为不同分类的准确率和重要性不同。例如，技术问题和账单问题误判的成本高，因此给予更高的权重。我们可以根据历史数据来设定初始权重，并在后期调整。

evaluator = WeightedEvaluator(weights={“TechAgent”: 0.35, “BillingAgent”: 0.35, “AfterSalesAgent”: 0.2, “GeneralAgent”: 0.1})

评估器会收集每个 Agent 输出的置信度分数，计算加权平均，并选择加权得分最高的类别作为路由目标。如果最高得分低于某个阈值（如 0.6），则判定为“无法识别”，路由给人工客服。

4.2 实现细节与配置

上下文数据流设计：CouncilContext中需要流转的关键数据包括：

• user_input: 原始工单描述。
• phase: 当前阶段（“priority_check”, “category_classification”）。
• emergency_score: 紧急技能给出的分数。
• category_scores: 字典，记录分类委员会中各 Agent 的得分。
• final_route: 最终路由目标。

控制器与技能配置：对于分类技能，使用LLMSkill并配置温度（temperature）为 0，以保证输出稳定性。提示词中明确要求输出 JSON 格式：{“is_relevant”: true/false, “confidence”: 0.95, “reason”: “...”}。

错误处理与降级：在任何技能调用失败（如 LLM API 超时）时，该 Agent 应返回一个默认的低置信度结果（如{“is_relevant”: false, “confidence”: 0.1}），而不是让整个委员会崩溃。这确保了系统的鲁棒性。

4.3 部署与监控

将上述委员会封装为一个服务（如 FastAPI 应用）。每个请求的完整上下文、每个 Agent 的响应、评估结果以及最终路由决定，都需要被详细日志记录。这些日志是宝贵的资产，用于：

1. 监控系统表现：统计路由准确率、各技能调用耗时、失败率。
2. 优化权重和阈值：定期检查误判案例，分析是哪个 Agent 判断失误，调整其在WeightedEvaluator中的权重，或调整技能提示词。
3. 发现流程缺陷：如果大量工单落入“无法识别”，说明需要定义新的技能或调整现有技能的分类边界。

踩坑实录：技能描述的精确性在初期，我们的GeneralSkill提示词描述过于宽泛：“处理其他所有问题”。导致很多本应属于技术或账单的问题被分到这里，因为 LLM 觉得它也符合“其他问题”。后来我们将提示词修改为：“处理关于产品功能使用、公司政策咨询、非技术性操作指南等通用咨询问题。注意：涉及错误、故障、支付、交易的问题不属于此类。” 并增加了反例，分类准确性立刻大幅提升。这告诉我们，技能（尤其是基于LLM的技能）的边界必须用清晰的语言和例子来定义。

5. 性能优化、测试与常见问题排查

5.1 性能优化策略

在Council架构中，性能瓶颈主要出现在两个方面：LLM调用延迟和Agent的并行/串行编排。

1. 并行执行优化：默认情况下，委员会中的Agent是顺序执行的。对于相互无依赖的Agent，务必启用并行执行。在定义Council时，可以指定执行策略。

from council.chains import Parallel parallel_chain = Parallel(agents=[agent1, agent2, agent3]) # agent1,2,3 会并行执行

但要注意，并行会同时消耗更多的 Token 和 API 配额，需要权衡速度和成本。

2. 技能缓存：对于纯函数式、输入相同则输出必然相同的技能（如某些数据查询、计算），可以实现缓存层。将输入参数的哈希值作为键，输出结果作为值缓存起来（可以使用内存缓存如functools.lru_cache或外部缓存如 Redis），能极大提升重复请求的响应速度。

3. LLM调用批处理与降级：如果多个技能使用同一个 LLM 模型，可以考虑在基础设施层实现请求批处理。对于非关键路径或对延迟不敏感的技能，可以使用更小、更快的模型（如 GPT-3.5-turbo 对比 GPT-4）作为降级方案。

4. 超时与熔断：为每个Skill的执行设置超时。如果一个技能长时间无响应，应中断其执行并返回一个默认或错误状态，防止整个请求被卡住。可以使用熔断器模式，当某个技能连续失败多次后，暂时将其“熔断”，直接返回降级结果，过一段时间再尝试恢复。

5.2 测试方法论

测试一个委员会比测试单个函数复杂，但Council的模块化设计让单元测试和集成测试成为可能。

1. 技能单元测试：单独测试每个Skill。模拟一个CouncilContext作为输入，验证其输出是否符合预期。对于 LLM Skill，可以使用固定的提示词和 Mock 的 LLM 响应来进行测试。

def test_tech_skill(): skill = TechSkill() context = CouncilContext.from_user_message(“我的软件崩溃了”) result = skill.execute(context) assert “is_relevant” in result assert result[“is_relevant”] == True assert result[“confidence”] > 0.8

2. 控制器单元测试：测试控制器的路由逻辑。提供不同的上下文，验证其选择的技能是否正确。

def test_llm_controller_routing(): controller = LLMController(...) context1 = CouncilContext.from_user_message(“帮我查一下账单”) selected1 = controller.select_skill(context1) assert selected1.name == “billing_skill”

3. 集成测试（委员会测试）：使用真实或模拟的数据，端到端地测试整个委员会。这是验证评估器逻辑和 Agent 间协作是否正确的关键。

def test_full_council_routing(): council = build_customer_service_council() # 构建完整的委员会 test_cases = [ (“系统蓝屏了”, “tech”), (“上个月多扣了我钱”, “billing”), (“我想退货”, “after_sales”), ] for input_text, expected_route in test_cases: context = council.execute_sync(input_text) assert context.final_decision[“route”] == expected_route

4. 黄金数据集与回归测试：维护一个“黄金数据集”，包含大量有标准答案的输入输出对。每次代码更新或模型升级后，运行整个委员会对黄金数据集进行测试，确保准确率没有下降（回归）。

5.3 常见问题排查表

在实际运维中，你会遇到各种各样的问题。下面是一个快速排查指南：

问题现象	可能原因	排查步骤与解决方案
委员会返回结果不一致	1. LLM 温度参数过高。 2. 技能或控制器有随机性逻辑。 3. 上下文数据被意外修改。	1. 将 LLM 技能的温度（temperature）设为 0。 2. 检查自定义技能中是否使用了随机函数。 3. 检查各个组件对CouncilContext的读写，确保没有全局状态污染。使用深度拷贝隔离中间数据。
某个 Agent 始终不生效	1. 控制器从未选中该 Agent 的技能。 2. 该 Skill 执行出错但被静默处理。 3. 评估器权重为0或极低。	1. 检查控制器的选择逻辑和技能描述。增加该技能的描述清晰度。 2. 查看该 Skill 的日志和错误信息。确保其execute方法有完善的异常捕获和日志输出。 3. 检查WeightedEvaluator的权重配置。
系统响应时间过长	1. Agent 串行执行，可并行化。 2. 某个 Skill（尤其是LLM）响应慢。 3. 网络延迟或下游服务慢。	1. 使用Parallel链对无依赖的 Agent 进行并行编排。 2. 为该 Skill 设置超时，并考虑使用缓存或更快的模型降级。 3. 为所有外部调用添加超时和重试机制，并监控其 P95/P99 延迟。
评估器决策不符合预期	1. 评估规则逻辑有误。 2. 上游 Agent 输出的数据格式不符合评估器预期。 3. 权重设置不合理。	1. 单元测试评估器逻辑，使用多种边界案例测试。 2. 打印或记录所有 Agent 的响应结果，检查其字段名和数据类型是否与评估器读取的字段一致。 3. 收集一批误判案例，人工分析后调整权重或规则。
内存消耗持续增长	1. 上下文对象在管道中不断累积未清理的中间数据。 2. 技能中存在内存泄漏。	1. 明确上下文数据的生命周期，非必要数据及时清理。对于长管道，考虑分段处理并清理早期阶段的中间数据。 2. 使用内存分析工具（如tracemalloc）定位泄漏点，检查技能中是否有全局列表或字典在无限增长。

最后再分享一个关于监控的小技巧：除了记录最终结果，一定要为每个Skill的执行耗时、成功/失败状态打点（Metrics）。为Controller的选择结果和Evaluator的最终决策也打上点。这样，你就能在监控仪表盘上清晰地看到整个决策管道的健康状态：哪个技能最慢、哪个技能最容易失败、控制器的路由分布是否均匀、评估器的决策分布如何。这些数据是驱动系统持续优化的最重要依据。