原生调用OpenAI API实现简单的"反射"智能体

核心知识点讲解

什么是"反射"智能体？

"反射"智能体（Reflex Agent）是一种基于规则和当前感知的智能体，它根据当前的输入和预设的规则直接生成响应，而不考虑历史状态或未来结果。这种智能体的行为就像"反射"一样，对输入做出即时反应。

"反射"智能体的核心特点：

• 反应式：直接对输入做出反应，无需复杂的推理过程
• 无状态：不维护或使用历史状态信息
• 简单规则：基于简单的规则或模式匹配生成响应
• 实时性：响应速度快，适合实时交互场景
• 可预测性：行为相对可预测，易于理解和调试

为什么从"反射"智能体开始？

"反射"智能体是构建更复杂智能体的基础，从它开始学习有以下优势：

1. 简单易懂：结构简单，易于理解和实现
2. 快速上手：可以快速搭建原型，验证基本概念
3. 基础技能：学习如何调用OpenAI API，处理消息等基础技能
4. 逐步进阶：为后续学习更复杂的智能体（如基于记忆的智能体）打下基础
5. 实用价值：即使是简单的反射智能体，也可以满足许多实际应用场景

OpenAI API简介

OpenAI API是一组RESTful API，允许开发者与OpenAI的大语言模型（如GPT-3.5、GPT-4等）进行交互。通过调用这些API，我们可以构建各种基于大语言模型的应用，包括聊天机器人、内容生成工具、智能助手等。

OpenAI API的核心功能：

• 文本生成：生成与输入相关的文本内容
• 聊天补全：进行多轮对话，保持上下文连贯性
• 嵌入生成：将文本转换为高维向量表示
• 微调：使用自定义数据微调模型

消息格式与处理

在使用OpenAI API进行聊天补全时，消息是通过一个数组传递的，每个消息包含以下字段：

• role：消息的角色，可以是"system"（系统）、"user"（用户）或"assistant"（助手）
• content：消息的内容

消息处理的基本原则：

1. 系统消息：设置助手的行为和个性，通常在对话开始时发送一次
2. 用户消息：用户的输入，触发助手的响应
3. 助手消息：助手的回复，作为对话历史的一部分
4. 消息顺序：消息数组的顺序很重要，它决定了对话的上下文

实现"反射"智能体的基本步骤

实现一个简单的"反射"智能体的基本步骤包括：

1. 设置API密钥：获取并配置OpenAI API密钥
2. 构建消息：根据用户输入构建消息数组
3. 调用API：使用OpenAI API生成响应
4. 处理响应：解析API返回的响应，提取助手的回复
5. 返回结果：将助手的回复返回给用户

实战案例：构建一个简单的"反射"智能体

步骤1：设置环境和安装依赖

首先，我们需要设置开发环境并安装必要的依赖：

# 创建并激活虚拟环境
python -m venv venv
source venv/bin/activate  # 在Windows上使用 venv\Scripts\activate

# 安装OpenAI Python库
pip install openai

步骤2：配置API密钥

在使用OpenAI API之前，我们需要获取并配置API密钥：

1. 获取API密钥：访问OpenAI官网（https://platform.openai.com/），登录账号，在"API Keys"页面创建新的API密钥
2. 配置API密钥：将API密钥存储在环境变量或配置文件中

步骤3：实现基本的"反射"智能体

现在，我们来实现一个基本的"反射"智能体：

import openai
import os

# 从环境变量获取API密钥
openai.api_key = os.getenv("OPENAI_API_KEY")

class ReflexAgent:
    """简单的反射智能体"""
    
    def __init__(self, system_prompt="你是一个友好的助手，能够回答用户的问题。"):
        """初始化智能体
        
        Args:
            system_prompt: 系统提示词，定义智能体的行为和个性
        """
        self.system_prompt = system_prompt
    
    def generate_response(self, user_input):
        """根据用户输入生成响应
        
        Args:
            user_input: 用户的输入文本
            
        Returns:
            str: 智能体的响应文本
        """
        # 构建消息数组
        messages = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": user_input}
        ]
        
        # 调用OpenAI API
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",  # 使用的模型
            messages=messages,      # 消息数组
            temperature=0.7,        # 温度参数，控制输出的随机性
            max_tokens=1000,        # 最大 tokens 数
            n=1,                    # 生成的响应数
            stop=None               # 停止词
        )
        
        # 提取助手的回复
        assistant_response = response.choices[0].message.content
        
        return assistant_response

# 测试智能体
if __name__ == "__main__":
    # 创建智能体实例
    agent = ReflexAgent()
    
    # 测试对话
    while True:
        user_input = input("用户: ")
        if user_input.lower() in ["退出", "quit", "exit"]:
            print("助手: 再见！")
            break
        
        # 生成响应
        response = agent.generate_response(user_input)
        print(f"助手: {response}")

步骤4：改进"反射"智能体

我们可以对基本的"反射"智能体进行一些改进，使其更加实用：

1. 添加错误处理：处理API调用可能出现的错误
2. 支持不同的模型：允许用户选择不同的模型
3. 可配置参数：允许配置温度、最大tokens等参数
4. 简单的规则匹配：添加一些简单的规则，使智能体能够对特定输入做出特定反应

import openai
import os
import time

class ImprovedReflexAgent:
    """改进的反射智能体"""
    
    def __init__(self, 
                 system_prompt="你是一个友好的助手，能够回答用户的问题。",
                 model="gpt-3.5-turbo",
                 temperature=0.7,
                 max_tokens=1000):
        """初始化智能体
        
        Args:
            system_prompt: 系统提示词，定义智能体的行为和个性
            model: 使用的模型
            temperature: 温度参数，控制输出的随机性
            max_tokens: 最大 tokens 数
        """
        self.system_prompt = system_prompt
        self.model = model
        self.temperature = temperature
        self.max_tokens = max_tokens
        self.api_key = os.getenv("OPENAI_API_KEY")
        if not self.api_key:
            raise ValueError("请设置OPENAI_API_KEY环境变量")
        openai.api_key = self.api_key
    
    def generate_response(self, user_input):
        """根据用户输入生成响应
        
        Args:
            user_input: 用户的输入文本
            
        Returns:
            str: 智能体的响应文本
        """
        # 简单的规则匹配
        if "你好" in user_input or "您好" in user_input:
            return "你好！我是你的智能助手，有什么可以帮助你的吗？"
        
        if "再见" in user_input or "拜拜" in user_input:
            return "再见！祝你有愉快的一天！"
        
        if "你是谁" in user_input or "你是什么" in user_input:
            return "我是一个基于OpenAI API构建的智能助手，能够回答你的问题，提供信息和帮助。"
        
        # 构建消息数组
        messages = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": user_input}
        ]
        
        try:
            # 调用OpenAI API
            response = openai.ChatCompletion.create(
                model=self.model,
                messages=messages,
                temperature=self.temperature,
                max_tokens=self.max_tokens,
                n=1,
                stop=None
            )
            
            # 提取助手的回复
            assistant_response = response.choices[0].message.content
            
            return assistant_response
            
        except openai.error.RateLimitError:
            return "抱歉，API调用频率过高，请稍后再试。"
        except openai.error.APIError as e:
            return f"API错误: {str(e)}"
        except Exception as e:
            return f"发生错误: {str(e)}"

# 测试智能体
if __name__ == "__main__":
    # 创建智能体实例
    agent = ImprovedReflexAgent(
        system_prompt="你是一个专业的技术助手，擅长回答关于编程、AI和技术相关的问题。请以清晰、准确的方式回答用户的问题。",
        model="gpt-3.5-turbo",
        temperature=0.6,
        max_tokens=1000
    )
    
    print("助手: 你好！我是你的技术助手，有什么可以帮助你的吗？")
    
    # 测试对话
    while True:
        user_input = input("用户: ")
        if user_input.lower() in ["退出", "quit", "exit"]:
            print("助手: 再见！")
            break
        
        # 生成响应
        response = agent.generate_response(user_input)
        print(f"助手: {response}")

代码示例

示例1：基本的"反射"智能体

import openai
import os

# 从环境变量获取API密钥
openai.api_key = os.getenv("OPENAI_API_KEY")

