Council框架：构建多AI智能体协作系统的工程实践指南

1. 项目概述：一个面向AI代理的“议会”框架

最近在折腾AI应用开发的朋友，可能都遇到过类似的困境：单个大语言模型（LLM）能力再强，面对复杂任务时也常常显得力不从心。比如，你需要它分析一份市场报告、生成代码、再给出商业建议——这要求模型同时具备数据分析、编程和商业洞察力，而目前几乎没有哪个单一模型能完美覆盖所有领域。于是，一个自然的想法就产生了：能不能让多个各有所长的AI“专家”坐在一起，像议会辩论一样，共同协作来解决一个复杂问题？这就是我今天要深入拆解的Council项目。

Council，直译过来就是“议会”或“委员会”。这个开源项目的核心思想，正是构建一个多智能体协作框架。它不是一个具体的应用，而是一个框架和工具包。你可以把它想象成一个高度可定制的“议事厅”，在这里，你可以定义不同的“议员”（AI Agent），每个议员拥有特定的技能（如调用某个API、使用某个工具、擅长某个领域的分析），并为他们设定辩论和决策的规则。最终，通过议员们的讨论、投票或执行链，得出一个比任何单一智能体都更优的解决方案。

这个框架的价值在于，它将多智能体协作从一种手工作坊式的、胶水代码粘合的模式，提升到了工程化和标准化的层面。对于开发者而言，你不用再从零开始设计智能体间的通信协议、任务调度和结果整合逻辑，Council提供了一套现成的“基础设施”。无论是想构建一个复杂的AI客服系统（路由、查询、总结由不同Agent负责），还是一个自动化的代码审查工具（安全检查、风格检查、性能分析由不同专家完成），Council都能提供一个清晰的架构蓝图。

2. 核心架构与设计哲学：如何构建一个高效的“AI议会”

Council的核心设计非常优雅，它抽象出了几个关键概念，理解了这些，你就掌握了使用和扩展它的钥匙。

2.1 核心组件四重奏：Agent、Chain、Controller、Evaluator

整个Council框架围绕着四个核心组件运转，它们的关系可以类比为一个真实的议会流程：

1. Agent（议员）：这是最基本的执行单元。一个Agent通常由三部分组成：

   • 技能（Skill）：Agent能做什么。比如，一个“Python程序员”Agent的技能可能是“编写Python代码”；一个“数据分析师”Agent的技能可能是“使用pandas进行数据清洗”。技能背后可以是一个简单的提示词模板，也可以封装一个复杂的工具调用（如执行SQL查询、调用外部API）。
   • 上下文（Context）：Agent执行任务时所处的环境信息，包括用户输入、历史对话、其他Agent的中间结果等。Council的Context对象负责在Agent之间传递和共享这些状态。
   • 预算（Budget）：这是一个非常实用的设计。每个Agent的执行（尤其是调用昂贵的LLM API）都会消耗预算（如Token数、API调用次数、时间）。预算机制可以防止某个Agent陷入死循环或产生过于冗长的输出，确保整个系统的资源消耗可控。

2. Chain（执行链）：这是最简单直接的协作模式。多个Agent按预定义的顺序依次执行，前一个Agent的输出作为后一个Agent的输入。这适合流程清晰、步骤依赖强的任务。例如，一个数据处理链：数据提取Agent -> 数据清洗Agent -> 数据分析Agent -> 报告生成Agent。

3. Controller（议长/调度器）：这是Council的“大脑”，负责更复杂的多Agent协作逻辑。Controller接收一个任务，然后决定如何调度旗下的Agents。最常见的Controller是LLMController，它利用一个大语言模型（如GPT-4）作为“议长”，根据任务描述，动态地决定调用哪个或哪几个Agent，甚至规划他们的执行顺序。这实现了初步的“动态规划”能力。

4. Evaluator（评估器）：议会讨论后需要表决。Evaluator就是用来评估Agent工作成果的组件。它可以是基于规则的（如检查输出是否包含关键词），也可以是基于另一个LLM的（如让一个“质量评审员”LLM对结果进行打分）。在Chain或Controller的执行过程中，Evaluator可以用来选择最佳的输出路径，或者对最终结果进行过滤和排序。

