今天先和大家一起从零实现 AI 代理，只用到 Python 和 OpenAI

发布日期：2024-06-22 03:37:50 浏览次数： 3981

作者：数翼

微信搜一搜，关注“数翼”

说起 Agent 框架，大家可能觉得很复杂吧，其实核心逻辑很简单，简单的让你觉得不是 Agent。今天先和大家一起从零实现 AI 代理，只用到 Python 和 OpenAI。

ReAct

我们用 ReAct 来实现 AI 代理。 ReAct 是「Reason - Act」的意思。

简单的描述一下流程就是：

• 用户给出问题
• AI 分析原因
• 调用工具行动
• 观察行动结果

• 如果满足问题，那么就结束了
• 如果不满足，重复第二步，最后直到问题的解决。

实现

引入 OpenAI 和一些基础类库：

import openai
import re
import httpx
import os
from dotenv import load_dotenv

_ = load_dotenv()
from openai import OpenAI

创建 OpenAI 客户端，

client = OpenAI()

简答测试一下问答能力，

chat_completion = client.chat.completions.create(
    model="gpt-3.5-turbo",
    messages=[{"role": "user", "content": "Hello world"}]
)
chat_completion.choices[0].message.content

可能会输出：

'你好，您需要什么帮助？'

Agent 类

下面实现一个 Python Agent 代理类，我们后面的代理都基于这个类实现。 __call__ 表示对象本身的调用。

class Agent:
    def __init__(self, system=""):
        self.system = system
        self.messages = []
        if self.system:
            self.messages.append({"role": "system", "content": system})

    def __call__(self, message):
        self.messages.append({"role": "user", "content": message})
        result = self.execute()
        self.messages.append({"role": "assistant", "content": result})
        return result

    def execute(self):
        completion = client.chat.completions.create(
                        model="gpt-4o", 
                        temperature=0,
                        messages=self.messages)
        return completion.choices[0].message.content

我们使用 gpt-40 模型，并定义了一个 execute 方法，来调用大模型。

编写提示语，这个提示语完成了 ReAct 指令，我们如果要自行实现 Agent，都可以以这个提示语为模块进行修改和扩展：

prompt = """
你在一个思考、行动、暂停、观察的循环中运行。
在循环结束时，你输出一个答案
使用思考来描述你对所问问题的想法。
使用行动来运行其中一个可用的操作 - 然后返回暂停。
观察将是运行这些操作的结果。

你可用的操作是：

计算：
例如计算：4 * 7 / 3
运行计算并返回数字 - 使用 Python，因此请确保在必要时使用浮点语法

平均狗体重：
例如平均狗体重：牧羊犬
在给定品种的情况下返回狗的平均体重

示例会话：

问题：斗牛犬的体重是多少？
想法：我应该使用 average_dog_weight 来查看狗的体重
动作：average_dog_weight：斗牛犬
暂停

您将再次收到以下信息：

观察：斗牛犬重 51 磅

然后您输出：

答案：斗牛犬重 51 磅
""".strip()

定义 Action

思考和推理有了，记下来我们定一下动作。Agent 有了行动的能力，无论是执行函数还是调用工具， Agent 才完整。

定义两个动作进行运算或者计算平均体重：

• calculate
• average_dog_weight

把所有的动作到放到 known_actions 里面，这样我们就可以根据动作名称执行相应的动作了。

def calculate(what):
    return eval(what)

def average_dog_weight(name):
    if name in "Scottish Terrier": 
        return("Scottish Terriers average 20 lbs")
    elif name in "Border Collie":
        return("a Border Collies average weight is 37 lbs")
    elif name in "玩具贵宾犬":
        return("玩具贵宾犬的平均体重为 7 磅")
    else:
        return("An average dog weights 50 lbs")

known_actions = {
    "calculate": calculate,
    "average_dog_weight": average_dog_weight
}

Agent 实例

提问玩具贵宾犬的重量：

abot = Agent(prompt)
result = abot("玩具贵宾犬有多重？")
print(result)

根据我们的提示语，可以看到

想法：我应该使用平均狗体重动作来查找玩具贵宾犬的平均体重。
动作：average_dog_weight：玩具贵宾犬
暂停

可以看到，Agent 返回了动作 average_dog_weight，我们手动调用一下这个方法：

result = average_dog_weight("玩具贵宾犬")

得到输出：

玩具贵宾犬的平均体重为 7 磅

我们把观察到的结果发送给 Agent:

next_prompt = "Observation: {}".format(result)
abot(next_prompt)

查看最终结果 abot.messages[-1]：

{'role': 'assistant',
  'content': 'Answer: 玩具贵宾犬的平均体重为 7 磅'}

到这里，我们就重现了一次 ReAct 的整个过程，但是每种不足的是，调用现实世界函数是我们手动调用的。

也就是每次观察 LLM 输出，再反馈给 Agent 都要手动参与，也就是 average_dog_weight 这一步。

自动调用

如果把 Agent 支持的函数都改成自动调用，那么 Agent 不就可以自己做事情了么。

我们编写一个循环，用正则解析LLM回答，最大尝试次数为 5 ，自动进行上面的步骤。

action_re = re.compile('^Action: (\w+): (.*)$')   # python regular expression to selection action
def query(question, max_turns=5):
    i = 0
    bot = Agent(prompt)
    next_prompt = question
    while i < max_turns:
        i += 1
        result = bot(next_prompt)
        print(result)
        actions = [
            action_re.match(a) 
            for a in result.split('\n') 
            if action_re.match(a)
        ]
        if actions:
            # There is an action to run
            action, action_input = actions[0].groups()
            if action not in known_actions:
                raise Exception("Unknown action: {}: {}".format(action, action_input))
            print(" -- running {} {}".format(action, action_input))
            observation = known_actions[action](action_input)
            print("Observation:", observation)
            next_prompt = "Observation: {}".format(observation)
        else:
            return

再一次进行提问：

question = """我有两只狗，一只边境牧羊犬和一只苏格兰梗犬。
它们的总体重是多少"""
query(question)

可以看到大致如下的输出：

想法：我需要找到边境牧羊犬和苏格兰梗的平均体重，然后将它们加在一起得到总体重。
动作：average_dog_weight：边境牧羊犬
暂停
-- 运行 average_dog_weight 边境牧羊犬
观察：边境牧羊犬的平均体重为 37 磅
想法：现在我需要找到苏格兰梗的平均体重。
动作：average_dog_weight：苏格兰梗
暂停
-- 运行 average_dog_weight 苏格兰梗
观察：苏格兰梗平均体重 20 磅
想法：我现在知道了两只狗的平均体重。我将把它们加在一起得到总体重。
动作：计算：37 + 20
暂停
-- 运行计算 37 + 20
观察：57
答案：边境牧羊犬和苏格兰梗的总体重为 57 磅。

这个过程是不是很熟悉，和 LangChain 的调用过程基本一致。