前言

AI Agent 技术正在彻底改变我们的工作方式。从简单的任务自动化到复杂的多角色协作,智能体已经展现出了惊人的潜力。我们团队最近一直在探索如何利用 AI Agent 提升技术内容创作效率,前后尝试了多个框架和方案。

在构建多智能体系统的过程中,我们遇到了一系列棘手的问题:

  • 模型适配复杂:不同角色的智能体需要不同能力的模型,每个模型都有自己的 API 规范和参数格式
  • 网络稳定性差:海外模型经常超时或断连,导致智能体执行中断,任务失败
  • 成本难以控制:多个智能体同时运行,API 调用量呈指数级增长,费用居高不下
  • 调试困难:缺乏统一的日志和监控工具,很难定位哪个智能体出了问题
  • 扩展性差:想添加新的智能体角色或切换模型,需要修改大量代码

在尝试了 LangChain、AutoGPT 等多个框架后,我们发现这些框架虽然功能强大,但都没有解决最底层的 API 接入问题。直到我们基于 4sapi 重构了整个系统,所有问题才迎刃而解。本文将详细分享如何使用 4sapi 构建一个完整的多智能体协作系统,实现技术博客的全自动生成。

一、为什么 4sapi 是多智能体系统的最佳底座

多智能体系统对 AI API 的要求远高于普通应用,它需要同时满足多模型支持、高稳定性、低延迟、低成本和可观测性等多个维度的需求。4sapi 在这些方面都表现得尤为出色:

1.1 一个接口调用所有模型

4sapi 100% 兼容 OpenAI API 协议,支持全球 26 款主流 AI 模型。这意味着我们可以为每个智能体角色分配最适合的模型,而不需要为每个模型编写单独的适配代码。

1.2 企业级稳定性保障

4sapi 采用多区域多节点部署,提供 99.99% 的可用性承诺。内置的故障转移和自动重试机制,可以确保智能体任务不会因为单个节点故障而中断。在我们两个月的测试中,没有出现过一次任务失败的情况。

1.3 智能路由大幅降低成本

4sapi 的智能路由功能可以根据任务复杂度自动选择最优模型。简单任务用轻量级模型,复杂任务才调用高性能模型。这一项功能就为我们节省了近 45% 的 API 调用成本。

1.4 完善的可观测性

4sapi 控制台提供了详细的调用日志和统计数据,我们可以实时查看每个智能体的调用次数、Token 消耗、响应时间和成功率。这对于调试和优化多智能体系统至关重要。

二、系统设计:自动化技术博客生成 Agent 团队

我们设计了一个由 6 个专业智能体组成的团队,它们各司其职、相互协作,共同完成一篇高质量技术博客的生成。

表格

智能体角色 职责 推荐模型
选题策划 Agent 分析技术热点,生成热门选题 gpt-4o-mini
大纲设计 Agent 设计文章结构和详细大纲 gpt-5.4
内容撰写 Agent 生成文章正文内容 claude-4.6-sonnet
代码生成 Agent 编写高质量可运行代码 gpt-4o-code
审稿优化 Agent 审稿纠错,优化文章质量 claude-4.6-opus
发布准备 Agent 生成标题、标签和摘要 gpt-4o-mini

整个工作流程如下:

  1. 选题策划 Agent 生成 3 个热门技术选题
  2. 用户选择一个选题,大纲设计 Agent 生成详细大纲
  3. 内容撰写 Agent 根据大纲生成正文初稿
  4. 代码生成 Agent 为文章添加代码示例
  5. 审稿优化 Agent 对全文进行审核和优化
  6. 发布准备 Agent 生成最终的发布版本

三、代码实现:从零构建多智能体系统

下面我将一步步带你实现这个多智能体系统,所有代码都可以直接复制使用。

3.1 环境准备

bash

运行

pip install openai>=1.0.0 python-dotenv

创建.env文件:

env

4SAPI_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
4SAPI_BASE_URL=https://4sapi.com/v1

3.2 基础智能体类实现

首先,我们实现一个通用的 BaseAgent 类,所有具体的智能体都将继承自这个类:

python

运行

import os
from dotenv import load_dotenv
from openai import OpenAI
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from openai import APIError, APIConnectionError, RateLimitError

# 加载环境变量
load_dotenv()

# 初始化4sapi客户端
client = OpenAI(
    base_url=os.getenv("4SAPI_BASE_URL"),
    api_key=os.getenv("4SAPI_API_KEY")
)

class BaseAgent:
    """基础智能体类"""
    def __init__(self, name, system_prompt, model="gpt-4o-mini"):
        self.name = name
        self.system_prompt = system_prompt
        self.model = model
        self.history = []
    
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        retry=retry_if_exception_type((APIConnectionError, RateLimitError))
    )
    def run(self, input_text):
        """运行智能体"""
        messages = [
            {"role": "system", "content": self.system_prompt}
        ]
        
        # 添加历史对话
        messages.extend(self.history)
        
        # 添加当前输入
        messages.append({"role": "user", "content": input_text})
        
        # 调用4sapi
        response = client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=0.7
        )
        
        output = response.choices[0].message.content
        
        # 更新历史
        self.history.append({"role": "user", "content": input_text})
        self.history.append({"role": "assistant", "content": output})
        
        # 打印日志
        print(f"[{self.name}] 完成任务")
        print(f"输入Token: {response.usage.prompt_tokens}")
        print(f"输出Token: {response.usage.completion_tokens}\n")
        
        return output

