从零构建智能体:只是一个循环

对我来说,理解一个新概念的最佳方式就是构建它并向别人解释。这篇文章两者兼顾。我将实验故事与实用教程结合在一起,我相信你会觉得它有用。

从零构建智能体:只是一个循环
微信 ezpoda免费咨询:AI编程 | AI模型微调| AI私有化部署
AI模型价格对比 | AI工具导航 | ONNX模型库 | Tripo 3D | Meshy AI | ElevenLabs | KlingAI | ArtSpace | Phot.AI | InVideo

我每天都在使用Claude、Codex、Cursor、Gemini、Copilot或Junie,但我仍然无法指出"聊天机器人"在哪一行变成了"智能体",我无法解释是什么让它们成为智能体,所以我自己从头写了一个简单版本来找出答案。

对我来说,理解一个新概念的最佳方式就是构建它并向别人解释。这篇文章两者兼顾。我将实验故事与实用教程结合在一起,我相信你会觉得它有用。

我们将从仅仅50行Python代码开始,连接OpenAI,通过Ollama切换到本地模型,构建一个同时使用两者的混合模式,添加工具,实现MCP,最后与Claude CLI进行比较。到最后,你将确切地看到底层发生了什么。

不用LangChain。不用LangGraph。不用CrewAI。只需要Python、一个LLM和一个while循环。

1、我们要构建什么(规格说明)

在构建某样东西之前,你必须定义它是什么以及它做什么。

AI智能体是一个程序,它:

  1. 接受用户的高级任务
  2. 推理下一步该做什么
  3. 采取行动(调用工具、搜索网络、读取文件)
  4. 观察结果
  5. 决定是继续还是返回最终答案
  6. 维护对话历史,使每个决策都建立在之前的基础上

普通的LLM调用是一个一次性操作:你发送提示,得到回复,结束。智能体不同,因为它会循环。它接收一个高级任务,推理下一步该做什么,采取行动,观察结果,并持续下去直到任务完成。

思考、行动、观察、决策的重复循环就是将语言模型转变为智能体的东西。

当今大多数智能体遵循一种称为ReAct(推理+行动)的模式。LLM不会直接跳到最终答案。它会产生一个关于下一步做什么的思考,然后是一个行动(工具调用),然后等待观察结果,再决定下一步做什么。

模型没有意识。在任何有意义的意义上都没有自我反思。它拥有的是对话历史——它采取的每个行动和获得的每个结果——都位于上下文窗口中。ReAct模式将其转化为表现得像自我反思和自我纠正的东西。而且它有效。

每个循环发生的事情如下:

  1. 你将当前对话发送给LLM:系统提示、用户消息和任何先前的工具结果
  2. LLM返回一个最终答案或它想要进行的工具调用列表
  3. 如果是最终答案,你就完成了
  4. 如果是工具调用,你执行它们,将结果附加到对话中,然后回到步骤1

这就是整个架构。

2、最小实现(云端API作为大脑)

首先,我们使用云端API作为大脑。我选择OpenAI是因为它有最干净的工具调用接口,但这适用于任何兼容OpenAI的API。

核心智能体机制就是这样:

def run_agent(task: str, client: OpenAI, model: str = "gpt-4o-mini") -> str:
    messages = [
        {
            "role": "system",
            "content": (
                "You are a helpful assistant. Use tools when needed. "
                "When you have a final answer, respond without calling any tools."
            ),
        },
        {"role": "user", "content": task},
    ]

    while True:
        response = client.chat.completions.create(
            model=model,
            messages=messages,
            tools=TOOLS,
            tool_choice="auto",
        )

        message = response.choices[0].message
        messages.append(message)

        # This is the decision point: does the model have an answer, or does it need tools?
        if not message.tool_calls:
            return message.content

        # If we reach here, the model called one or more tools
        for tool_call in message.tool_calls:
            name = tool_call.function.name
            args = json.loads(tool_call.function.arguments)

            print(f"  > calling {name}({args})")

            fn = TOOL_FUNCTIONS.get(name)
            result = fn(**args) if fn else f"Unknown tool: {name}"

            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result,
            })
        # End of iteration — go back to the top of the while loop

关键行是if not message.tool_calls。如果模型返回文本而没有请求任何工具,它在发出信号表示它拥有回答所需的一切。智能体退出并返回该文本。如果模型请求工具,智能体执行它们,将结果附加到对话历史中,并将所有内容发送回模型进行下一轮。

3、定义工具

三个简单的工具来具体化:当前日期/时间、计算器和天气存根:

import json
import os
from datetime import datetime
from openai import OpenAI


def get_current_date() -> str:
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")


def calculate(expression: str) -> str:
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return str(result)
    except Exception as e:
        return f"Error: {e}"


def get_weather(city: str) -> str:
    # Replace with a real weather API call
    return f"Weather in {city}: 72°F, partly cloudy"


TOOL_FUNCTIONS = {
    "get_current_date": get_current_date,
    "calculate": calculate,
    "get_weather": get_weather,
}

运行它:

if __name__ == "__main__":
    client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

    task = "What's today's date? Also, what is 15% of 847? And what's the weather in Tokyo?"
    print(f"Task: {task}\n")
    answer = run_agent(task, client)
    print(f"\nAnswer: {answer}")

输出:

Task: What's today's date? Also, what is 15% of 847? And what's the weather in Tokyo?

  > calling get_current_date({})
  > calling calculate({'expression': '847 * 0.15'})
  > calling get_weather({'city': 'Tokyo'})

