实现一个AI Agent

2025-01-25

导入

Agent是什么？这段时间deepseek R1爆火，当大家使用deepseek R1进行聊天的时候，我们是一问一答的形式，和之前的chat GPT一样。这样的对话形式的AI有一个缺点，那就是不能和电脑本地环境进行交互。比如你让它们生成一段代码，它们直接给你代码，无法帮你在本地建立文件，然后写入，甚至运行测试，你只能拷贝它们给出的代码然后自己建立文件然后粘贴进去，并且编译或者解释运行。而Agent就是能够做到这一点，你可能想到了cursor，同样大火的一款IDE，没错，Cursor IDE就是一个Agent。

而实现一个Agent其实跟大模型本身没有关系，我们不需要修改任何的大模型部分，重点是prompt 提示词。可能不好理解，但是其实很简单，我举个例子就都懂了。比如：还是上面提到的生成代码任务，我现在想让大模型帮我生成一个小游戏代码，并且直接本地新建文件写入。我可以写类似这样的提示词，请在本地当前目录下新建一个文件，然后帮我生成一个html射击小游戏，直接写入这个文件。你可以这样调用本地工具<tool>write_file(path,content)</tool>，作用是把content内容写入path目录文件中，请等待工具执行结果（由我来确认），在获得工具执行结果前停止执行，获得结果后根据执行结果选择完成或者重新尝试。这样大模型就会以为本地有一个可用的写入文件的工具，并且按照<tool>write_file(path,content)</tool>格式进行调用，如果我回复调用成功，大模型就会认为文件写入成功，并且完成任务。而实际上大模型本身并没有调用工具和写入本地文件的能力，上面这行格式只是一行文字，一行字符串而已。那么谁来调用呢？开发者来调用，开发者编写Agent主程序解析大模型的返回，如果解析到<tool>write_file(path,content)</tool>这行，就调用本地的工具函数写入，然后返回写入成功，再次传入大模型，大模型看到执行成功，就认为完成了工具调用，然后可以继续完成下面的流程。

因此可以得到Agent的关键三要素大模型，Agent主程序，工具。

流程

从上面可以看到，要想实现Agent，我们必须严格大模型的输出格式和终止流程时机。我们可以在系统提示词system prompt中严格要求大模型，然后用户提示词user prompt中，输入用户任务要求，任务完成前每次大模型的输出都是辅助提示词asistant prompt。

而Agent的完整流程是：大模型进行思考thought，然后选择调用哪个工具tool，根据返回结果继续思考thought，如果返回成功就输出完成，否则继续调用工具tool，如此往复循环。

一个系统提示词system prompt的示例如下：

请按照如下流程解决任务：
首先进行思考，使用<thought>标签包围思考内容。然后决定使用哪个可用工具，用<tool>标签包围工具。接着，将根据工具的返回结果得到<observation>标签包围的观察结果。再根据观察结果继续思考，如此循环，直到可以给出<answer>标签包围的最终答案。

例子:
<question>中国首都是哪里？</question>
<thought>我需要找到中国的首都，可以使用搜索工具。</thought>
<tool>get_capital("中国")</tool>
<observation>中国首都是北京。</observation>
<thought>搜索结果显示中国首都是北京，得到答案。</thought>
<answer>中国首都是北京。</answer>

请严格遵守：
- 输出格式严格按照XML格式输出
- 你每次回答都必须包括<thought>标签 以及 <tool>标签 或 <answer>标签
- 输出 <tool> 后立即停止生成，等待工具返回的 <observation>，自己不要生成<observation>
- 工具参数中的文件路径请使用绝对路径

可用工具列表：
- read_file(file_path): 用于读取文件内容
- write_file(file_path, content): 将指定内容写入指定文件
- run_command(command): 用于执行终端命令

环境信息：
操作系统：Windows

从上面提示词可以总结出代码的执行流程：

上面的系统提示词 + user用户提示词（+工具返回观察<observation>，如果不是第一次的话）传入大模型（使用openai库和openrouter API即可）
获取大模型返回，一定包含<thought>标签和<tool>标签或者<answer>标签，此时表示任务完成
检测返回是否包含<answer>标签，是则直接返回结果，不是则继续检测<tool>标签
解析<tool>标签包围的内容，获取使用到的工具函数名和参数，并且本地调用函数，如果是终端命令工具，则设置需要用户输入y同意才能进行，因为比较危险，工具执行结果用<observation>标签包围后一起传入到大模型中，继续这个循环即可

实现Agent

主循环按照上面的流程即可,这也是最核心的代码：

为了读者更直观，我再强调下，大模型返回的数据形式大概是什么样的：

<thought>用户要求在当前目录下新建一个文件夹，并在其中创建一个HTML射击小游戏。
我需要先创建文件夹，然后编写HTML文件。首先使用run_command命令创建文件夹。</thought>
<tool>run_command(mkdir game_folder)</tool>
<thought>
文件夹创建成功。现在需要在game_folder目录下创建一个HTML射击游戏。
我需要编写一个包含基本射 击游戏功能的HTML文件。我将创建一个简单的canvas射击游戏。
</thought>

<tool>write_file("physics_demo/ball_physics.py", "
<!DOCTYPE html>
<html>
<head>
     <title>射击小游戏</title>
....省略
</body>
</html>
“)</tool>
<thought>
HTML射击游戏文件已经成功创建在game_folder目录中。
。。。。。。省略
</thought>

对着上面的形式，下面的正则表达式应该就十分清晰了。

def run(self, user_input: str):
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"<question>{user_input}</question>"}
        ]

        while True:
            content = self.call_model(messages)

            thought_match = re.search(r"<thought>(.*?)</thought>", content, re.DOTALL)
            if thought_match:
                thought = thought_match.group(1)
                print(f"\n Thought: {thought}")

            # 检测模型是否输出最终答案
            if "<answer>" in content:
                final_answer = re.search(r"<answer>(.*?)</answer>", content, re.DOTALL)
                return final_answer.group(1)

            # 检测工具使用
            tool_found = re.search(r"<tool>(.*?)</tool>", content, re.DOTALL)
            if not tool_found:
                raise RuntimeError("模型未输出 <tool>")
            tool = tool_found.group(1)
            tool_name, args = self.parse_toolfunc(tool)

            print(f"\n Tool: {tool_name}({', '.join(args)})")
            # 终端命令需要询问用户
            should_continue = input(f"\n\n是否继续？（Y/N）") if tool_name == "run_command" else "y"
            if should_continue.lower() != 'y':
                print("\n 操作已取消。")
                return "操作被用户取消"

            try:
                observation = self.tools[tool_name](*args)
            except Exception as e:
                observation = f"工具执行错误：{str(e)}"
            print(f"\n Observation：{observation}")
            obs_msg = f"<observation>{observation}</observation>"
            messages.append({"role": "user", "content": obs_msg})

调用大模型，使用openai库即可，deepseek/deepseek-chat-v3.1:free这个免费模型，当然使用收费模型效果更好，在openrouter中搜索这个模型，就可以看到API调用格式：

from openai import OpenAI

client = OpenAI(
  base_url="https://openrouter.ai/api/v1",
  api_key="<OPENROUTER_API_KEY>",
)

completion = client.chat.completions.create(
  extra_headers={
    "HTTP-Referer": "<YOUR_SITE_URL>", # Optional. Site URL for rankings on openrouter.ai.
    "X-Title": "<YOUR_SITE_NAME>", # Optional. Site title for rankings on openrouter.ai.
  },
  extra_body={},
  model="deepseek/deepseek-chat-v3.1:free",
  messages=[
    {
      "role": "user",
      "content": "What is the meaning of life?"
    }
  ]
)
print(completion.choices[0].message.content)

就参考这个形式即可。解析工具函数的名字和参数，可以使用ast模块，也很简单：

def parse_toolfunc(self, code_str):
        """工具函数名和参数解析"""
        try:
            # 使用ast模块解析函数调用
            tree = ast.parse(code_str.strip(), mode='eval')
            if not isinstance(tree.body, ast.Call):
                raise ValueError("不是有效的函数调用")

            func_name = tree.body.func.id
            args = []
            for arg in tree.body.args:
                # 对每个参数节点求值
                try:
                    arg_value = ast.literal_eval(arg)
                except (ValueError, SyntaxError):
                    # 处理无法直接求值的复杂表达式
                    arg_value = ast.get_source_segment(code_str, arg)
                args.append(arg_value)

            return func_name, args
        except SyntaxError:
            raise ValueError("Invalid function call syntax")

