纯Python手搓AI Agent:绕开LangGraph与AutoGPT的自主

本文揭示了一个惊人事实：LangGraph和AutoGPT等复杂框架的核心功能，仅需100-200行纯Python代码即可实现70-80%的效果。我们将深入探讨AI Agent的本质循环机制，并提供一个完整的轻量级实现方案。

引言

AI Agent正成为2024年最热门的技术趋势之一。从AutoGPT到LangGraph，各种框架如雨后春笋般涌现。这些框架承诺让AI系统能够自主规划、决策和执行任务。

然而，这些框架的复杂度令人望而却步。数百页文档、繁琐配置、陡峭学习曲线，使得许多开发者望而却步。

本文将打破这一困境。我们将证明，核心Agent逻辑其实非常简单。一个精心设计的纯Python实现，可以在极短代码量下实现关键功能。

Agent的本质

Agent的核心是一个无限循环。在这个循环中：

1. LLM接收当前状态
2. 决定下一步行动
3. 执行行动并获取结果
4. 将结果反馈给LLM
5. 重复直到任务完成

这就是Agent的本质。LangGraph和AutoGPT不过是围绕这个核心逻辑添加了状态管理、工具集成和错误处理等高级功能。

纯Python实现方案

让我们开始构建一个轻量级Agent。这个实现将包含核心循环、工具调用和状态管理三个关键部分。

环境准备

首先安装必要的依赖：

pip install openai python-dotenv

创建一个.env文件存储API密钥：

OPENAI_API_KEY=your_api_key_here

核心代码实现

以下是一个完整的Agent实现，仅需150行代码：

import os
import json
import openai
from typing import List, Dict, Any, Optional
from dotenv import load_dotenv

load_dotenv()
openai.api_key = os.getenv("OPENAI_API_KEY")

class SimpleAgent:
    def __init__(self, tools: List[Dict[str, Any]], system_prompt: str = ""):
        self.tools = tools
        self.system_prompt = system_prompt
        self.chat_history = []

    def run(self, user_input: str, max_iterations: int = 10) -> str:
        self.chat_history = [{"role": "system", "content": self.system_prompt}]
        self.chat_history.append({"role": "user", "content": user_input})

        for _ in range(max_iterations):
            response = self._get_llm_response()

            if self._is_final_answer(response):
                return self._extract_final_answer(response)

            action = self._extract_action(response)

            if action is None:
                return "无法确定下一步行动。"

            tool_name = action.get("name")
            tool_params = action.get("parameters", {})

            if tool_name == "finish":
                return tool_params.get("response", "任务完成。")

            tool_result = self._execute_tool(tool_name, tool_params)

            self.chat_history.append({
                "role": "assistant",
                "content": json.dumps(action)
            })

            self.chat_history.append({
                "role": "tool",
                "content": f"工具执行结果: {tool_result}"
            })

        return "达到最大迭代次数，任务未完成。"

    def _get_llm_response(self) -> str:
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=self.chat_history
        )
        return response.choices[0].message['content']

    def _is_final_answer(self, response: str) -> bool:
        return "FINAL_ANSWER:" in response

    def _extract_final_answer(self, response: str) -> str:
        return response.split("FINAL_ANSWER:")[1].strip()

    def _extract_action(self, response: str) -> Optional[Dict[str, Any]]:
        try:
            start = response.find("{")
            end = response.rfind("}") + 1
            if start == -1 or end == 0:
                return None
            action = json.loads(response[start:end])
            return action
        except:
            return None

    def _execute_tool(self, tool_name: str, params: Dict[str, Any]) -> str:
        for tool in self.tools:
            if tool["name"] == tool_name:
                try:
                    result = tool["function"](**params)
                    return str(result)
                except Exception as e:
                    return f"工具执行错误: {str(e)}"
        return f"未找到工具: {tool_name}"

工具定义

让我们定义一些实用工具：

def calculator(expression: str) -> float:
    """计算数学表达式"""
    return eval(expression)

def web_search(query: str) -> str:
    """模拟网络搜索"""
    return f"搜索结果: 关于'{query}'的信息..."

def file_reader(file_path: str) -> str:
    """读取文件内容"""
    try:
        with open(file_path, 'r') as f:
            return f.read()
    except:
        return f"无法读取文件: {file_path}"

tools = [
    {
        "name": "calculator",
        "description": "计算数学表达式",
        "parameters": {
            "type": "object",
            "properties": {
                "expression": {
                    "type": "string",
                    "description": "要计算的数学表达式"
                }
            },
            "required": ["expression"]
        },
        "function": calculator
    },
    {
        "name": "web_search",
        "description": "搜索网络信息",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "搜索查询"
                }
            },
            "required": ["query"]
        },
        "function": web_search
    },
    {
        "name": "file_reader",
        "description": "读取文件内容",
        "parameters": {
            "type": "object",
            "properties": {
                "file_path": {
                    "type": "string",
                    "description": "文件路径"
                }
            },
            "required": ["file_path"]
        },
        "function": file_reader
    }
]

