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的关键。假设我们构建一个“内容安全委员会”来处理用户提交的评论初始化与输入用户评论“这个产品太棒了我一定要买”被放入一个新的CouncilContext对象。委员会执行框架开始执行你预先定义好的“委员会”。这个委员会可能由三个Agent组成AgentA情感分析委员它的控制器决定调用“正面情感检测”技能。技能调用一个轻量级的情感分析模型在上下文中写入结果{“sentiment”: “positive”, “score”: 0.95}。AgentB垃圾广告检测委员它的控制器调用“广告关键词匹配”技能。技能检查评论中是否包含“购买”、“点击链接”等短语写入结果{“is_spam”: false, “matched_keywords”: []}。AgentC违规词检测委员它的控制器调用“违禁词过滤”技能。技能比对内部词库写入结果{“has_banned_words”: false}。评估与裁决三个Agent执行完毕后它们产生的响应即技能输出的结果对象被收集起来交给一个预设的Evaluator比如MajorityEvaluator多数决评估器或WeightedEvaluator加权评估器。评估器读取每个响应中的关键字段如is_spam,has_banned_words按照规则进行裁决。例如规则可以是只要has_banned_words为true或is_spam为true则拒绝评论否则结合情感分数给予通过。输出与结束评估器将最终裁决结果如{“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( llmllm, response_threshold0.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 context3. 评估器的训练与迭代高级话题在长期运行中你可以收集大量的决策案例包括所有 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。第三步设计委员会流程我们采用两阶段委员会设计这是处理优先级和复杂分类的常见模式。第一阶段委员会优先级委员会仅包含EmergencyAgent。它的评估器规则很简单如果EmergencySkill输出的置信度超过 90%则直接判定为紧急问题路由到“紧急处理队列”流程结束。否则将上下文包含原始用户问题传递给第二阶段委员会。第二阶段委员会分类委员会包含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 的响应、评估结果以及最终路由决定都需要被详细日志记录。这些日志是宝贵的资产用于监控系统表现统计路由准确率、各技能调用耗时、失败率。优化权重和阈值定期检查误判案例分析是哪个 Agent 判断失误调整其在WeightedEvaluator中的权重或调整技能提示词。发现流程缺陷如果大量工单落入“无法识别”说明需要定义新的技能或调整现有技能的分类边界。踩坑实录技能描述的精确性在初期我们的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.82. 控制器单元测试测试控制器的路由逻辑。提供不同的上下文验证其选择的技能是否正确。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_route4. 黄金数据集与回归测试维护一个“黄金数据集”包含大量有标准答案的输入输出对。每次代码更新或模型升级后运行整个委员会对黄金数据集进行测试确保准确率没有下降回归。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的最终决策也打上点。这样你就能在监控仪表盘上清晰地看到整个决策管道的健康状态哪个技能最慢、哪个技能最容易失败、控制器的路由分布是否均匀、评估器的决策分布如何。这些数据是驱动系统持续优化的最重要依据。