在多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（通常是一个单独的对话模型）负责监控讨论并生成主持人总结，类似于讨论版的管理员角色。
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:
agent_results = {}
for agent in agents:
agent_results[agent.id] = agent.deep_research(
forum_engine.get_last_discussion()
)
forum_summary = forum_engine.host_discussion(
agent_results, self.current_iteration
)
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()
for thread in threads:
thread.join()
return results
动态负载管理
每个Agent内部的负载管理通过nodes/目录下的不同处理节点实现。这些节点可以独立扩展和调优，比如在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_ENGINE_API_KEY="your_api_key"
INSIGHT_ENGINE_BASE_URL="https://api.moonshot.cn/v1"
INSIGHT_ENGINE_MODEL_NAME="kimi-k2-0711-preview"
INSIGHT_ENGINE_BASE_URL="https://api.openai.com/v1"
INSIGHT_ENGINE_MODEL_NAME="gpt-4"
这种配置化设计允许开发者在不修改核心代码的情况下更换后端服务提供商，实现真正的"可插拔"架构。
轻量化部署策略
系统采用轻量化的部署策略，支持三种启动模式：完整系统启动、单独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()
subprocess.run([
"streamlit", "run", f"SingleEngineApp/{agent_type}_engine_streamlit_app.py",
"--server.port", str(port)
])
技术挑战与解决方案
Agent状态管理
多Agent系统中，状态一致性是核心挑战。BettaFish通过state/目录下的状态管理模块解决这一问题：
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
性能优化实践
批处理优化
在情感分析模块中，系统实现了批处理机制来提高处理效率：
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):
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的实践经验告诉我们：简洁的架构设计、清晰的通信协议、有效的负载均衡策略，以及充分的扩展性考量，是构建可维护、可扩展系统的关键要素。
参考资料: