Agent 学习笔记：框架 ｜ openAI Swarm

# 导入必要的库
from openai import OpenAI
from pydantic import BaseModel
from typing import Optional
import json

# 创建OpenAI客户端
client = OpenAI()

# 客服 Routine 示例
system_message = (
    "你是ACME Inc的客服代理。"
    "回答简短。"
    "按照以下步骤处理："
    "1. 询问客户问题，了解他们遇到的具体问题。"
    "2. 提出修复建议。"
    "3. 如果客户不满意，提供退款。"
    "4. 如果客户接受退款，请查找商品ID并执行退款。"
)

# 示例函数：查找商品
def look_up_item(search_query):
    """用于查找商品ID，可以通过描述或关键词进行搜索"""
    # 返回硬编码的商品ID，实际场景中会调用数据库或API查找
    return "item_132612938"

# 示例函数：执行退款
def execute_refund(item_id, reason="未提供"):
    """执行退款操作"""
    print("退款摘要:", item_id, reason)
    return "退款成功"

# Routine 运行示例
def run_full_turn(system_message, messages):
    # 使用OpenAI模型完成聊天对话
    response = client.chat.completions.create(
        model="gpt-4o-mini",  # 使用的模型
        messages=[{"role": "system", "content": system_message}] + messages,
    )
    message = response.choices[0].message
    messages.append(message)

    # 输出客服代理的回复
    if message.content:
        print("代理:", message.content)

    return message

# 消息列表
messages = []

# 运行示例（此时代理根据系统消息进行对话）
while True:
    user_input = input("用户: ")
    messages.append({"role": "user", "content": user_input})
    run_full_turn(system_message, messages)

# 导入必要的库
from openai import OpenAI
from pydantic import BaseModel
import json

# 创建OpenAI客户端
client = OpenAI()

# 客服 Routine 示例
system_message = (
    "你是ACME Inc的客服代理。"
    "回答简短。"
    "按照以下步骤处理："
    "1. 询问客户问题，了解他们遇到的具体问题。"
    "2. 提出修复建议。"
    "3. 如果客户不满意，提供退款。"
    "4. 如果客户接受退款，请查找商品ID并执行退款。"
)

# 查找商品的工具函数
def look_up_item(search_query):
    """用于查找商品ID，可以通过描述或关键词进行搜索"""
    return "item_132612938"  # 模拟返回商品ID

# 执行退款的工具函数
def execute_refund(item_id, reason="未提供"):
    """执行退款操作"""
    print(f"执行退款操作，商品ID: {item_id}, 原因: {reason}")
    return "退款成功"

# 将工具函数转换为模型可以调用的格式
def function_to_schema(func):
    """将Python函数转换为函数调用模式，供模型调用"""
    return {
        "name": func.__name__,
        "description": func.__doc__,
        "parameters": {
            "type": "object",
            "properties": {
                "search_query": {"type": "string"},
                "item_id": {"type": "string"},
                "reason": {"type": "string"}
            }
        }
    }

# Routine 执行示例，支持工具调用
def run_full_turn(system_message, messages, tools):
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "system", "content": system_message}] + messages,
        tools=tools
    )
    message = response.choices[0].message
    messages.append(message)

    if message.content:
        print("代理:", message.content)

    # 检查是否有工具调用
    if "tool_call" in message:
        tool_call = message["tool_call"]
        tool_name = tool_call["name"]
        tool_args = json.loads(tool_call["arguments"])
        
        # 根据工具调用名称，调用相应的工具函数
        if tool_name == "look_up_item":
            result = look_up_item(**tool_args)
        elif tool_name == "execute_refund":
            result = execute_refund(**tool_args)
        
        print(f"工具调用结果: {result}")
        messages.append({"role": "tool", "content": result})

    return message

# 准备工具和工具模式
tools = [look_up_item, execute_refund]
tool_schemas = [function_to_schema(tool) for tool in tools]

# 消息列表
messages = []

# 运行示例，使用工具调用
while True:
    user_input = input("用户: ")
    messages.append({"role": "user", "content": user_input})
    run_full_turn(system_message, messages, tool_schemas)

# 导入必要的库
from pydantic import BaseModel
import json

# 定义代理类
class Agent(BaseModel):
    name: str
    instructions: str
    tools: list

# 工具函数
def look_up_item(search_query):
    return "item_132612938"

def execute_refund(item_id, reason="未提供"):
    return "退款成功"

def place_order(item_name):
    return "订单已下"

# 定义代理
refund_agent = Agent(
    name="退款代理",
    instructions="你是退款代理，帮助用户进行退款。",
    tools=[execute_refund]
)

sales_agent = Agent(
    name="销售代理",
    instructions="你是销售代理，帮助用户下订单。",
    tools=[place_order]
)

# Handoff 转接函数
def transfer_to_refund():
    return refund_agent

# 消息处理函数
def run_full_turn(agent, messages):
    print(f"{agent.name}:", agent.instructions)
    
    # 交互消息
    while True:
        user_input = input("用户: ")
        messages.append({"role": "user", "content": user_input})
        
        # 根据用户输入，代理决定是否进行转接
        if "退款" in user_input:
            agent = transfer_to_refund()
            print(f"转接到 {agent.name}")
            break
        else:
            result = place_order(user_input)  # 默认下单工具
            print(f"代理调用结果: {result}")
    
    return agent

# 初始化销售代理处理
messages = []
agent = sales_agent

# 模拟对话，代理处理转接
agent = run_full_turn(agent, messages)

