Multi-Agent Hackathon 2025实战指南：用OpenAgents构建智能协作系统

创新无界，传播有力！这是我第一次参加Multi-Agent Hackathon，想和大家分享一下用OpenAgents 构建智能协作系统的实战经验。
在这里插入图片描述

写在前面：为什么要搞多智能体？

老实说，刚开始接触多智能体（Multi-Agent）的时候，我心里也犯嘀咕：一个智能体能干好的事，非得整一群，这不是把简单问题复杂化吗？
30天彻底颠覆了我的认知。看着参赛群里的开发者像变魔术一样，用一堆Agent搭出了几百个真实可用的应用，我才明白：单打独斗是加法，团队协作才是乘法。

我来把我的项目跟大家分享一下

一、OpenAgents：给AI智能体连上网线

1.1 它到底是个啥？

现在的多智能体系统，很多都有点"自闭"：

  • 用完即走：任务做完就散伙，跟临时工似的。
  • 信息孤岛：各干各的，经验没法共享。
  • 鱼的记忆：重启一下，啥都忘了。

OpenAgents 就是为了解决这些痛点生的。它更像是一个AI界的社交网络，让智能体们能长期在线、互相加好友、组队打怪，甚至还能跨项目分享经验。

1.2 跟其他框架比，强在哪？

OpenAgents
图：OpenAgents仪表盘图

1.3 核心模块（说人话版）

  • Agent Network（地盘）：
    这就是智能体的活动空间。你建个Network，它就一直在线。

    # 开个房，24小时不打烊
network = Network(name="MyClub", protocols=["websocket", "http"])

  • Mods（外挂）：
    想让智能体干啥，就给它装啥Mod。想写文档？装Documents。想玩游戏？装Minecraft。

    network.add_mod("Documents")  # 大家好，我是文档编辑员

  • 协议（语言）：
    不管你的Agent是用Python写的，还是Java写的，只要连上协议，就能在一个群里聊天。

  • 记忆（脑子）：
    集成了Milvus向量数据库。你跟它说过的话，它都能记住，不用像复读机一样重复。

1.4 三分钟极速上手

举办方OpenAgents贴心的为大家准备了demo和教程：如何报名获取支持
在这里插入图片描述

别整那些虚的，直接跑起来先看效果：
在这里插入图片描述

推荐用pip安装（最快）：

conda create -n openagents python=3.12 && conda activate openagents
pip install openagents

建个新网络：

openagents init ./my_first_network
openagents network start ./my_first_network

搞定！现在打开浏览器访问 localhost:8700，首次打开选择管理员登录，并选择模板，因为我是多Agent所以选择了第二个multi-agent-chartroom
在这里插入图片描述
需要点击管理网络，选择管理员登录，可以设置自己的初始密码，要8位，这个密码要记住
在这里插入图片描述
这里有个坑需要注意，在启动前要初始化一个
写个最简单的Agent：

# 这是一个会说"你好"的复读机Agent
class HelloAgent(WorkerAgent):
    async def process_message(self, message):
        return f"收到！你说的是：{message.content}"

二、多智能体怎么设计才好用？

2.1 别为了多智能体而多智能体

多智能体系统的核心逻辑其实就一句话：专业的人干专业的事。

  • 分工：让写代码的去写代码，画图的去画图。
  • 容错：一个倒下了，另一个顶上，不至于全盘崩溃。

2.2 几种好用的队形

在Hackathon里，我见过几种比较稳的架构：

  1. 流水线模式（最稳）：
    A做完给B，B做完给C。比如：搜集资料 -> 写初稿 -> 润色。适合流程固定的任务。

  2. PK模式（最有创意）：
    三个Agent同时写文案，最后选个最好的。适合需要脑洞的场景。

  3. 老板模式（最复杂）：
    一个"老板"Agent负责拆解任务、分派活儿，底下几个"打工"Agent干活。适合复杂的大项目。