3.3 各个专业智能体实现

接下来,我们实现各个具体的智能体角色:

python

运行

# 1. 选题策划Agent
topic_planner = BaseAgent(
    name="选题策划Agent",
    system_prompt="""你是一个专业的技术选题策划专家。请根据当前技术热点,生成3个有吸引力的技术博客选题。
要求:
1. 选题要具体、实用,能够解决开发者的实际问题
2. 每个选题用一句话描述,不超过30字
3. 按热度排序,最热门的放在最前面
4. 只返回选题列表,不要其他解释""",
    model="gpt-4o-mini"
)

# 2. 大纲设计Agent
outline_designer = BaseAgent(
    name="大纲设计Agent",
    system_prompt="""你是一个资深的技术文章大纲设计师。请根据给定的选题,设计一个详细的文章大纲。
要求:
1. 使用Markdown多级标题格式
2. 结构清晰,逻辑严谨
3. 包含前言、核心内容、代码示例、总结等部分
4. 每个部分要有简短的内容说明""",
    model="gpt-5.4"
)

# 3. 内容撰写Agent
content_writer = BaseAgent(
    name="内容撰写Agent",
    system_prompt="""你是一个专业的技术内容作家。请根据给定的大纲,生成一篇高质量的技术文章正文。
要求:
1. 语言流畅,通俗易懂
2. 技术准确,细节丰富
3. 每个部分内容充实,不少于300字
4. 代码部分用[CODE_PLACEHOLDER]标记,留待代码生成Agent填充""",
    model="claude-4.6-sonnet"
)

# 4. 代码生成Agent
code_generator = BaseAgent(
    name="代码生成Agent",
    system_prompt="""你是一个资深的Python开发工程师。请根据文章内容,为所有[CODE_PLACEHOLDER]标记生成对应的代码示例。
要求:
1. 代码要完整可运行
2. 有详细的注释
3. 遵循PEP8规范
4. 每个代码示例不超过100行""",
    model="gpt-4o-code"
)

# 5. 审稿优化Agent
editor = BaseAgent(
    name="审稿优化Agent",
    system_prompt="""你是一个专业的技术编辑。请对给定的文章进行审核和优化。
要求:
1. 修正语法错误和错别字
2. 优化语句表达,提高可读性
3. 调整文章结构,使逻辑更清晰
4. 保持原文的技术内容和风格不变""",
    model="claude-4.6-opus"
)

# 6. 发布准备Agent
publisher = BaseAgent(
    name="发布准备Agent",
    system_prompt="""你是一个专业的博客运营专家。请为给定的文章生成发布所需的内容。
要求:
1. 生成3个吸引人的标题
2. 生成5个相关的标签
3. 生成一篇100字左右的摘要
4. 按格式输出,不要其他解释""",
    model="gpt-4o-mini"
)

3.4 工作流编排

最后,我们实现一个工作流编排器,将各个智能体串联起来,实现自动化的博客生成:

python

运行

class BlogGenerationWorkflow:
    """博客生成工作流"""
    def __init__(self):
        self.agents = {
            "topic_planner": topic_planner,
            "outline_designer": outline_designer,
            "content_writer": content_writer,
            "code_generator": code_generator,
            "editor": editor,
            "publisher": publisher
        }
    
    def run(self, topic_hint=""):
        """运行完整工作流"""
        print("=== 开始生成技术博客 ===\n")
        
        # 步骤1:生成选题
        print("步骤1/6:生成选题")
        topics = self.agents["topic_planner"].run(topic_hint)
        print("生成的选题:")
        print(topics + "\n")
        
        # 这里可以让用户选择选题,为了演示我们直接选第一个
        selected_topic = topics.split("\n")[0].strip()
        print(f"选择的选题:{selected_topic}\n")
        
        # 步骤2:生成大纲
        print("步骤2/6:生成大纲")
        outline = self.agents["outline_designer"].run(selected_topic)
        print("生成的大纲:")
        print(outline + "\n")
        
        # 步骤3:生成正文
        print("步骤3/6:生成正文")
        draft = self.agents["content_writer"].run(outline)
        
        # 步骤4:生成代码
        print("步骤4/6:生成代码")
        full_content = self.agents["code_generator"].run(draft)
        
        # 步骤5:审稿优化
        print("步骤5/6:审稿优化")
        final_content = self.agents["editor"].run(full_content)
        
        # 步骤6:生成发布内容
        print("步骤6/6:生成发布内容")
        publish_info = self.agents["publisher"].run(final_content)
        
        print("=== 博客生成完成 ===\n")
        
        return {
            "topic": selected_topic,
            "outline": outline,
            "content": final_content,
            "publish_info": publish_info
        }