工具函数则是：

def read_file(file_path):
    """用于读取文件内容"""
    with open(file_path, "r", encoding="utf-8") as f:
        return f.read()

def write_file(file_path, content):
    """将指定内容写入指定文件"""
    with open(file_path, "w", encoding="utf-8") as f:
        f.write(content.replace("\\n", "\n"))
    return "写入成功"

def run_command(command):
    """用于执行终端命令"""
    import subprocess
    run_result = subprocess.run(command, shell=True, capture_output=True, text=True)
    return "执行成功" if run_result.returncode == 0 else run_result.stderr

完整代码

系统提示词文件system_prompt.py:

system_prompt = """
请按照如下流程解决任务：
首先进行思考，使用<thought>标签包围思考内容。然后决定使用哪个可用工具，用<tool>标签包围工具。接着，将根据工具的返回结果得到<observation>标签包围的观察结果。再根据观察结果继续思考，如此循环，知道可以给出<answer>标签包围的最终答案。

例子:
<question>中国首都是哪里？</question>
<thought>我需要找到中国的首都，可以使用搜索工具。</thought>
<tool>get_capital("中国")</tool>
<observation>中国首都是北京。</observation>
<thought>搜索结果显示中国首都是北京，得到答案。</thought>
<answer>中国首都是北京。</answer>

请严格遵守：
- 输出格式严格按照XML格式输出
- 你每次回答都必须包括<thought>标签 以及 <tool>标签 或 <answer>标签
- 输出 <tool> 后立即停止生成，等待工具返回的 <observation>，自己不要生成<observation>
- 工具参数中的文件路径请使用绝对路径

可用工具列表：
- read_file(file_path): 用于读取文件内容
- write_file(file_path, content): 将指定内容写入指定文件
- run_command(command): 用于执行终端命令

环境信息：
操作系统：Windows
"""

工具类文件tools.py：

def read_file(file_path):
    """用于读取文件内容"""
    with open(file_path, "r", encoding="utf-8") as f:
        return f.read()

def write_file(file_path, content):
    """将指定内容写入指定文件"""
    with open(file_path, "w", encoding="utf-8") as f:
        f.write(content.replace("\\n", "\n"))
    return "写入成功"

def run_command(command):
    """用于执行终端命令"""
    import subprocess
    run_result = subprocess.run(command, shell=True, capture_output=True, text=True)
    return "执行成功" if run_result.returncode == 0 else run_result.stderr

tools_list = [read_file, write_file, run_command]

Agent主文件agent.py：（替换成你自己的API key即可，因为API提供商非常之多，如果不用openrouter的话，而使用比如硅基流动之类的话，替换对应的base_url和你要使用的model即可，这里就以免费的deepseek-chat-v3.1 API为例）

import ast
import os
import re
from openai import OpenAI
from system_prompt import system_prompt
import tools