class ReflexAgent:
    """简单的反射智能体"""
    
    def __init__(self, system_prompt="你是一个友好的助手，能够回答用户的问题。"):
        """初始化智能体
        
        Args:
            system_prompt: 系统提示词，定义智能体的行为和个性
        """
        self.system_prompt = system_prompt
    
    def generate_response(self, user_input):
        """根据用户输入生成响应
        
        Args:
            user_input: 用户的输入文本
            
        Returns:
            str: 智能体的响应文本
        """
        # 构建消息数组
        messages = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": user_input}
        ]
        
        # 调用OpenAI API
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",  # 使用的模型
            messages=messages,      # 消息数组
            temperature=0.7,        # 温度参数，控制输出的随机性
            max_tokens=1000,        # 最大 tokens 数
            n=1,                    # 生成的响应数
            stop=None               # 停止词
        )
        
        # 提取助手的回复
        assistant_response = response.choices[0].message.content
        
        return assistant_response

# 测试智能体
if __name__ == "__main__":
    # 创建智能体实例
    agent = ReflexAgent()
    
    # 测试对话
    while True:
        user_input = input("用户: ")
        if user_input.lower() in ["退出", "quit", "exit"]:
            print("助手: 再见！")
            break
        
        # 生成响应
        response = agent.generate_response(user_input)
        print(f"助手: {response}")

示例2：改进的"反射"智能体

import openai
import os
import time

class ImprovedReflexAgent:
    """改进的反射智能体"""
    
    def __init__(self, 
                 system_prompt="你是一个友好的助手，能够回答用户的问题。",
                 model="gpt-3.5-turbo",
                 temperature=0.7,
                 max_tokens=1000):
        """初始化智能体
        
        Args:
            system_prompt: 系统提示词，定义智能体的行为和个性
            model: 使用的模型
            temperature: 温度参数，控制输出的随机性
            max_tokens: 最大 tokens 数
        """
        self.system_prompt = system_prompt
        self.model = model
        self.temperature = temperature
        self.max_tokens = max_tokens
        self.api_key = os.getenv("OPENAI_API_KEY")
        if not self.api_key:
            raise ValueError("请设置OPENAI_API_KEY环境变量")
        openai.api_key = self.api_key
    
    def generate_response(self, user_input):
        """根据用户输入生成响应
        
        Args:
            user_input: 用户的输入文本
            
        Returns:
            str: 智能体的响应文本
        """
        # 简单的规则匹配
        if "你好" in user_input or "您好" in user_input:
            return "你好！我是你的智能助手，有什么可以帮助你的吗？"
        
        if "再见" in user_input or "拜拜" in user_input:
            return "再见！祝你有愉快的一天！"
        
        if "你是谁" in user_input or "你是什么" in user_input:
            return "我是一个基于OpenAI API构建的智能助手，能够回答你的问题，提供信息和帮助。"
        
        # 构建消息数组
        messages = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": user_input}
        ]
        
        try:
            # 调用OpenAI API
            response = openai.ChatCompletion.create(
                model=self.model,
                messages=messages,
                temperature=self.temperature,
                max_tokens=self.max_tokens,
                n=1,
                stop=None
            )
            
            # 提取助手的回复
            assistant_response = response.choices[0].message.content
            
            return assistant_response
            
        except openai.error.RateLimitError:
            return "抱歉，API调用频率过高，请稍后再试。"
        except openai.error.APIError as e:
            return f"API错误: {str(e)}"
        except Exception as e:
            return f"发生错误: {str(e)}"

# 测试智能体
if __name__ == "__main__":
    # 创建智能体实例
    agent = ImprovedReflexAgent(
        system_prompt="你是一个专业的技术助手，擅长回答关于编程、AI和技术相关的问题。请以清晰、准确的方式回答用户的问题。",
        model="gpt-3.5-turbo",
        temperature=0.6,
        max_tokens=1000
    )
    
    print("助手: 你好！我是你的技术助手，有什么可以帮助你的吗？")
    
    # 测试对话
    while True:
        user_input = input("用户: ")
        if user_input.lower() in ["退出", "quit", "exit"]:
            print("助手: 再见！")
            break
        
        # 生成响应
        response = agent.generate_response(user_input)
        print(f"助手: {response}")

示例3：领域特定的"反射"智能体

import openai
import os