2.2 工作流程：一次典型的“议会审议”

假设我们要用Council构建一个“技术博客助手”，任务是“为‘如何使用Council框架’这个主题写一篇博文”。

1. 任务提交：用户提出请求，系统创建初始Context，包含任务描述。
2. Controller调度：LLMController（议长）分析任务，认为需要三个专家：一个“大纲生成Agent”，一个“内容撰写Agent”，一个“代码示例生成Agent”。
3. Agent执行与辩论：
   • Controller可能选择让三个Agent并行执行，各自生成一份博客草稿。
   • 或者，采用顺序链：先由“大纲生成Agent”产出结构，然后“内容撰写Agent”和“代码示例生成Agent”根据大纲并行填充内容。
4. 评估与整合：LLMEvaluator（评估器）对多个Agent产生的草稿进行评分，选择综合得分最高的一份。或者，由一个“编辑Agent”负责将所有草稿的优点整合成最终版本。
5. 结果返回：最终生成的博文通过Context返回给用户。

整个过程中，预算系统会监控每个Agent的Token消耗，确保总成本不超标。Context则像一份共享的会议纪要，记录着所有中间讨论和结果。

2.3 设计优势与适用场景

这种设计带来了几个显著优势：

• 模块化与可复用：Agent、Skill、Evaluator都是独立的模块，可以像乐高积木一样随意组合，构建不同的应用。
• 灵活性高：既支持简单的线性Chain，也支持由LLM驱动的智能动态规划（Controller）。
• 可控性强：预算和评估机制让整个系统在成本和质量上都是可控的。
• 易于扩展：你可以轻松地接入新的LLM提供商（OpenAI, Anthropic, 本地模型等），定义新的Skill或Evaluator。

它特别适合以下场景：

• 复杂任务分解：需要多个步骤和专业知识的任务，如市场调研、竞品分析、项目计划制定。
• 质量与多样性要求：需要从多个角度生成内容或方案，并从中择优，如创意写作、方案设计。
• 流程自动化：将企业中已有的、由不同角色负责的审核、处理流程自动化，如内容审核流水线、客户工单分类与处理。

3. 从零开始：搭建你的第一个Council应用

理论讲得再多，不如亲手跑一遍。下面我将带你一步步搭建一个简单的Council应用，这个应用的功能是：根据用户提出的编程问题，自动判断其所属的技术领域，并调用相应的“专家Agent”来解答。

3.1 环境准备与安装

首先，确保你的Python环境在3.8以上。Council可以通过pip直接安装，非常方便。

# 创建并进入项目目录 mkdir my_council_app && cd my_council_app # 创建虚拟环境（推荐） python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Mac/Linux: source venv/bin/activate # 安装Council核心库 pip install council-ai

注意：Council是一个较新的项目，API可能仍在快速迭代中。建议在开始正式项目前，查看其官方GitHub仓库（infektyd/council）的最新文档和Release Notes，以了解是否有重大变更。

除了Council本身，我们还需要安装用于调用LLM的SDK。这里以OpenAI为例：

pip install openai

请确保你已经设置了OpenAI的API密钥作为环境变量：

# 在终端中设置（临时） export OPENAI_API_KEY='your-api-key-here' # 或者在代码中设置（不推荐，可能泄露） import os os.environ["OPENAI_API_KEY"] = "your-api-key-here"

3.2 定义第一个Skill（技能）

Skill是Agent能力的基石。我们先创建一个最简单的Skill，它使用LLM来回答问题。