三、实战：手把手带你做一个"营销文案生成器"

3.1 灵感来源

Hackathon时间紧，选题很关键。我当时想：电商运营每天都要写一堆文案，还得适配小红书、朋友圈、淘宝不同风格，太累了。能不能搞个全自动营销编辑部？

3.2 架构设计

我就设计了这么一条流水线：

  1. 解析Agent：不管你扔进来的是京东链接还是产品图，先把它"翻译"成标准的产品档案。
  2. 文案天团（3人组）：
    • 电商标题Agent：负责写那种堆关键词的淘宝标题。
    • 社群Agent：负责写"家人们谁懂啊"这种朋友圈文案。
    • 种草Agent：负责写带Emoji的小红书笔记。
  3. 热词Agent：最后把关，把文案里的老土词换成今天的热搜词（比如把"很好用"换成"绝绝子"）。
  4. 投放Agent：假装发布出去，并回传数据。

3.3 代码秀（部分核心）

这也是基于OpenAgents写的最核心部分，大家感受一下：

1. 解析Agent（情报官）

class ParseAgent(MarketingAgent):
    def process_request(self, material: str) -> str:
        # 假装在解析，其实是写死的逻辑（Hackathon常用套路）
        if "http" in material:
            return "发现链接！提取到：降噪耳机Pro，价格299..."
        return "这是个图片，看起来像个耳机..."

2. 种草Agent（文案手）

class NoteAgent(CopywritingAgent):
    def generate_copy(self, facts: str) -> str:
        return f"""
        姐妹们！挖掘到一款宝藏耳机🎧！
        {facts}
        真的太香了，通勤党必备！#好物推荐
        """

3. 热词Agent（注入灵魂）

class HotwordAgent(MarketingAgent):
    def process_request(self, copy: str) -> str:
        # 简单的字符串替换，但效果拔群
        optimized = copy.replace("很好", "绝绝子").replace("便宜", "白菜价")
        return f"✅ 注入灵魂完成：\n{optimized}"

(完整代码太长，放在文末附录了，想跑的可以直接粘)

四、那些年我踩过的坑（血泪史）

4.1 别让系统卡成PPT

刚开始我全用的同步代码（time.sleep），结果一个Agent卡住，整个系统都瘫痪了。
划重点：一定要用 async/await！ 异步才是多智能体的灵魂。

4.2 第一次演示就翻车

第一次给评委演示，Agent直接报错退出了，场面一度十分尴尬。
教训：

  1. Try-Catch保平安：无论如何别让进程挂掉，哪怕返回个"我出错了"也比崩溃强。
  2. 重试机制：网络请求失败了？自动重试3次，往往能救命。

4.3 那些让人头秃的"玄学"问题

除了代码逻辑，环境和配置也是大坑，这里列几个我遇到的真实案例，大家避个雷：

  • 服务死活打不开：
    启动了 network start 但浏览器访问 localhost:8700 一直转圈？

    • 原因：多半是端口被占用了，或者Docker容器没完全起来。
    • 解决：lsof -i:8700 查一下，杀掉占用的进程，或者重启大法好。

  • 模版选不了：
    在Studio里创建Agent时，点击模版没反应？

    • 原因：只有初次启动才会有选择模板页面。
    • 解决：停掉所有运行中的network，重新初始化工作区启动studio。

  • 聊着聊着没额度了：
    跑着跑着突然报错 429 Too Many Requests 或者 Quota Exceeded。

    • 原因：多智能体交互的时候，Token消耗是指数级的！几个Agent互相夸几句，几千Token就没了,我以为是挂了，结果去查了下原来是token被耗光了。
    • 解决：一定要在代码里做限流，或者申请个额度大的Key。
      在这里插入图片描述

  • 智能体陷入"死循环"：
    这是最恐怖的。A发消息给B，B回复A，A又回复B… 两人聊了一晚上，第二天一看Token耗尽，钱包空了。

    • 原因：没设置停止条件（Stop Condition）。
    • 解决：在Agent配置里把 react_to_all_messages 设为 false，或者在代码里加个判断：if sender == self.name: return（别自己回自己）。