class DomainSpecificReflexAgent:
    """领域特定的反射智能体"""
    
    def __init__(self, domain):
        """初始化智能体
        
        Args:
            domain: 智能体的专业领域
        """
        self.domain = domain
        self.system_prompt = self._get_domain_prompt(domain)
        self.api_key = os.getenv("OPENAI_API_KEY")
        if not self.api_key:
            raise ValueError("请设置OPENAI_API_KEY环境变量")
        openai.api_key = self.api_key
    
    def _get_domain_prompt(self, domain):
        """根据领域获取系统提示词
        
        Args:
            domain: 智能体的专业领域
            
        Returns:
            str: 系统提示词
        """
        prompts = {
            "编程": "你是一位专业的编程专家，擅长多种编程语言和技术栈。请以清晰、准确的方式回答用户的编程问题，提供详细的代码示例和解释。",
            "数学": "你是一位专业的数学家，擅长解决各种数学问题。请以清晰、准确的方式回答用户的数学问题，提供详细的解题步骤和解释。",
            "医疗": "你是一位专业的医疗顾问，能够回答用户的健康问题。请注意，你的回答仅供参考，不能替代专业医生的诊断和治疗建议。",
            "法律": "你是一位专业的法律顾问，能够回答用户的法律问题。请注意，你的回答仅供参考，不能替代专业律师的咨询和建议。",
            "金融": "你是一位专业的金融顾问，擅长投资、理财和财务规划。请以清晰、准确的方式回答用户的金融问题，提供专业的建议和分析。",
        }
        
        return prompts.get(domain, "你是一个友好的助手，能够回答用户的问题。")
    
    def generate_response(self, user_input):
        """根据用户输入生成响应
        
        Args:
            user_input: 用户的输入文本
            
        Returns:
            str: 智能体的响应文本
        """
        # 构建消息数组
        messages = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": user_input}
        ]
        
        try:
            # 调用OpenAI API
            response = openai.ChatCompletion.create(
                model="gpt-3.5-turbo",
                messages=messages,
                temperature=0.7,
                max_tokens=1000,
                n=1,
                stop=None
            )
            
            # 提取助手的回复
            assistant_response = response.choices[0].message.content
            
            return assistant_response
            
        except Exception as e:
            return f"发生错误: {str(e)}"

# 测试智能体
if __name__ == "__main__":
    # 选择领域
    domains = ["编程", "数学", "医疗", "法律", "金融"]
    print("请选择智能体的专业领域:")
    for i, domain in enumerate(domains, 1):
        print(f"{i}. {domain}")
    
    choice = int(input("请输入数字选择领域: "))
    if 1 <= choice <= len(domains):
        selected_domain = domains[choice-1]
        print(f"你选择了: {selected_domain}")
        
        # 创建智能体实例
        agent = DomainSpecificReflexAgent(selected_domain)
        
        print(f"助手 ({selected_domain}): 你好！我是你的{selected_domain}专业助手，有什么可以帮助你的吗？")
        
        # 测试对话
        while True:
            user_input = input("用户: ")
            if user_input.lower() in ["退出", "quit", "exit"]:
                print(f"助手 ({selected_domain}): 再见！")
                break
            
            # 生成响应
            response = agent.generate_response(user_input)
            print(f"助手 ({selected_domain}): {response}")
    else:
        print("无效的选择")

示例4：使用函数调用的"反射"智能体

import openai
import os
import json