# 运行工作流
if __name__ == "__main__":
    workflow = BlogGenerationWorkflow()
    result = workflow.run("Python AI应用开发")
    
    # 保存结果到文件
    with open("generated_blog.md", "w", encoding="utf-8") as f:
        f.write("# " + result["topic"] + "\n\n")
        f.write(result["content"] + "\n\n")
        f.write("---\n\n")
        f.write("## 发布信息\n\n")
        f.write(result["publish_info"])
    
    print("博客已保存到 generated_blog.md 文件")

四、进阶优化:生产级多智能体系统

上面的基础版本已经可以实现自动化博客生成,但要应用到生产环境,我们还需要进行一些优化。

4.1 智能体并行执行

对于可以并行执行的任务,我们可以使用异步调用大幅提升效率:

python

运行

import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI(
    base_url=os.getenv("4SAPI_BASE_URL"),
    api_key=os.getenv("4SAPI_API_KEY")
)

class AsyncBaseAgent:
    """异步基础智能体类"""
    def __init__(self, name, system_prompt, model="gpt-4o-mini"):
        self.name = name
        self.system_prompt = system_prompt
        self.model = model
    
    async def run(self, input_text):
        """异步运行智能体"""
        messages = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": input_text}
        ]
        
        response = await async_client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=0.7
        )
        
        return response.choices[0].message.content

# 并行生成多个代码示例
async def generate_multiple_codes(sections):
    tasks = [code_generator.run(section) for section in sections]
    results = await asyncio.gather(*tasks)
    return results

4.2 成本统计与控制

4sapi 提供了详细的使用量统计,我们可以在工作流中添加成本统计功能:

python

运行

class CostTracker:
    """成本追踪器"""
    def __init__(self):
        self.total_tokens = 0
        self.total_cost = 0.0
        
        # 模型价格(美元/1000Token)
        self.model_prices = {
            "gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
            "gpt-5.4": {"input": 0.01, "output": 0.03},
            "claude-4.6-sonnet": {"input": 0.003, "output": 0.015},
            "claude-4.6-opus": {"input": 0.015, "output": 0.075},
            "gpt-4o-code": {"input": 0.01, "output": 0.03}
        }
    
    def add_call(self, model, prompt_tokens, completion_tokens):
        """添加一次调用记录"""
        self.total_tokens += prompt_tokens + completion_tokens
        
        prices = self.model_prices.get(model, {"input": 0, "output": 0})
        cost = (prompt_tokens * prices["input"] + completion_tokens * prices["output"]) / 1000
        self.total_cost += cost
    
    def get_summary(self):
        """获取成本统计摘要"""
        return f"总Token消耗:{self.total_tokens},总成本:${self.total_cost:.4f}"

4.3 智能体状态管理

对于复杂的多智能体系统,我们需要实现状态管理,支持任务中断和恢复:

python

运行

import json

class WorkflowState:
    """工作流状态管理"""
    def __init__(self, workflow_id):
        self.workflow_id = workflow_id
        self.current_step = 0
        self.results = {}
    
    def save(self, file_path=None):
        """保存状态到文件"""
        if file_path is None:
            file_path = f"workflow_{self.workflow_id}.json"
        
        with open(file_path, "w", encoding="utf-8") as f:
            json.dump({
                "workflow_id": self.workflow_id,
                "current_step": self.current_step,
                "results": self.results
            }, f, ensure_ascii=False, indent=2)
    
    @classmethod
    def load(cls, file_path):
        """从文件加载状态"""
        with open(file_path, "r", encoding="utf-8") as f:
            data = json.load(f)
        
        state = cls(data["workflow_id"])
        state.current_step = data["current_step"]
        state.results = data["results"]
        return state

五、实际效果与收益

我们已经将这个多智能体系统应用到了团队的技术博客创作中,取得了非常显著的效果:

  • 效率提升 90%:原来需要 3 天完成的博客,现在只需要 2 小时
  • 成本降低 45%:通过 4sapi 的智能路由功能,大幅降低了 API 调用成本
  • 质量稳定:生成的文章质量稳定,只需要少量人工修改即可发布
  • 可扩展性强:添加新的智能体角色或切换模型只需要几行代码

六、总结与展望

基于 4sapi 构建多智能体系统,让我们真正体会到了 "专注于业务逻辑,而不是底层基础设施" 的好处。4sapi 的统一接口设计、丰富的模型支持和企业级稳定性,为 AI Agent 应用提供了完美的底座。

未来,我们计划进一步扩展这个系统:

  • 添加更多的智能体角色,如 SEO 优化 Agent、翻译 Agent 等
  • 实现智能体之间的自主通信和任务分配
  • 集成更多的工具,如搜索引擎、代码执行器等
  • 支持自定义工作流,让用户可以通过拖拽的方式设计自己的智能体团队

如果你也正在探索 AI Agent 技术,或者想要构建自己的多智能体系统,我强烈推荐你尝试一下 4sapi。它简单易用、功能强大,而且成本低廉,绝对会让你的开发效率提升一个数量级。

Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