Answer: Today is 2026-04-30 09:14:22. 15% of 847 is 127.05.
The weather in Tokyo is 72°F and partly cloudy.

在第一轮中,LLM识别了它需要的所有三个工具,逐一调用它们,获得了结果,并组装了最终答案。没有框架。没有编排层。

4、通过Ollama将云端API替换为本地LLM

Ollama暴露了一个兼容OpenAI的API,这意味着相同的智能体代码只需一个更改就可以在本地模型上运行:

ollama_client = OpenAI(
    base_url="http://localhost:11434/v1",
    api_key="ollama",  # required by the client library, ignored by Ollama
)

answer = run_agent(task, ollama_client, model="qwen2.5")

就是这样。代码完全不知道它是在与OpenAI的服务器通信还是与运行在你机器上的模型通信。

5、构建混合模式(本地编排,云端委派)

你可以在本地编排,只在任务真正需要时才为云端调用付费:

def ask_cloud_expert(question: str) -> str:
    """Delegate complex questions to a cloud model."""
    cloud_client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
    response = cloud_client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": question}],
    )
    return response.choices[0].message.content

现在当你运行:

answer = run_agent(
    task="What's 2+2? Also, explain the philosophical implications of the Ship of Theseus paradox.",
    client=ollama_client,
    model="qwen2.5"
)

本地模型处理2+2(通过计算器工具),意识到哲学问题超出了它的能力范围,然后调用ask_cloud_expert()从GPT-4获得适当的答案。你只需为一个云端API调用付费,而不是几十个。

6、添加更多工具

让我们用展示真实世界能力的工具来扩展智能体:

from pathlib import Path

def web_search(query: str) -> str:
    # Stub — replace with Brave Search API, SerpAPI, or Tavily
    return (
        f"Search results for '{query}':\n"
        f"1. Wikipedia: comprehensive overview\n"
        f"2. Recent article: explained in 5 minutes\n"
        f"3. Official docs"
    )

def read_file(path: str) -> str:
    return Path(path).read_text()

def write_file(path: str, content: str) -> str:
    Path(path).write_text(content)
    return f"wrote {len(content)} chars to '{path}'"

有了这六个工具,智能体现在可以回答需要当前信息的问题、执行计算、查看天气、搜索网络,以及读写文件。

7、MCP客户端:从外部服务器发现工具

MCP(模型上下文协议),由Anthropic于2024年11月推出,是解决工具共享问题的标准。它定义了一种统一的方式,让任何智能体都能发现和调用来自任何服务器的工具。

你的DIY智能体变成了一个MCP客户端。你不是硬编码工具定义,而是调用服务器并获取它暴露的任何工具。

智能体逻辑不变。改变的是工具的来源和维护者。

8、MCP服务器:将你的工具暴露给任何智能体

这是一个暴露两个工具的完整MCP服务器:

# mcp_server.py — a real MCP server in 10 lines
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("mini-tools")

@mcp.tool()
def to_uppercase(text: str) -> str:
    """Convert text to uppercase."""
    return text.upper()

@mcp.tool()
def count_words(text: str) -> int:
    """Count the number of words in a string."""
    return len(text.split())

if __name__ == "__main__":
    mcp.run()

任何兼容MCP的智能体现在都可以使用你的工具——Claude Desktop、Cursor、你的DIY智能体,任何人。

9、与Claude CLI的比较

Claude Code是生产工具。我的智能体是学习工具。这是诚实的比较。

Claude Code能做我的智能体做不到的事情:当任务很大时,它会启动具有隔离上下文窗口的子智能体;在运行破坏性命令之前会提示确认;跨会话维护持久记忆;用调整后的参数重试失败的工具调用;当接近上下文限制时压缩先前的消息。我的智能体都不做这些。

我的智能体拥有的:我可以阅读它的每一行代码。当出现问题时,我确切知道该看哪里。我可以用Ollama完全离线运行它,或者连接混合模式,只按需为云端调用付费。

如果我需要交付可靠的东西,我用Claude Code。如果我需要理解底层发生了什么,或者原型化一个用框架很难实现的东西,我从循环开始。

10、框架在哪里发挥作用

你不需要LangGraph来学习什么是智能体。当重试、检查点和审批门不再是可选的时候,你才需要它。

LangGraph通过将智能体建模为具有显式节点和边的状态机来解决这些问题。更多设置,但你获得了检查点、结构化错误处理、人在回路中的步骤,以及对智能体正在做什么和为什么做的完全可观察性。

CrewAIAutoGen专注于多智能体协调。不是一个拥有许多工具的智能体,而是定义多个具有专业角色的智能体并协调它们如何通信。

50行版本是一个草图。LangGraph是同一张草图变成了一栋有适当承重墙的建筑。

用于生产:使用框架。用于理解实际发生了什么:自己写循环。

11、构建这个教会了我什么

我想了解AI智能体是如何工作的。现在我了解了。

构建这个给了我我缺失的完整心智模型。我可以确切地看到智能体可能在哪里卡住,为什么它选择一个工具而不是另一个,以及什么时候添加更多工具实际上会让事情变得更糟。

没有魔法。模型观察对话历史,决定它是否有足够的信息来回答还是需要工具,然后重复直到完成。其他一切——重试逻辑、人在回路、记忆、多智能体编排——都建立在这个循环之上。

先构建简单版本。然后再决定。


原文链接: Building an AI Agent from Scratch: No Magic, Just a Deterministic Loop

汇智网翻译整理,转载请标明出处