1. 框架简介

Swarms 是一个面向生产环境的企业级多智能体编排框架，提供完整的智能体基础设施，支持大规模部署和与现有系统的集成。

技术亮点

• 三层解耦架构：流程编排、智能体执行、通信存储分离
• 共享对话历史：智能体通过 Conversation 对象自动传递上下文
• Flow DSL：字符串定义复杂工作流（"agent1 -> agent2, agent3"）
• 多模式融合：ReAct + 规划 + 反思 + 工具执行
• 分布式支持：Redis/Pulsar 实现多节点会话共享

2. 快速开始

from swarms import Agent, SequentialWorkflow

# 创建智能体
researcher = Agent(
    agent_name="Researcher",
    system_prompt="You are a research specialist.",
    model_name="gpt-4o-mini",
)

writer = Agent(
    agent_name="Writer",
    system_prompt="You are a writer.",
    model_name="gpt-4o-mini",
)

# 创建工作流
workflow = SequentialWorkflow(agents=[researcher, writer])

# 执行任务
result = workflow.run("研究并撰写关于AI的报告")

3. 框架的规划执行反思如何实现？

这个框架使用了多种模式，其实不同模式就是对应不同的Prompt模板，由Prompt引导，

ReAct模式

比如以实现ReAct模式为例：swarms/prompts/react_base_prompt.py

REACT_SYS_PROMPT = """You are a thoughtful and methodical AI agent. 
You solve problems through careful reasoning and by using external tools when needed. 
You use a "Thought → Action → Observation" loop...

Follow this structure:
Question: [The user's input]
Thought 1: Understand the question...
Action 1: [Use a tool]
Observation 1: [The result from the tool]
Thought 2: Reflect on the observation...
Action 2: [Next tool]
...
"""

使用方式：

agent = Agent(
    system_prompt=REACT_SYS_PROMPT,  # 使用ReAct提示词
    tools=[search_tool, calculator_tool],
    max_loops=5,  # 允许多次循环
)

自主循环模式

核心执行机制：

def _run(self, task, img=None, streaming_callback=None, *args, **kwargs):
    # 1. 初始化任务
    self.short_memory.add(role="User", content=task)
  
    # 2. 规划阶段（可选）
    if self.planning:
        self.create_plan(task)  # 生成计划
  
    # 3. 主循环：规划 → 执行 → 反思
    while loop_count < self.max_loops:
        loop_count += 1
  
        # 3.1 推理提示（反思机制）
        if self.reasoning_prompt_on and self.max_loops >= 2:
            self.short_memory.add(
                role=self.agent_name,
                content=f"Current Internal Reasoning Loop: {loop_count}/{self.max_loops}"
            )
  
        # 3.2 调用LLM生成响应（执行）
        response = self.call_llm(
            task=self.short_memory.return_history_as_string(),
            current_loop=loop_count,
        )
  
        # 3.3 将响应存入内存
        self.short_memory.add(role=self.agent_name, content=response)
  
        # 3.4 工具执行（Action）
        if self.tools:
            self.tool_execution_retry(response, loop_count)
  
        # 3.5 检查停止条件
        if self.stopping_condition(response):
            break
  
    return history_output_formatter(self.short_memory, type=self.output_type)

规划功能实现：

def create_plan(self, task, *args, **kwargs):
    """创建执行计划"""
    # 获取对话历史
    history = self.short_memory.get_str()
  
    # 构建规划提示词
    if self.planning_prompt:
        planning_prompt = f"{history}\n\n{self.planning_prompt}\n\nTask: {task}"
    else:
        planning_prompt = f"{history}\n\nCreate a comprehensive step-by-step plan to complete: {task}"
  
    # 使用LLM生成计划
    plan = self.llm.run(task=planning_prompt, *args, **kwargs)
  
    # 将计划存入内存
    self.short_memory.add(role=self.agent_name, content=plan)
  
    return None

使用方式：

agent = Agent(
    planning=True,  # 启用规划
    planning_prompt="Break down the task into steps...",
    max_loops=5,
)

反思机制实现

反思通过多轮循环和推理提示词实现：