# skills/basic_qa_skill.py import logging from council.contexts import ChatMessage, SkillContext from council.skills import LLMSkill from council.llm import OpenAILLM # 配置LLM客户端，这里使用gpt-3.5-turbo，成本较低适合实验 llm_client = OpenAILLM(model="gpt-3.5-turbo") class BasicQASkill(LLMSkill): """一个基础的问答技能，将用户问题直接抛给LLM""" def __init__(self): # 初始化父类，传入LLM客户端和一个系统提示词 system_prompt = "你是一个乐于助人的技术专家。请用清晰、准确的语言回答用户的问题。" super().__init__(llm_client, system_prompt) def execute(self, context: SkillContext) -> ChatMessage: # 从上下文中获取最新的用户消息 user_message = context.current.last_user_message # 调用LLM生成回答 result = self._llm_client.chat_completion([user_message]) # 返回一个AI类型的消息 return ChatMessage.agent(result)

这个BasicQASkill继承自LLMSkill，它封装了与LLM交互的细节。system_prompt定义了AI的角色。execute方法是技能的执行入口，它接收SkillContext（包含对话历史等信息），并返回一个ChatMessage。

3.3 创建专属Agent（议员）

有了Skill，我们就可以创建Agent了。一个Agent可以拥有多个Skill，但这里我们先给每个Agent分配一个专属技能，打造“专家”人设。

# agents/__init__.py from council.agents import Agent from council.controllers import BasicController from council.processors import BasicProcessor from council.chains import Chain # 导入我们定义的技能 from skills.basic_qa_skill import BasicQASkill from skills.python_skill import PythonExpertSkill # 假设我们之后会定义 from skills.web_skill import WebDevSkill # 假设我们之后会定义 def create_python_agent() -> Agent: """创建Python专家Agent""" skill = BasicQASkill() # 这里可以先共用基础技能，后续替换为更专业的Python技能 chain = Chain(name="Python解答链", description="专门处理Python相关问题", runners=[skill]) controller = BasicController(chains=[chain]) processor = BasicProcessor() return Agent(controller, processor, name="Python专家") def create_web_agent() -> Agent: """创建Web开发专家Agent""" skill = BasicQASkill() # 同上 chain = Chain(name="Web解答链", description="专门处理HTML/CSS/JS相关问题", runners=[skill]) controller = BasicController(chains=[chain]) processor = BasicProcessor() return Agent(controller, processor, name="Web开发专家") def create_general_agent() -> Agent: """创建通用问题处理Agent""" skill = BasicQASkill() chain = Chain(name="通用解答链", description="处理其他通用技术问题", runners=[skill]) controller = BasicController(chains=[chain]) processor = BasicProcessor() return Agent(controller, processor, name="通用助手")

这里我们创建了三个Agent，每个Agent内部都包含一个Chain（目前只有单个Skill）、一个BasicController（最简单的控制器，按顺序或并行执行Chain）和一个BasicProcessor（处理输入输出）。BasicController在这里的作用相对简单，因为每个Agent只有一个Chain。在更复杂的场景中，Controller会负责在多个Chain间做选择。

3.4 实现智能路由：LLMController（议长）

现在我们有几位“专家”了，需要一个“议长”来根据问题类型分发任务。这就是LLMController的用武之地。

# controllers/router_controller.py from council.controllers import LLMController from council.llm import OpenAILLM from council.contexts import ChatMessage class RouterController(LLMController): """一个智能路由控制器，使用LLM分析问题并分发给对应的Agent链""" def __init__(self, chains): llm = OpenAILLM(model="gpt-3.5-turbo") # 给LLM的指令，告诉它如何根据问题选择专家 prompt = ( "你是一个技术问题路由员。请分析用户的问题，并严格从以下选项中选择最相关的专家来处理：\n" "1. Python专家 - 如果问题关于Python语法、库（如pandas, numpy）、框架（如Django, Flask）。\n" "2. Web开发专家 - 如果问题关于HTML、CSS、JavaScript、前端框架（React, Vue）或HTTP协议。\n" "3. 通用助手 - 如果问题关于其他编程语言（Java, C++）、算法、数据结构、软件工程理念或无法明确归类。\n" "只返回选项的数字（1, 2, 3），不要有任何其他解释。" ) super().__init__(llm=llm, chains=chains, response_threshold=0.7, top_k=1, prompt=prompt) def _parse_llm_response(self, response: ChatMessage) -> str: # 解析LLM的返回，期望是“1”、“2”或“3” choice = response.message.strip() if choice in ["1", "2", "3"]: # 将数字映射到Chain的名称（需要与传入的chains顺序对应） chain_names = [chain.name for chain in self._chains] # 简单映射：1->Python解答链， 2->Web解答链， 3->通用解答链 # 这里假设chains传入的顺序是 [python_chain, web_chain, general_chain] return chain_names[int(choice) - 1] else: # 如果LLM没有按要求返回，则默认路由到通用助手 logging.warning(f"LLM returned unexpected choice: '{choice}'. Defaulting to general chain.") return "通用解答链"