4.4 日志！日志！日志！

调试多智能体最痛苦的就是：你不知道哪一环断了。
建议：给每个Agent都加上详细的日志，最好把输入输出都打出来，排查问题快十倍。

五、最后聊两句

参加Hackathon最爽的不是拿奖，而是那几天几夜和队友一起死磕代码、解决问题的过程。OpenAgents 给了我们一个很好的底座，让我们能站在巨人的肩膀上搞事情。

无论你是想用代码改变世界，还是单纯想来交个朋友，Multi-Agent Hackathon 2025 都是个绝佳的机会。

参考资源

附录：完整代码示例

为了方便大家学习,我把完整的营销文案生成系统代码整理在这里,可以直接复制使用。

"""
营销文案生成系统 - 基于OpenAgents的完整实现
作者：[你的名字]
日期：2025年
"""

import asyncio
import logging
import random
from typing import Dict, Optional, List
from openagents.agents.worker_agent import WorkerAgent
from openagents.models.agent_config import AgentConfig
from openagents.models.network_config import NetworkConfig


class MarketingAgent(WorkerAgent):
    """营销智能体基类"""
    
    def __init__(self, config: AgentConfig, knowledge_base: Dict = None):
        super().__init__(config)
        self.knowledge_base = knowledge_base or {}
        self.name = config.agent_id
        
        # 配置日志
        self.logger = logging.getLogger(self.name)
        self.logger.setLevel(logging.INFO)
        handler = logging.StreamHandler()
        handler.setFormatter(logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        ))
        self.logger.addHandler(handler)
    
    async def process_message(self, message):
        """处理消息"""
        content = message.content
        self.logger.info(f"收到请求: {content}")
        
        try:
            response = self.process_request(content)
            self.logger.info(f"返回响应: {response}")
            return response
        except Exception as e:
            self.logger.error(f"处理失败: {str(e)}", exc_info=True)
            return "抱歉,系统出现异常,请稍后再试"
    
    def process_request(self, user_input: str) -> str:
        """处理请求,子类需要实现"""
        raise NotImplementedError("子类必须实现此方法")
    
    def query_knowledge_base(self, query: str) -> str:
        """查知识库"""
        for key, value in self.knowledge_base.items():
            if key in query:
                return value
        return "抱歉,我没有找到相关信息"


class ParseAgent(MarketingAgent):
    """素材解析专家"""
    
    def __init__(self, config: AgentConfig):
        super().__init__(config)
        self.fact_list = []
    
    def process_request(self, material: str) -> str:
        """解析营销素材,输出事实清单"""
        self.logger.info(f"开始解析素材: {material}")
        
        # 模拟解析过程
        if material.startswith("http"):
            facts = self.parse_link(material)
        elif material.endswith((".jpg", ".png", ".jpeg")):
            facts = self.parse_image(material)
        else:
            facts = self.parse_text(material)
        
        self.fact_list = facts
        result = f"✅ 素材解析完成\n事实清单:\n" + "\n".join([f"  • {fact}" for fact in facts])
        self.logger.info(f"解析完成,共提取{len(facts)}个事实")
        return result
    
    def parse_link(self, link: str) -> List[str]:
        """解析链接"""
        return [
            "商品名称: 智能降噪耳机Pro",
            "价格: 299元",
            "核心卖点: 40dB深度降噪,30小时续航",
            "目标人群: 18-35岁都市白领",
            "适用场景: 通勤、办公、学习"
        ]
    
    def parse_image(self, image_path: str) -> List[str]:
        """解析图片"""
        return [
            "产品类型: 电子产品",
            "视觉风格: 简约科技风",
            "主色调: 黑色+蓝色",
            "展示场景: 现代办公环境",
            "突出功能: 降噪、舒适佩戴"
        ]
    
    def parse_text(self, text: str) -> List[str]:
        """解析文本"""
        return [
            "产品类别: 智能耳机",
            "品牌定位: 中高端",
            "价格区间: 200-500元",
            "核心优势: 性价比高"
        ]