# 在循环中添加推理提示
if self.reasoning_prompt_on and self.max_loops >= 2:
    # 添加当前循环信息（引导反思）
    self.short_memory.add(
        role=self.agent_name,
        content=f"Current Internal Reasoning Loop: {loop_count}/{self.max_loops}"
    )

# 最后一轮添加最终反思提示
if loop_count == self.max_loops:
    self.short_memory.add(
        role=self.agent_name,
        content=f"🎉 Final Internal Reasoning Loop: {loop_count}/{self.max_loops} Prepare your comprehensive response."
    )

推理提示词生成：

def generate_reasoning_prompt(max_loops: int) -> str:
    return f"""
    Your task is to perform **exactly one loop per generation**, 
    until either the problem is solved or you have completed {max_loops} loops.

    ## Loop Structure (per generation):
    1. **Summarize the Current State** - Recap known information
    2. **Generate Hypotheses** - Explore possible next steps
    3. **Evaluate and Choose** - Narrow down based on logic
    4. **Act and Update Memory** - Take the chosen step
    5. **Reflect** - Consider whether this step brings you closer to solving
    """

Reflexion模式（专门的反思Agent）

框架提供了专门的ReflexionAgent类：flexion_agent.py

class ReflexionAgent:
    """
    实现Reflexion框架的Agent：
    1. Acting on tasks (执行)
    2. Evaluating its performance (评估)
    3. Generating self-reflections (反思)
    4. Using these reflections to improve future responses (改进)
    """
  
    def step(self, task, iteration=0, previous_response=None):
        # 1. 执行任务
        response = self.act(task, relevant_memories)
  
        # 2. 评估响应
        evaluation, score = self.evaluate(task, response)
  
        # 3. 生成反思
        reflection = self.reflect(task, response, evaluation)
  
        # 4. 存储到内存
        self.memory.add_short_term_memory({
            "task": task,
            "response": response,
            "evaluation": evaluation,
            "reflection": reflection,
            "score": score,
        })
  
        return {
            "response": response,
            "evaluation": evaluation,
            "reflection": reflection,
        }

自主Agent模式（Plan-Think-Act-Reflect）

框架还提供了完整的自主Agent提示词：

AUTONOMOUS_AGENT_SYSTEM_PROMPT = """
## AUTONOMOUS LOOP WORKFLOW

### PHASE 1: PLANNING
1. Analyze the main task thoroughly
2. Break it down into smaller, manageable subtasks
3. Assign appropriate priorities
4. Use the `create_plan` tool to formalize your plan

### PHASE 2: EXECUTION
1. **Brief Analysis**: Use the `think` tool ONCE
2. **Take Action**: Use available tools to complete the work
3. **Complete Subtask**: Call `subtask_done` when finished

### PHASE 3: THINKING (Between Tasks)
1. Assess current state
2. Determine next action
3. Keep it brief
"""

完整执行流程

from swarms import Agent
from swarms.prompts.react_base_prompt import REACT_SYS_PROMPT

# 创建支持ReAct + 规划 + 反思的Agent
agent = Agent(
    agent_name="SmartAgent",
    system_prompt=REACT_SYS_PROMPT,  # ReAct模式
    planning=True,  # 启用规划
    planning_prompt="Break down the task into clear steps...",
    reasoning_prompt_on=True,  # 启用推理/反思
    max_loops=5,  # 允许5轮循环
    tools=[search_tool, calculator_tool],  # 工具支持
)

# 执行任务
result = agent.run("研究并分析AI发展趋势")

# 内部执行流程：
# Loop 1: 规划阶段
#   - create_plan() → 生成计划
#   - LLM生成响应 → 存入内存
#   - 工具执行（如果需要）
#
# Loop 2-4: 执行+反思阶段
#   - 添加推理提示："Current Internal Reasoning Loop: 2/5"
#   - LLM基于历史生成响应（包含反思）
#   - 工具执行
#   - 存入内存
#
# Loop 5: 最终反思
#   - 添加最终提示："Final Internal Reasoning Loop: 5/5"
#   - LLM生成综合响应
#   - 返回结果