这个RouterController继承自LLMController。它的核心是prompt，我们通过精心设计的提示词，引导LLM扮演路由员的角色。response_threshold是置信度阈值，低于此值的结果可能被忽略。_parse_llm_response方法用于将LLM的文本输出解析成框架能理解的Chain名称。

3.5 组装完整应用并运行

最后，我们把所有部件组装起来，形成一个完整的多Agent系统。

# main.py import logging from council.contexts import AgentContext, ChatMessage from council.agents import Agent from council.processors import BasicProcessor from agents import create_python_agent, create_web_agent, create_general_agent from controllers.router_controller import RouterController from council.chains import Chain logging.basicConfig(level=logging.INFO) def main(): # 1. 创建各个专家Agent内部的Chain（实际应用中，这部分可能已经在Agent内部定义好） python_agent = create_python_agent() web_agent = create_web_agent() general_agent = create_general_agent() # 为了给Controller使用，我们需要提取出每个Agent的主Chain。 # 这里简化处理，我们直接使用Agent的chain属性（假设我们修改了Agent创建方式使其暴露chain）。 # 更标准的做法是直接构建Chain列表，但为了演示清晰，我们采用以下方式： python_chain = Chain(name="Python解答链", runners=[python_agent]) web_chain = Chain(name="Web解答链", runners=[web_agent]) general_chain = Chain(name="通用解答链", runners=[general_agent]) all_chains = [python_chain, web_chain, general_chain] # 2. 创建智能路由控制器（议长） router_controller = RouterController(chains=all_chains) # 3. 创建顶层的“议会”Agent，其核心就是这个路由控制器 council_processor = BasicProcessor() council_agent = Agent(controller=router_controller, processor=council_processor, name="技术议会") # 4. 运行测试 test_questions = [ "如何在Python中反转一个列表？", "CSS的Flex布局和Grid布局有什么区别？", "什么是快速排序算法的时间复杂度？", "用Django如何实现用户认证？" ] for question in test_questions: print(f"\n用户问题: {question}") print("-" * 40) # 创建执行上下文 context = AgentContext.from_user_message(question) # 设置预算（例如，最多消耗1000个Token） context.budget = 1000 try: # 执行议会Agent result = council_agent.execute(context) # 打印结果 if result and result.messages: best_message = result.messages[-1] # 通常最后一个消息是最佳答案 print(f"回答 ({best_message.source}): {best_message.message}") print(f"本次消耗Token: {context.budget.spent}") else: print("未获得有效回答。") except Exception as e: print(f"执行出错: {e}") if __name__ == "__main__": main()

运行这个main.py，你会看到系统对每个问题，首先会由RouterController（LLM）判断问题类型，然后自动路由到对应的专家Agent进行解答，并打印出回答内容和消耗的Token数。

实操心得：在初次运行这类多Agent系统时，最容易出现的问题是LLM的路由判断不准。如果发现路由错误（比如把Python问题分给了Web专家），不要急于修改代码，首先应该优化RouterController中的提示词（prompt）。让指令更清晰，提供更明确的例子，往往能极大提升路由准确性。这是提示词工程（Prompt Engineering）在多Agent系统中的关键应用。

4. 进阶技巧与实战优化

一个能跑通的Demo只是起点。要让Council应用真正可靠、高效，还需要考虑很多工程细节。

4.1 技能（Skill）的深度定制

