在多 Agent 系统愈发复杂的今天，如何构建一个轻量级、高扩展性的分布式架构成为技术焦点。BettaFish（微舆）项目提供了一个引人注目的工程实现：从零开始构建的多 Agent 舆情分析系统，不依赖任何外部框架，通过纯 Python 模块化设计实现了分布式协作。这种 "逆向工程" 的方法论为多 Agent 系统设计提供了宝贵经验。
分布式架构的核心设计哲学
BettaFish 最显著的技术特征是拒绝框架依赖，这迫使开发者从底层思考分布式系统的本质问题。系统采用分层模块化架构，将功能明确的 Agent 组件分布在独立的工作目录中：QueryEngine/、MediaEngine/、InsightEngine/、ReportEngine/。每个模块都拥有独立的agent.py主逻辑、tools/工具集、utils/工具函数，形成微服务化的功能解耦。
这种设计避免了复杂框架的学习成本和运行时开销。开发者可以通过简单的文件操作启动、停止或重启特定 Agent，而无需处理容器化、进程管理或服务发现等复杂问题。更重要的是，每个 Agent 都可以独立部署在不同服务器上，通过标准的 HTTP 接口进行通信，为真正的分布式部署提供了基础。
Agent 间通信的 "论坛" 协议设计
BettaFish 最具创新性的设计是 ForumEngine 论坛协作引擎，它实现了 Agent 间的协作通信协议。传统的多 Agent 系统往往依赖复杂的消息队列或事件系统，而 BettaFish 采用了类似人类论坛的通信模式。
论坛协议的实现机制
ForumEngine 通过monitor.py和llm_host.py实现论坛管理机制。每个 Agent 产生的信息不是直接传递给其他 Agent，而是发布到虚拟论坛中。论坛主持人 LLM（通常是一个单独的对话模型）负责监控讨论并生成主持人总结，类似于讨论版的管理员角色。
# Agent间论坛通信简化示例
class ForumReader:
def post_to_forum(self, agent_id, content, topic):
"""Agent发布内容到论坛"""
forum_entry = {
'agent_id': agent_id,
'content': content,
'timestamp': time.time(),
'topic': topic
}
self.save_to_forum(forum_entry)
def get_forum_discussion(self, topic):
"""Agent读取论坛讨论内容"""
return self.fetch_forum_content(topic)
这种设计有几个技术优势。首先，它自然实现了消息的持久化和可追溯性。其次，论坛模式提供了天然的会话边界，避免了 Agent 间的直接耦合。最后，主持人 LLM 的存在可以将无结构的对话转化为结构化的讨论摘要，减少信息过载。
协作循环的工程实现
BettaFish 的协作循环是系统的核心工作流程。一次完整的分析包含多个循环阶段，每个 Agent 基于论坛主持人的引导进行专项搜索和反思。循环控制逻辑通过状态机实现：
# 循环协作的简化实现
class CollaborativeLoop:
def __init__(self):
self.max_iterations = 5
self.current_iteration = 0
def execute_cycle(self, agents, forum_engine):
while self.current_iteration < self.max_iterations:
# 阶段5.1: 深度研究
agent_results = {}
for agent in agents:
agent_results[agent.id] = agent.deep_research(
forum_engine.get_last_discussion()
)
# 阶段5.2: 论坛协作
forum_summary = forum_engine.host_discussion(
agent_results, self.current_iteration
)
# 阶段5.3: 交流融合
for agent in agents:
agent.adjust_strategy(forum_summary)
self.current_iteration += 1
# 最终结果整合
return self.integrate_results(agent_results)
这种循环设计通过 "辩证式讨论" 避免了单一模型的思维局限。每个 Agent 都会根据其他 Agent 的观点调整自己的分析方向，从而产生集体智能效应。
负载均衡策略的工程实现
在分布式系统中，负载均衡不仅关乎计算资源的分配，更涉及任务调度的优化。BettaFish 通过几个层面的策略实现了有效的负载均衡。
并行启动与任务分配
系统启动时，Flask 主应用接收到查询后，会并行启动三个核心 Agent：Query Agent、Media Agent、Insight Agent。这种并行启动机制避免了串行处理的等待时间，充分利用了多核计算资源。
# 并行启动的简化实现
import threading
import asyncio
class ParallelLauncher:
def launch_agents_concurrently(self, query):
agents = [
QueryEngine(query),
MediaEngine(query),
InsightEngine(query)
]
# 线程池并行启动
threads = []
results = {}
for i, agent in enumerate(agents):
thread = threading.Thread(
target=self._execute_agent,
args=(agent, results, i)
)
threads.append(thread)
thread.start()
# 等待所有Agent完成
for thread in threads:
thread.join()
return results
动态负载管理
每个 Agent 内部的负载管理通过nodes/目录下的不同处理节点实现。这些节点可以独立扩展和调优，比如在InsightEngine中：
# InsightEngine的节点配置
NODES_CONFIG = {
'search_node': {
'max_search_results': 50,
'batch_size': 32,
'parallel_requests': 5
},
'summary_node': {
'max_content_length': 8000,
'comprehensive_search_limit': 10
},
'formatting_node': {
'confidence_threshold': 0.8
}
}
通过这种配置化设计，系统可以根据实际负载情况动态调整每个处理节点的参数，实现细粒度的负载均衡。
系统扩展性的技术考量
BettaFish 的扩展性设计体现在多个层面，从代码结构到部署模式都考虑了未来扩展的需求。
模块化扩展接口
系统提供了清晰的扩展接口，开发者可以通过修改配置文件轻松集成新的 LLM 模型或数据源。比如在.env文件中：
# Insight Agent配置示例
INSIGHT_ENGINE_API_KEY="your_api_key"
INSIGHT_ENGINE_BASE_URL="https://api.moonshot.cn/v1"
INSIGHT_ENGINE_MODEL_NAME="kimi-k2-0711-preview"
# 替换为其他LLM提供商
INSIGHT_ENGINE_BASE_URL="https://api.openai.com/v1"
INSIGHT_ENGINE_MODEL_NAME="gpt-4"
这种配置化设计允许开发者在不修改核心代码的情况下更换后端服务提供商，实现真正的 "可插拔" 架构。
轻量化部署策略
系统采用轻量化的部署策略，支持三种启动模式：完整系统启动、单独 Agent 启动、爬虫系统单独使用。这种多样化的部署选择为不同场景提供了灵活性。
# 单独Agent启动示例
def start_single_agent(agent_type, port):
if agent_type == "query":
from QueryEngine.agent import QueryAgent
agent = QueryAgent()
elif agent_type == "media":
from MediaEngine.agent import MediaAgent
agent = MediaAgent()
elif agent_type == "insight":
from InsightEngine.agent import InsightAgent
agent = InsightAgent()
# Streamlit独立部署
subprocess.run([
"streamlit", "run", f"SingleEngineApp/{agent_type}_engine_streamlit_app.py",
"--server.port", str(port)
])
技术挑战与解决方案
Agent 状态管理
多 Agent 系统中，状态一致性是核心挑战。BettaFish 通过state/目录下的状态管理模块解决这一问题：
# InsightEngine的状态定义
class AgentState:
def __init__(self):
self.conversation_history = []
self.search_results = {}
self.current_focus = None
self.iteration_count = 0
def update_state(self, agent_id, content):
self.conversation_history.append({
'agent': agent_id,
'content': content,
'timestamp': time.time()
})
错误恢复机制
系统实现了retry_helper.py工具来处理网络请求失败等常见问题：
# 重试机制的实现
class RetryHelper:
@retry(stop=stop_after_attempt(3), wait=wait_exponential())
def fetch_with_retry(self, url, **kwargs):
try:
response = requests.get(url, **kwargs)
response.raise_for_status()
return response
except (ConnectionError, TimeoutError) as e:
print(f"请求失败，重试中: {e}")
raise
性能优化实践
批处理优化
在情感分析模块中，系统实现了批处理机制来提高处理效率：
# SentimentAnalysisModel的批处理配置
SENTIMENT_CONFIG = {
'batch_size': 32,
'max_sequence_length': 512,
'confidence_threshold': 0.8
}
def batch_analyze(texts, model):
results = []
for i in range(0, len(texts), SENTIMENT_CONFIG['batch_size']):
batch = texts[i:i + SENTIMENT_CONFIG['batch_size']]
batch_results = model.predict(batch)
results.extend(batch_results)
return results
缓存策略
系统通过数据库查询缓存减少了重复计算。对于相同关键词的查询，系统会缓存之前的结果并在有效期内复用：
# 查询缓存的实现
class QueryCache:
def __init__(self, cache_duration=3600): # 1小时缓存
self.cache = {}
self.cache_duration = cache_duration
def get_cached_result(self, query_hash):
if query_hash in self.cache:
result, timestamp = self.cache[query_hash]
if time.time() - timestamp < self.cache_duration:
return result
return None
工程实践的启示
BettaFish 项目展示了在复杂系统设计中 "逆向工程" 方法的价值。通过拒绝依赖外部框架，开发者被迫从根本问题出发思考解决方案，这往往能产生更简洁、更可控的架构设计。
ForumEngine 的论坛协作机制提供了 Agent 间通信的新思路，批处理和缓存策略展示了在多 Agent 系统中进行性能优化的实用技巧。这种从 0 开始的实现方式虽然前期投入较大，但为系统的长期维护和扩展提供了更大的自由度。
在构建分布式 AI 系统时，BettaFish 的实践经验告诉我们：简洁的架构设计、清晰的通信协议、有效的负载均衡策略，以及充分的扩展性考量，是构建可维护、可扩展系统的关键要素。
参考资料:
- BettaFish 项目 GitHub 仓库：https://github.com/666ghj/BettaFish
- Hacker News 讨论帖：https://news.ycombinator.com/item?id=41935647