class Agent:
    def __init__(self, tools):
        self.tools = { func.__name__: func for func in tools }
        self.model = "deepseek/deepseek-chat-v3.1:free"
        self.client = OpenAI(
            base_url="https://openrouter.ai/api/v1",
            api_key="替换成你的APIkey",
        )

    def run(self, user_input: str):
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"<question>{user_input}</question>"}
        ]

        while True:
            content = self.call_model(messages)

            thought_match = re.search(r"<thought>(.*?)</thought>", content, re.DOTALL)
            if thought_match:
                thought = thought_match.group(1)
                print(f"\n Thought: {thought}")

            # 检测模型是否输出最终答案
            if "<answer>" in content:
                final_answer = re.search(r"<answer>(.*?)</answer>", content, re.DOTALL)
                return final_answer.group(1)

            # 检测工具使用
            tool_found = re.search(r"<tool>(.*?)</tool>", content, re.DOTALL)
            if not tool_found:
                raise RuntimeError("模型未输出 <tool>")
            tool = tool_found.group(1)
            tool_name, args = self.parse_toolfunc(tool)

            print(f"\n Tool: {tool_name}({', '.join(args)})")
            # 终端命令需要询问用户
            should_continue = input(f"\n\n是否继续？（Y/N）") if tool_name == "run_command" else "y"
            if should_continue.lower() != 'y':
                print("\n 操作已取消。")
                return "操作被用户取消"

            try:
                observation = self.tools[tool_name](*args)
            except Exception as e:
                observation = f"工具执行错误：{str(e)}"
            print(f"\n Observation：{observation}")
            obs_msg = f"<observation>{observation}</observation>"
            messages.append({"role": "user", "content": obs_msg})

    def call_model(self, messages):
        print("\n 正在调用大模型，请稍等...")
        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
        )
        content = response.choices[0].message.content
        messages.append({"role": "assistant", "content": content})
        return content

    def parse_toolfunc(self, code_str):
        """工具函数名和参数解析"""
        try:
            # 使用ast模块解析函数调用
            tree = ast.parse(code_str.strip(), mode='eval')
            if not isinstance(tree.body, ast.Call):
                raise ValueError("不是有效的函数调用")

            func_name = tree.body.func.id
            args = []
            for arg in tree.body.args:
                # 对每个参数节点求值
                try:
                    arg_value = ast.literal_eval(arg)
                except (ValueError, SyntaxError):
                    # 处理无法直接求值的复杂表达式
                    arg_value = ast.get_source_segment(code_str, arg)
                args.append(arg_value)

            return func_name, args
        except SyntaxError:
            raise ValueError("Invalid function call syntax")


def main():
    agent = Agent(tools=tools.tools_list)

    task = input("输入任务描述：")

    final_answer = agent.run(task)

    print(f"\n 最终答案：{final_answer}")

if __name__ == "__main__":
    main()

结果

运行python agent.py即可,用一个简单的用户任务为例：帮我在当前目录下新建一个文件夹，然后在新建文件夹内写一个html的射击小游戏。

输出如下：

输入任务描述：帮我在当前目录下新建一个文件夹，然后在新建文件夹内写一个html的射击小游戏

 正在调用大模型，请稍等...

 Thought: 用户要求在当前目录下新建一个文件夹，并在其中创建一个HTML射击小游戏。我需要先创建文件夹，然后编写HTML文件。首先使用run_command命令创建文件夹。

 Tool: run_command(mkdir game_folder)


是否继续？（Y/N）y

 Observation：执行成功

 正在调用大模型，请稍等...

 Thought: 文件夹创建成功。现在需要在game_folder目录下创建一个HTML射击游戏。我需要编写一个包含基本射 击游戏功能的HTML文件。我将创建一个简单的canvas射击游戏。

 Tool: write_file(game_folder/shooting_game.html, <!DOCTYPE html>
<html>
<head>
    <title>射击小游戏</title>

        ... //中间省略

        // 游戏循环
        function gameLoop() {
            update();
            draw();
            requestAnimationFrame(gameLoop);
        }

        // 启动游戏
        gameLoop();
    </script>
</body>
</html>)

 Observation：写入成功

 正在调用大模型，请稍等...

 Thought: HTML射击游戏文件已经成功创建在game_folder目录中。游戏包含基本的射击功能：玩家可以使用左右 箭头键移动，空格键射击，敌人会从上方生成，击中敌人可以得分，碰到敌人则游戏结束。现在可以给用户确认完成。

 最终答案：已完成！在当前目录下创建了game_folder文件夹，并在其中创建了一个HTML射击小游戏。游戏功能包括：
- 使用左右箭头键移动玩家
- 空格键射击
- 敌人从上方生成
- 击中敌人得分
- 碰到敌人游戏结束

您可以通过浏览器打开 game_folder/shooting_game.html 文件来玩游戏。

可以看到大模型首先思考，并且返回工具调用请求，即run_command(mkdir game_folder)，也就是生成一个新的文件夹，并且征求我们的意见，因为是命令行工具，同意后，生成html代码，然后写入文件，然后给出最终答案。（其中代码部分太长，中间省略）

游戏结果：

可以看到游戏运行良好。

这个例子比较简单，Agent只调用了2次工具，让我们再用一个调用一点的例子看一看，比如：在当前目录下新建一个文件夹，在新建文件夹内帮我实现一个项目，用python编写一个小球落地的物理效果演示