前面的BasicQASkill只是一个起点。实战中的Skill需要更强大。

• 工具调用（Tool Calling）：让Agent不仅能说，还能做。你可以封装一个Skill，使其能执行Shell命令、查询数据库、调用第三方API（如GitHub API、天气API）。Council的SkillContext可以传递参数，Skill内部可以解析这些参数并执行具体操作。

  class SQLQuerySkill(SkillBase): def __init__(self, db_connection): self.db = db_connection super().__init__() def execute(self, context: SkillContext): query = self._parse_query_from_context(context) # 从上下文或LLM生成中解析SQL result = self.db.execute(query) return ChatMessage.agent(f"查询结果：{result}")

• 复杂提示词与上下文管理：Skill可以维护复杂的对话历史，实现多轮交互。例如，一个“调试助手”Skill可以记住之前尝试过的解决方案，避免重复。

• 流式输出（Streaming）：对于生成长篇内容（如代码、报告）的Skill，可以支持流式输出，提升用户体验。这需要与支持流式响应的LLM客户端结合。

4.2 控制流（Controller）的多样化选择

LLMController功能强大但成本较高（每次路由都需调用LLM）。Council提供了其他选择：

• BasicController：简单可靠，按固定顺序或并行执行Chain。适合流程确定、无需动态决策的场景。
• FilterController：先让所有Chain并行执行，然后通过Evaluator对结果进行过滤和排序，只返回最优的。这适用于“海选”场景，但资源消耗较大。
• 自定义Controller：你可以继承ControllerBase实现任何你想要的调度逻辑。例如，一个基于规则的控制器：如果问题包含“error”，则路由到“调试专家”；如果包含“how to”，则路由到“教程生成专家”。

4.3 评估器（Evaluator）的设计艺术

评估器是保证输出质量的关键。除了用另一个LLM做评估，还有很多轻量级方法：

• 基于规则的评估：检查输出长度、是否包含禁止词汇、是否符合特定格式（如JSON）。
• 交叉验证（Cross-Validation）：让两个不同的Agent回答同一问题，然后比较答案的一致性，或者让第三个“裁判”Agent评判哪个更好。
• 事实核查（Fact-Checking）：对于涉及事实的答案，可以调用知识库检索API进行验证。
• 成本与质量权衡：设计一个综合评估函数，同时考虑LLM的置信度分数、回答长度（避免过于简短）、Token消耗成本，实现性价比最优。

4.4 预算（Budget）管理与成本控制

在商业应用中，成本控制至关重要。Council的Budget对象可以绑定到整个执行上下文（AgentContext）。

• 全局预算：为一次用户会话设置总Token上限。
• 链式预算：可以为某个特定的Chain或Skill设置子预算，防止某个环节消耗过多资源。
• 预算监控与熔断：在Agent执行过程中，实时检查context.budget.remaining，如果剩余预算不足，可以提前终止非关键任务或返回降级结果。

  def execute(self, context: SkillContext): if context.budget.remaining < 50: return ChatMessage.agent("预算不足，无法完成完整分析。以下是简要回答：...") # ... 正常执行

4.5 错误处理与系统韧性

多Agent系统环节多，出错概率也高。必须构建健壮的错误处理机制。

1. Agent/Skill级别容错：每个Skill的execute方法都应该用try...except包裹，返回一个友好的错误消息，而不是让整个系统崩溃。
2. 降级策略：当某个专家Agent失败时，Controller应能感知并路由到备用的通用Agent。
3. 超时控制：对于调用外部API或执行耗时操作的Skill，必须设置超时，避免整个请求被挂起。
4. 日志与监控：详细记录每个Agent的输入、输出、耗时、Token消耗和错误信息。这对于调试和优化系统性能不可或缺。可以集成像Prometheus和Grafana这样的监控工具。

5. 常见问题与排查实录

在实际开发和部署Council应用时，我踩过不少坑。这里总结几个最常见的问题和解决思路。

5.1 LLM路由不准或响应格式错误