Agent提示词设计

设计一个有效的系统提示词至关重要：

system_prompt = """
你是一个智能助手，可以使用以下工具完成任务：

1. calculator - 计算数学表达式
2. web_search - 搜索网络信息
3. file_reader - 读取文件内容

当你需要执行工具时，请按照以下JSON格式回应：
{
    "name": "工具名称",
    "parameters": {
        "参数名": "参数值"
    }
}

如果你已经完成任务，请使用finish工具：
{
    "name": "finish",
    "parameters": {
        "response": "最终答案"
    }
}

如果你不确定如何回答，请直接提供文本回应。
"""

实战演示

让我们测试这个简单的Agent：

agent = SimpleAgent(tools, system_prompt)

# 示例1: 数学计算
result = agent.run("计算 15 * 25 + 100 等于多少？")
print(result)

# 示例2: 信息检索
result = agent.run("搜索关于Python编程的最新趋势")
print(result)

# 示例3: 文件处理
# 创建一个测试文件
with open("test.txt", "w") as f:
    f.write("这是一个测试文件，用于演示Agent的文件读取功能。")

result = agent.run("读取test.txt文件的内容")
print(result)

性能对比分析

我们的纯Python实现与成熟框架相比有以下优缺点：

优势

1. 极简部署：仅需3个依赖，无需复杂配置。
2. 高度可控：代码完全透明，可自由定制。
3. 轻量高效：启动速度快，资源占用少。
4. 易于调试：逻辑清晰，问题定位简单。

劣势

1. 功能有限：缺少高级特性如多Agent协作、持久化存储。
2. 错误处理：需要手动实现各种边界情况处理。
3. 扩展性：添加复杂功能需要更多代码。
4. 性能优化：缺少专业优化，处理复杂任务效率较低。

应用场景

我们的轻量级Agent非常适合以下场景：

1. 个人项目：快速构建原型系统验证想法。
2. 学习研究：深入理解Agent原理。
3. 简单任务：处理结构化明确的重复性工作。
4. 资源受限：在低配置环境中运行。
5. 定制需求：需要高度定制化的特殊应用。

高级扩展

虽然我们的基础实现简洁，但可以轻松添加更多功能：

状态持久化

import pickle

class PersistentAgent(SimpleAgent):
    def __init__(self, tools, system_prompt, state_file="agent_state.pkl"):
        super().__init__(tools, system_prompt)
        self.state_file = state_file
        self.load_state()

    def save_state(self):
        with open(self.state_file, "wb") as f:
            pickle.dump({
                "chat_history": self.chat_history,
                "tools": self.tools
            }, f)

    def load_state(self):
        try:
            with open(self.state_file, "rb") as f:
                state = pickle.load(f)
                self.chat_history = state["chat_history"]
                self.tools = state["tools"]
        except:
            pass

    def run(self, user_input: str, max_iterations: int = 10) -> str:
        result = super().run(user_input, max_iterations)
        self.save_state()
        return result

并行工具调用

from concurrent.futures import ThreadPoolExecutor

class ParallelAgent(SimpleAgent):
    def _execute_tools(self, actions: List[Dict[str, Any]]) -> List[str]:
        results = []
        with ThreadPoolExecutor() as executor:
            futures = []
            for action in actions:
                tool_name = action.get("name")
                tool_params = action.get("parameters", {})
                future = executor.submit(self._execute_tool, tool_name, tool_params)
                futures.append(future)

            for future in futures:
                results.append(future.result())

        return results

未来展望

虽然我们的纯Python实现已经具备核心功能，但仍有巨大发展空间：

1. 多模态支持：集成图像、音频等多模态处理能力。
2. 长记忆：实现长期记忆存储和检索机制。
3. 自我优化：添加性能监控和自动调优功能。
4. 分布式架构：支持多节点协作处理复杂任务。

结论

通过本文，我们证明了纯Python手搓AI Agent不仅可行，而且具有显著优势。在特定场景下，这种轻量级实现甚至比重量级框架更合适。

关键在于理解Agent的本质——一个简单的循环决策过程。围绕这一核心，我们可以构建高度定制、资源高效的AI系统。

记住，框架只是工具，理解底层原理才是构建强大AI应用的关键。在这个快速发展的领域，掌握基础实现比依赖现成框架更具长期价值。

AI Agent, LangGraph, AutoGPT, Python实现, 轻量级架构, 开源项目, 技术实现, 代码实战, AI框架对比, 自主开发