Thought: 用户要求在当前目录下创建一个新文件夹，并在其中用Python实现一个小球落地的物理效果演示。我需要先创建文件夹，然后编写Python代码来实现物理模拟。首先，我应该使用run_command来创建文件夹。
 Tool: run_command(mkdir physics_demo)
是否继续？（Y/N）y
 Observation：执行成功
 正在调用大模型，请稍等...
 Thought: 文件夹已创建成功。现在需要在physics_demo文件夹中创建一个Python文件来实现小球落地的物理效果。我将使用write_file来创建并写入Python代码。代码应该模拟小球受重力影响下落并在触地时反弹的物理效果，可以使用pygame库来实现图形化演示。
 Tool: write_file(physics_demo/ball_physics.py, import pygame
import sys
# 初始化pygame
pygame.init() 
...#省略
pygame.quit()
sys.exit()
 Thought: 文件内容读取成功，代码看起来正确。现在让我创建一个README文件来说明如何运行这个项目，并测试一下代码是否能正常运行。
 Tool: write_file(physics_demo/README.md, # 小球落地物理演示
## 项目描述
这是一个使用Python和Pygame库实现的小球落地物理效果演示程序
。 。。。。。省略
Observation：写入成功
 正在调用大模型，请稍等... 
Thought: 让我使用更详细的命令来查看physics_demo文件夹中的具体文件，确认ball_physics.py文件确实存在。
Tool: run_command(dir physics_demo/a)
Observation：执行成功
正在调用大模型，请稍等...
Thought: 让我使用更具体的命令来查看physics_demo文件夹的内容，确认文件确实创建成功。
Tool: run_command(dir physics_demo\*.*)
Observation：执行成功
 Thought: README文件已创建成功。现在让我测试一下程序是否能正常运行。首先需要检查当前目录是否安装了pygame，如果没有的话需要安装。让我先尝试运行Python程序。
 Tool: run_command(cd physics_demo && python ball_physics.py)
是否继续？（Y/N）y  
.....省略

可以看到不仅生成了代码文件，还生成了README.md说明文件，并且还尝试不同的方法确认项目文件夹和文件是否存在，并且自动运行了测试。

运行效果如下，可以看到运行也很不错：

总结

主要讲解了Agent这个技术的底层原理，实则就是大模型通过Agent主流程调用工具，但是就可以有比单纯的Chat更强的交互感，算是提示词工程的一次进化，由prompt engineering 提示词工程进化成agent engineering 智能体工程。且随着可用工具的增加，Agent可以变得非常强大，比如网络搜索工具，可以联机搜索更多的信息；文生图工具，可以根据提示词本地生成图片；不同格式文件分析器工具，可以分析不同文件的内容等。以及MCP协议，专门标准化大模型和工具之间的接口，虽然MCP本质上还是一个程序，比如python程序或者nodejs程序，上面演示的tool.py模块其实就是MCP的雏形，只不过是直接import模块调用，而MCP是是通过各种协议接口调用，比如stdio,SSE,streamableHttp，但是因为标准化了接口和数据交换格式，使得各种软件都可以实现MCP作为工具，比如支持MCP的地图软件，可以作为可用工具被大模型调用，使得大模型可以为你规划路线等。并且工具类型和大模型类型的倾向不同，不同Agent的适用领域也不同，金融领域，编程领域，日常生活领域等等不同的Agent可以更具针对性。

之前说，AI大模型使用工具，其实就是提示词，其实不准确，应该说提示词是实现调用工具的一种最直接，也最巧妙的方式，就像上面的代码例子一样，也可以说是相对早期的一个方式，因为随着工具的增多，单纯在系统提示词中添加工具函数列表效率肯定是低下的，因此肯定会被替代。AI Agent调用工具还可以是Function Calling的方式，这是OpenAI提出的一种方式，大模型会自动调用工具，不需要添加到提示词中，只需要在API调用中设置好即可。可以认为Function Calling, MCP是更加现代的Agent实现工具调用的方式，而system prompt提示词是早期方式。

poi～

Star

实现一个AI Agent

导入

流程

实现Agent

完整代码

结果

总结