class Agent(BaseModel):
    name: str = "Agent"                    # 代理名称
    model: str = "gpt-4o"                  # 使用的模型
    instructions: Union[str, Callable[[], str]] = "You are a helpful agent."  # 代理指令
    functions: List[AgentFunction] = []     # 可用函数列表
    tool_choice: str = None                # 工具选择
    parallel_tool_calls: bool = True       # 是否允许并行调用工具

class Response(BaseModel):
    messages: List = []           # 对话消息列表
    agent: Optional[Agent] = None # 当前活跃代理
    context_variables: dict = {}  # 上下文变量

class Result(BaseModel):
    """
    Encapsulates the possible return values for an agent function.

    Attributes:
        value (str): The result value as a string.
        agent (Agent): The agent instance, if applicable.
        context_variables (dict): A dictionary of context variables.
    """

    value: str = ""              # 函数返回值
    agent: Optional[Agent] = None # 可能切换到的新代理
    context_variables: dict = {}  # 函数执行后的上下文变量

def run(
    self,
    agent: Agent,          # 初始代理
    messages: List,        # 初始消息列表
    context_variables: dict = {},  # 上下文变量
    model_override: str = None,    # 模型覆盖
    stream: bool = False,          # 是否使用流式输出
    debug: bool = False,           # 是否开启调试
    max_turns: int = float("inf"), # 最大对话轮次
    execute_tools: bool = True,    # 是否执行工具调用
) -> Response:

# 如果是流式模式，转向run_and_stream方法
if stream:
    return self.run_and_stream(...)

# 复制初始状态，避免修改原始数据
active_agent = agent
context_variables = copy.deepcopy(context_variables)
history = copy.deepcopy(messages)
init_len = len(messages)  # 记录初始消息长度

while len(history) - init_len < max_turns and active_agent:
    # 获取AI回复
    completion = self.get_chat_completion(
        agent=active_agent,
        history=history,
        context_variables=context_variables,
        model_override=model_override,
        stream=stream,
        debug=debug,
    )
    
    # 处理回复消息
    message = completion.choices[0].message
    message.sender = active_agent.name
    history.append(json.loads(message.model_dump_json()))

    # 如果没有工具调用或不执行工具，结束对话
    if not message.tool_calls or not execute_tools:
        break

    # 处理工具调用
    partial_response = self.handle_tool_calls(
        message.tool_calls, 
        active_agent.functions, 
        context_variables, 
        debug
    )
    
    # 更新状态
    history.extend(partial_response.messages)  # 添加工具调用结果到历史
    context_variables.update(partial_response.context_variables)  # 更新上下文
    if partial_response.agent:
        active_agent = partial_response.agent  # 可能切换到新代理

return Response(
    messages=history[init_len:],  # 只返回新增的消息
    agent=active_agent,           # 当前活跃代理
    context_variables=context_variables  # 最终的上下文变量
)

from swarm import Swarm, Agent

client = Swarm()

english_agent = Agent(
    name="English Agent",
    instructions="You only speak English.",
)

spanish_agent = Agent(
    name="Spanish Agent",
    instructions="You only speak Spanish.",
)


def transfer_to_spanish_agent():
    """Transfer spanish speaking users immediately."""
    return spanish_agent


english_agent.functions.append(transfer_to_spanish_agent)

messages = [{"role": "user", "content": "Hola. ¿Como estás?"}]
response = client.run(agent=english_agent, messages=messages)

print(response.messages[-1]["content"])

def handoff_to_architect():
    return architect

def handoff_to_developer():
    return developer

def handoff_to_tester():
    return tester
def handoff_to_product_manager():
    return product_manager

from swarm import Swarm, Agent

client = Swarm(client=oai_client)


# 定义产品经理智能体
product_manager = Agent(
    name="product_manager",
    instructions="You are the Product Manager, responsible for creating a comprehensive Product Requirements Document (PRD) for a simple calculator application. Your goal is to ensure that the requirements are clear and complete to facilitate subsequent design and development. Once the requirements are finalized, you will hand off the task to the Architect by using tool: `handoff_to_architect()`. YOU CAN't TLAK TO HUMAN",
    functions=[handoff_to_architect],  # 定义产品经理的功能
    model=model_name,
)

# 定义架构师智能体
architect = Agent(
    name="architect",
    instructions="You are the Architect, responsible for designing the system architecture for a simple calculator application. Your goal is to provide a clear and robust architecture that meets the product requirements outlined in the PRD. Your design should be modular, scalable, and suitable for both desktop and mobile platforms. Once your design is complete, you will hand off the task to the Developer by using tool: `handoff_to_developer()`. YOU CAN't TLAK TO HUMAN",
    functions=[handoff_to_developer],  # 定义架构师的功能
    model=model_name,
)

# 定义开发工程师智能体
developer = Agent(
    name="developer",
    instructions="You are the Developer, responsible for implementing code based on the system architecture designed for a simple calculator application. Your goal is to follow the architectural specifications closely to ensure each module functions as intended. Your implementation should be efficient, maintainable, and well-documented. Once the code is complete, you will hand off the task to the tester by using tool: `handoff_to_tester()`. YOU CAN't TLAK TO HUMAN ",
    functions=[handoff_to_tester,handoff_to_architect,handoff_to_product_manager],  # 定义开发工程师的功能
    model=model_name,
)

# 定义测试工程师智能体
tester = Agent(
    name="tester",
    instructions="You are the Testing Engineer, responsible for creating and executing test cases to ensure that the calculator application functions as specified in the PRD and meets the standards of the architecture and code implementation. Your goal is to validate that each feature works as intended, handling both expected and edge cases. Once testing is complete, you will document the results and provide feedback to the Product Manager or Developer for any required fixes. ",
    functions=[handoff_to_developer],  # 定义测试工程师的功能
    model=model_name,
)

from swarm.repl import run_demo_loop

run_demo_loop(product_manager)