class CopywritingAgent(MarketingAgent):
    """文案生成专家基类"""
    
    def __init__(self, config: AgentConfig, copy_type: str):
        super().__init__(config)
        self.copy_type = copy_type
    
    def process_request(self, fact_list: List[str]) -> str:
        """根据事实清单生成文案"""
        facts = "\n".join(fact_list)
        self.logger.info(f"开始生成{self.copy_type}文案")
        copy = self.generate_copy(facts)
        self.logger.info(f"文案生成完成")
        return copy
    
    def generate_copy(self, facts: str) -> str:
        """生成文案,子类实现"""
        raise NotImplementedError("子类必须实现此方法")


class EcommerceTitleAgent(CopywritingAgent):
    """电商标题生成专家"""
    
    def __init__(self, config: AgentConfig):
        super().__init__(config, "ecommerce_title")
    
    def generate_copy(self, facts: str) -> str:
        """生成电商标题"""
        return f"""📦 电商标题:
【2024新品】智能降噪耳机Pro 40dB深度降噪 30小时超长续航 蓝牙5.3 无线耳机 适用于通勤办公学习

优化建议:
• 包含核心关键词:智能降噪、40dB、30小时续航
• 突出卖点:新品、蓝牙5.3
• 覆盖场景:通勤、办公、学习
• 字数适中,便于搜索"""


class CommunityTalkAgent(CopywritingAgent):
    """社群话术生成专家"""
    
    def __init__(self, config: AgentConfig):
        super().__init__(config, "community_talk")
    
    def generate_copy(self, facts: str) -> str:
        """生成社群话术"""
        return f"""💬 社群推广话术:
【开场白】
"家人们!今天给大家推荐一款通勤必备神器🎧"

【产品介绍】
"这款智能降噪耳机Pro真的太香了!40dB深度降噪,地铁上完全听不到噪音,30小时续航,一周充一次电就够了"

【互动引导】
"有没有通勤党的小伙伴?平时坐地铁是不是很吵?评论区告诉我,抽3位送小礼物哦~"

【转化话术】
"现在下单立减50元,只要249元!链接在下方,手慢无!"

话术特点:
• 亲切自然的语气,符合社群氛围
• 场景化描述,引发共鸣
• 互动引导,提升参与度
• 限时优惠,促进转化"""


class NoteAgent(CopywritingAgent):
    """种草笔记生成专家"""
    
    def __init__(self, config: AgentConfig):
        super().__init__(config, "note")
    
    def generate_copy(self, facts: str) -> str:
        """生成种草笔记"""
        return f"""📝 小红书/抖音种草笔记:

标题:通勤党必看!这款降噪耳机让我地铁通勤变成私人影院✨

正文:
姐妹们!终于找到通勤救星了!🎉

作为一个每天地铁通勤1小时的打工人,地铁上的噪音真的让我崩溃...直到我遇到了这款智能降噪耳机Pro!

✨40dB深度降噪
一戴上,地铁的轰鸣声瞬间消失,整个世界都安静了!真的像在私人影院一样沉浸!

🔋30小时超长续航
充一次电用一周,完全不用焦虑电量!出差旅行带着也超安心~

🎧佩戴超舒适
轻量化设计,戴一整天耳朵也不疼,耳罩材质很柔软

💰性价比超高
只要299元,比大牌便宜一半,效果完全不输!

真心推荐给所有通勤党、学生党,早八人的必备神器!冲鸭!🚀

#通勤神器 #降噪耳机 #打工人必备 #好物推荐 #学生党

笔记亮点:
• 真实场景描述,引发共鸣
• 使用emoji增加亲和力
• 突出核心卖点和使用感受
• 合理使用话题标签,提升曝光"""