class FunctionCallingReflexAgent:
    """使用函数调用的反射智能体"""
    
    def __init__(self):
        """初始化智能体"""
        self.api_key = os.getenv("OPENAI_API_KEY")
        if not self.api_key:
            raise ValueError("请设置OPENAI_API_KEY环境变量")
        openai.api_key = self.api_key
    
    def get_weather(self, location):
        """获取指定地点的天气信息
        
        Args:
            location: 地点名称
            
        Returns:
            dict: 天气信息
        """
        # 模拟天气API调用
        weather_data = {
            "location": location,
            "temperature": "25°C",
            "condition": "晴天",
            "humidity": "60%",
            "wind_speed": "10 km/h"
        }
        return weather_data
    
    def get_current_time(self):
        """获取当前时间
        
        Returns:
            str: 当前时间
        """
        import datetime
        now = datetime.datetime.now()
        return now.strftime("%Y-%m-%d %H:%M:%S")
    
    def generate_response(self, user_input):
        """根据用户输入生成响应
        
        Args:
            user_input: 用户的输入文本
            
        Returns:
            str: 智能体的响应文本
        """
        # 定义可用函数
        functions = [
            {
                "name": "get_weather",
                "description": "获取指定地点的天气信息",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "location": {
                            "type": "string",
                            "description": "地点名称，如北京、上海等"
                        }
                    },
                    "required": ["location"]
                }
            },
            {
                "name": "get_current_time",
                "description": "获取当前时间",
                "parameters": {
                    "type": "object",
                    "properties": {}
                }
            }
        ]
        
        # 构建消息数组
        messages = [
            {
                "role": "system",
                "content": "你是一个智能助手，能够回答用户的问题，并在需要时调用函数获取信息。"
            },
            {
                "role": "user",
                "content": user_input
            }
        ]
        
        try:
            # 调用OpenAI API
            response = openai.ChatCompletion.create(
                model="gpt-3.5-turbo",
                messages=messages,
                functions=functions,
                function_call="auto"
            )
            
            # 处理API响应
            response_message = response.choices[0].message
            
            # 检查是否需要调用函数
            if response_message.get("function_call"):
                function_name = response_message["function_call"]["name"]
                function_args = json.loads(response_message["function_call"]["arguments"])
                
                # 调用相应的函数
                if function_name == "get_weather":
                    weather_data = self.get_weather(function_args["location"])
                    function_response = json.dumps(weather_data)
                elif function_name == "get_current_time":
                    current_time = self.get_current_time()
                    function_response = json.dumps({"current_time": current_time})
                else:
                    function_response = json.dumps({"error": "未知函数"})
                
                # 将函数调用结果添加到消息中
                messages.append(response_message)
                messages.append({
                    "role": "function",
                    "name": function_name,
                    "content": function_response
                })
                
                # 再次调用API获取最终响应
                second_response = openai.ChatCompletion.create(
                    model="gpt-3.5-turbo",
                    messages=messages
                )
                
                return second_response.choices[0].message.content
            else:
                # 直接返回助手的回复
                return response_message.content
                
        except Exception as e:
            return f"发生错误: {str(e)}"

# 测试智能体
if __name__ == "__main__":
    # 创建智能体实例
    agent = FunctionCallingReflexAgent()
    
    print("助手: 你好！我是你的智能助手，能够回答你的问题，并提供天气信息和当前时间。有什么可以帮助你的吗？")
    
    # 测试对话
    while True:
        user_input = input("用户: ")
        if user_input.lower() in ["退出", "quit", "exit"]:
            print("助手: 再见！")
            break
        
        # 生成响应
        response = agent.generate_response(user_input)
        print(f"助手: {response}")

总结与思考

关键要点回顾

1. "反射"智能体的概念：基于规则和当前感知的智能体，根据输入直接生成响应，不考虑历史状态

2. OpenAI API的使用：如何设置API密钥，构建消息，调用API，处理响应

3. 消息格式与处理：系统消息、用户消息和助手消息的格式和处理方法

4. 错误处理：如何处理API调用可能出现的错误

5. 功能扩展：如何添加简单的规则匹配，支持不同的模型，配置参数等

6. 领域特定智能体：如何构建针对特定领域的专业智能体

7. 函数调用：如何使用函数调用扩展智能体的能力

实践建议

• 安全使用API密钥：不要在代码中硬编码API密钥，使用环境变量或配置文件管理

• 合理设置参数：根据实际需求调整温度、最大tokens等参数

• 添加错误处理：实现完善的错误处理机制，提高系统的稳定性

• 优化提示词：根据具体场景优化系统提示词，提高响应质量

• 测试不同模型：尝试使用不同的模型，比较其性能和效果

• 监控API使用：注意监控API的使用情况，避免超出配额或预算

• 考虑成本：不同模型和参数设置会影响API调用成本，需要在性能和成本之间找到平衡

未来学习方向

• 基于记忆的智能体：学习如何构建能够记住和使用历史信息的智能体

• 基于目标的智能体：学习如何构建能够设定目标并规划实现步骤的智能体

• 多智能体系统：学习如何构建多个智能体协作完成复杂任务的系统

• 工具使用：学习如何让智能体使用外部工具和服务

• RAG系统：学习如何构建检索增强生成系统，提高智能体的知识获取能力

• 部署和扩展：学习如何部署智能体到生产环境，处理并发请求

• 评估和优化：学习如何评估智能体的性能，持续优化其表现

通过本教程的学习，你已经掌握了使用原生OpenAI API构建简单"反射"智能体的基本技能。这是构建更复杂智能体的基础，也是理解AI智能体工作原理的重要一步。在接下来的课程中，我们将学习如何构建更高级的智能体，如基于记忆的智能体、能够使用工具的智能体等，逐步实现更强大、更实用的AI助手。