• 症状：RouterController总是选择错误的Chain，或者返回的无法被_parse_llm_response解析。
• 排查：
  1. 检查提示词：这是最常见的原因。确保你的路由提示词指令清晰、无歧义。使用“只返回数字”、“不要有任何其他文本”等强约束。可以提供几个“输入-输出”的例子（Few-shot Learning）来引导LLM。
  2. 降低top_k：在LLMController初始化时，设置top_k=1，强制LLM只返回一个最可能的选择。
  3. 调整温度（Temperature）：如果使用的是可以配置参数的LLM客户端，将温度设为0（或较低值），使输出更确定、更可预测。
  4. 手动测试：将你的提示词和用户问题单独提出来，用OpenAI Playground或类似工具测试，观察LLM的原始输出是否符合预期。

5.2 执行速度慢，延迟高

• 症状：处理一个简单问题需要十几秒甚至更久。
• 排查：
  1. 串行与并行：检查你的Controller和Chain配置。如果多个Agent/Chain之间没有依赖关系，使用BasicController的并行模式，或者使用FilterController，可以显著缩短总耗时。
  2. LLM响应慢：考虑更换为更快的模型（如从GPT-4切回GPT-3.5-Turbo），或者为LLM调用设置合理的超时时间。
  3. Skill优化：检查自定义Skill中是否有同步的、耗时的I/O操作（如网络请求、大文件读取）。考虑将其异步化，或增加缓存机制。
  4. 上下文长度：传递的上下文（对话历史）是否过长？过长的上下文会增加LLM处理时间和Token消耗。实现一个智能的上下文窗口或摘要机制。

5.3 Token消耗超出预算或成本失控

• 症状：账单激增，或者频繁触发预算不足错误。
• 排查与优化：
  1. 精细化预算设置：不要只设一个全局总预算。为不同的Chain和Skill设置更细粒度的子预算。对于成本高的“重量级”专家（如调用GPT-4的Agent），给予更严格的预算。
  2. 压缩上下文：在将历史消息传递给LLM前，使用一个轻量级模型（如gpt-3.5-turbo）对长上下文进行摘要，只保留关键信息。
  3. 缓存结果：对于常见、重复的问题（例如“Python怎么安装包？”），可以将问答对缓存起来（使用内存缓存如Redis，或磁盘缓存），下次直接返回缓存结果，避免调用LLM。
  4. 评估成本效益：并非所有环节都需要最强模型。对于路由、简单分类等任务，完全可以使用小模型或规则系统。将昂贵的LLM调用用在最需要创造性和复杂推理的环节。

5.4 多Agent协作产生混乱或低质输出

• 症状：最终答案质量不高，或者多个Agent的输出相互矛盾。
• 排查与改进：
  1. 强化评估器（Evaluator）：不要仅仅依赖最后一个Agent的输出。引入一个强大的“评审员”Evaluator（可以使用更强的LLM，如GPT-4），对所有候选答案进行打分和排序，甚至进行改写和融合。
  2. 设计协作流程：对于复杂任务，不要简单并行。设计顺序链，让后一个Agent基于前一个Agent的成果进行深化。例如，“调研Agent -> 大纲Agent -> 写作Agent -> 润色Agent”。
  3. 明确角色与边界：在定义每个Agent的System Prompt时，务必清晰界定其职责和边界，避免越界回答。例如，明确告诉“代码专家”：“你只负责提供代码片段和解释，不负责提供商业建议”。
  4. 人工反馈循环（Human-in-the-loop）：在关键决策点引入人工审核。例如，让Controller在决定采用哪个方案前，将选项提交给人做最终裁定。这在落地高风险业务场景时非常有效。

Council框架为我们提供了一个强大的工具箱，将多智能体系统的构想变得工程化、可实施。它最大的魅力不在于解决了某个具体问题，而在于提供了一套范式，让我们可以像搭积木一样，将不同的AI能力组合起来，去应对那些单一模型无法解决的复杂挑战。从简单的路由问答到复杂的多轮辩论与决策，其可能性取决于你的设计和创意。