class HotwordAgent(MarketingAgent):
    """热词优化专家"""
    
    def __init__(self, config: AgentConfig):
        super().__init__(config)
        self.hotwords = {
            "降噪": ["深度降噪", "主动降噪", "ANC降噪"],
            "续航": ["超长续航", "持久续航", "长续航"],
            "新品": ["2024新品", "全新上市", "首发"]
        }
        self.ctr_predictions = {}
    
    def process_request(self, copy: str) -> str:
        """优化文案中的热词,预测CTR"""
        self.logger.info("开始热词优化")
        optimized_copy = self.replace_hotwords(copy)
        ctr = self.predict_ctr(optimized_copy)
        self.logger.info(f"热词优化完成,预测CTR: {ctr:.2%}")
        return f"✅ 热词优化完成\n{optimized_copy}\n\n📊 预测CTR: {ctr:.2%}"
    
    def replace_hotwords(self, copy: str) -> str:
        """替换文案中的热词"""
        import random
        optimized = copy
        
        for keyword, alternatives in self.hotwords.items():
            if keyword in optimized:
                # 随机选择一个热词替换
                new_word = random.choice(alternatives)
                optimized = optimized.replace(keyword, new_word, 1)
        
        return optimized
    
    def predict_ctr(self, copy: str) -> float:
        """预测点击率(CTR)"""
        base_ctr = 0.05
        
        # 热词加分
        for alternatives in self.hotwords.values():
            for word in alternatives:
                if word in copy:
                    base_ctr += 0.01
        
        # 长度影响
        if 50 <= len(copy) <= 200:
            base_ctr += 0.02
        
        return min(base_ctr, 0.15)


class DeliveryAgent(MarketingAgent):
    """投放数据专家"""
    
    def __init__(self, config: AgentConfig):
        super().__init__(config)
        self.delivery_data = []
    
    def process_request(self, copy: str) -> str:
        """调用投放接口,回传数据"""
        # 模拟调用投放接口
        delivery_id = self.call_delivery_api(copy)
        
        # 模拟回传数据
        metrics = self.get_delivery_metrics(delivery_id)
        
        return f"✅ 投放成功\n投放ID: {delivery_id}\n\n📈 实时数据:\n{metrics}"
    
    def call_delivery_api(self, copy: str) -> str:
        """调用投放API"""
        import time
        delivery_id = f"DLV{int(time.time())}"
        self.logger.info(f"调用投放接口: {delivery_id}")
        self.logger.info(f"文案内容: {copy[:50]}...")
        return delivery_id
    
    def get_delivery_metrics(self, delivery_id: str) -> str:
        """获取投放数据"""
        metrics = {
            "曝光量": random.randint(10000, 50000),
            "点击量": random.randint(500, 2000),
            "点击率": f"{random.uniform(3, 8):.2f}%",
            "转化量": random.randint(50, 200),
            "转化率": f"{random.uniform(2, 6):.2f}%"
        }
        
        return "\n".join([f"  • {k}: {v}" for k, v in metrics.items()])


class MockMessage:
    """模拟消息"""
    def __init__(self, content):
        self.content = content


async def build_marketing_system():
    """构建完整系统"""
    
    # 创建网络配置
    network_config = NetworkConfig(
        network_id="marketing-system",
        name="营销文案生成系统",
        protocols=["websocket"]
    )
    
    # 创建智能体配置
    parse_config = AgentConfig(
        agent_id="ParseAgent",
        network_id="marketing-system"
    )
    
    ecommerce_config = AgentConfig(
        agent_id="EcommerceTitleAgent",
        network_id="marketing-system"
    )
    
    community_config = AgentConfig(
        agent_id="CommunityTalkAgent",
        network_id="marketing-system"
    )
    
    note_config = AgentConfig(
        agent_id="NoteAgent",
        network_id="marketing-system"
    )
    
    hotword_config = AgentConfig(
        agent_id="HotwordAgent",
        network_id="marketing-system"
    )
    
    delivery_config = AgentConfig(
        agent_id="DeliveryAgent",
        network_id="marketing-system"
    )
    
    # 创建智能体实例
    parse_agent = ParseAgent(parse_config)
    ecommerce_agent = EcommerceTitleAgent(ecommerce_config)
    community_agent = CommunityTalkAgent(community_config)
    note_agent = NoteAgent(note_config)
    hotword_agent = HotwordAgent(hotword_config)
    delivery_agent = DeliveryAgent(delivery_config)
    
    # 连接到网络
    await parse_agent.connect()
    await ecommerce_agent.connect()
    await community_agent.connect()
    await note_agent.connect()
    await hotword_agent.connect()
    await delivery_agent.connect()
    
    return {
        "parse": parse_agent,
        "ecommerce": ecommerce_agent,
        "community": community_agent,
        "note": note_agent,
        "hotword": hotword_agent,
        "delivery": delivery_agent
    }


async def test_system():
    """测试营销文案系统"""
    print("=" * 60)
    print("营销文案生成系统测试")
    print("=" * 60)
    
    agents = await build_marketing_system()
    
    # 测试素材
    test_material = "https://example.com/product/noise-cancelling-headphones"
    
    print(f"\n{'=' * 60}")
    print(f"输入素材: {test_material}")
    print(f"{'=' * 60}")
    
    # 步骤1: 解析素材
    print("\n【步骤1】解析素材")
    parse_result = await agents["parse"].process_message(MockMessage(test_material))
    print(parse_result)
    
    # 提取事实清单
    fact_list = agents["parse"].fact_list
    
    # 步骤2: 并行生成文案
    print(f"\n{'=' * 60}")
    print("【步骤2】并行生成文案")
    print(f"{'=' * 60}")
    
    # 电商标题
    print("\n>>> 电商标题")
    ecommerce_copy = agents["ecommerce"].process_request(fact_list)
    print(ecommerce_copy)
    
    # 社群话术
    print("\n>>> 社群话术")
    community_copy = agents["community"].process_request(fact_list)
    print(community_copy)
    
    # 种草笔记
    print("\n>>> 种草笔记")
    note_copy = agents["note"].process_request(fact_list)
    print(note_copy)
    
    # 步骤3: 热词优化
    print(f"\n{'=' * 60}")
    print("【步骤3】热词优化")
    print(f"{'=' * 60}")
    
    optimized_ecommerce = agents["hotword"].process_request(ecommerce_copy)
    print("\n>>> 电商标题优化:")
    print(optimized_ecommerce)
    
    # 步骤4: 投放
    print(f"\n{'=' * 60}")
    print("【步骤4】投放")
    print(f"{'=' * 60}")
    
    delivery_result = agents["delivery"].process_request(optimized_ecommerce)
    print(delivery_result)
    
    print(f"\n{'=' * 60}")
    print("测试完成!")
    print("=" * 60)


if __name__ == "__main__":
    asyncio.run(test_system())
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

AI智能外呼系统如何让企业在拓客方面取得实效？—2026年十大电销机器人深度测评

avatar AI Agent技术社区
cover

靠谱的 ai智能外呼电话机器人推荐

avatar AI Agent技术社区

AI Agent 的能力从哪里来？一文讲清后训练、上下文学习和外部能力

当我们看到一个 Agent 表现得越来越“聪明”时，不妨换一个角度去理解它。有些能力来自模型长期训练形成的基本功；有些能力来自当前上下文中的规则、示例和任务约束；还有一些能力来自运行时连接的知识库、工具和业务系统。从工程视角来看，现代 Agent 已经不再只是一个大模型，而是一套围绕大模型构建的系统。模型负责推理，Context 负责任务适配，外部系统负责提供知识、状态与执行能力。真正决定 Age

avatar AI Agent技术社区