导读

• 目标读者：会一点 Python，希望真正理解智能体如何构成的开发者和 AI 产品人员。
• 核心问题：怎样不用 Agent 框架，只用 Python 和模型 API，亲手跑通对话、工具、执行循环与简单长期记忆？
• 一句话观点：模型提供判断和生成能力；智能体是应用围绕模型建立的运行系统，它还需要状态、循环、工具、记忆、权限和终止条件。
• 读者收获：从零写出一个最小智能体，理解每增加一层能力解决了什么问题，也知道它还缺少什么。

如果你还分不清模型、API、SDK、兼容接口、上下文和记忆，建议先读系列总览：

智能体到底是什么：从模型、API 到工具与记忆

总览篇负责解释整套系统的边界；本文只做一件事：把最小内核写出来并运行起来。

实现路线

这篇文章不从复杂框架开始，而是逐层增加能力：

一次 API 调用
    ↓
可以连续对话
    ↓
模型可以选择工具
    ↓
程序循环执行工具并反馈结果
    ↓
保存并使用长期偏好
    ↓
增加权限、预算、日志和评估
    ↓
理解 Claude Code 为什么更强

每一步都保留上一阶段的代码和局限。这样读者看到的不只是最终结果，还能理解一个智能体为什么需要这些组成部分。

---

一、准备环境

本文使用 Python 和 OpenAI Responses API 构建示例。

为什么用 Responses API，而不是 Chat Completions？

如果你之前用过 client.chat.completions.create(...)，可能会好奇两者有什么区别。

• Chat Completions（/v1/chat/completions）：经典对话接口。每次请求都要自己维护 messages 数组，把完整历史反复发给模型；返回的是 choices[0].message。
• Responses API（/v1/responses）：面向智能体场景设计。请求用 input 和 instructions，返回的是一组类型明确的 Item（文本、工具调用、工具结果等）；可以通过 previous_response_id 让服务端串联上下文，不必每轮重传全部历史；也支持网页搜索、代码执行等内置工具。

OpenAI 推荐新项目优先使用 Responses API，尤其是多轮对话、工具调用和智能体循环。本文后面的代码都基于 client.responses.create(...)。更多对比见 迁移指南。

环境管理用 uv。uv 是 Astral 出品的 Python 包与项目管理工具，用 Rust 编写，比 pip/venv 快很多，也能替代 pyenv、pipx 等常见工具。

创建虚拟环境并安装 SDK：

uv venv
source .venv/bin/activate  // Windows命令略有不同
uv pip install openai -i https://pypi.tuna.tsinghua.edu.cn/simple

-i 指定 PyPI 镜像源；国内常用 清华大学镜像。

通过环境变量提供 API Key；若使用本地或 OpenAI 兼容代理，还需设置 Base URL：

export OPENAI_API_KEY="你的 API Key"
# 可选：本地模型或兼容代理，例如 Ollama、vLLM、One API
export OPENAI_BASE_URL="http://127.0.0.1:8080/v1"

不要把 API Key 写进代码、文章示例、Git 仓库或笔记。OPENAI_BASE_URL 不设置时，SDK 默认连接 OpenAI 官方 API。

本文示例按步骤拆成 5 个可独立运行的文件，放在项目根目录（与虚拟环境同级）：

文件	对应步骤	运行
agent1.py	一次 API 调用	python agent1.py
agent2.py	连续对话	python agent2.py
agent3.py	工具 + 单次任务循环	python agent3.py
agent4.py	多轮对话 + 工具	python agent4.py
agent5.py	长期记忆	python agent5.py

每一步文件都自带完整代码，可直接运行，无需手动拼装。下面各节说明该步解决了什么问题；完整代码集中在每节末尾的一个代码块中，也可直接使用对应 .py 文件。

公共初始化（各文件开头相同）：

import os

from openai import OpenAI

client = OpenAI(
    base_url=os.getenv("OPENAI_BASE_URL"),
)
MODEL = os.getenv("OPENAI_MODEL", "gpt-5.6-sol")

模型名集中在一个常量中，后续可以通过 OPENAI_MODEL 替换，而不需要修改业务代码。

---

二、第一步：完成一次普通模型调用

先写最简单的函数：接收用户输入、向模型发起请求、返回文本。非流式会等整段回答生成完才一次性返回；流式则边生成边打印。

运行：python agent1.py

完整代码（agent1.py）：

"""第一步：一次 API 调用（非流式 + 流式）。"""

import os

from openai import OpenAI

client = OpenAI(
    base_url=os.getenv("OPENAI_BASE_URL"),
)
MODEL = os.getenv("OPENAI_MODEL", "gpt-5.6-sol")


def ask_once(user_input: str) -> str:
    response = client.responses.create(
        model=MODEL,
        input=user_input,
    )
    return response.output_text


def ask_once_stream(user_input: str) -> str:
    stream = client.responses.create(
        model=MODEL,
        input=user_input,
        stream=True,
    )

    chunks: list[str] = []
    for event in stream:
        if event.type == "response.output_text.delta":
            print(event.delta, end="", flush=True)
            chunks.append(event.delta)

    print()
    return "".join(chunks)


if __name__ == "__main__":
    prompt = "请用三句话解释什么是智能体。"
    print("=== 非流式 ===")
    print(ask_once(prompt))
    print("\n=== 流式 ===")
    ask_once_stream(prompt)

它完成了：

1. 接收用户输入。
2. 向模型发起请求。
3. 返回模型生成的文本。

但它还不是智能体。

这个函数没有自己的运行状态，不会连续执行任务，不能访问外部数据，也不能真正采取行动。它只是对模型 API 的一次封装。

此时的能力边界

输入文本 → 模型 → 输出文本

模型可能知道很多通用知识，但它不知道当前电脑上的文件，也不能读取数据库、发送邮件或记住上次运行时用户说过什么。

---

三、第二步：让程序能够连续对话

如果每次调用彼此独立，用户第二次说“详细解释一下”，模型并不知道“它”指什么。

OpenAI 官方 API 可以用 previous_response_id 把上下文存在服务端；但许多本地兼容网关不支持这种续聊。更通用的做法是在程序里维护一份会话历史，每轮把完整历史放进 input 发给模型。历史就是一个 Python 列表，结构和 JSON 数组一致：

[
  {"role": "user", "content": "什么是智能体？"},
  {"role": "assistant", "content": "智能体是……"}
]

下面用 history 在内存里保存对话，并用流式输出。程序退出后记忆消失；若要跨次运行保留，可以把 history 用 json.dump 写到文件。

运行：python agent2.py

完整代码（agent2.py）：

"""第二步：连续对话（内存 history + 流式输出）。"""

import os

from openai import OpenAI

client = OpenAI(
    base_url=os.getenv("OPENAI_BASE_URL"),
)
MODEL = os.getenv("OPENAI_MODEL", "gpt-5.6-sol")


def chat() -> None:
    history: list[dict[str, str]] = []

    while True:
        user_input = input("你：").strip()

        if user_input in {"exit", "quit"}:
            break
        if not user_input:
            continue

        history.append({"role": "user", "content": user_input})

        stream = client.responses.create(
            model=MODEL,
            input=history,
            stream=True,
        )

        print("助手：", end="", flush=True)
        answer_parts: list[str] = []
        for event in stream:
            if event.type == "response.output_text.delta":
                print(event.delta, end="", flush=True)
                answer_parts.append(event.delta)
        print()

        history.append({"role": "assistant", "content": "".join(answer_parts)})


if __name__ == "__main__":
    chat()

为什么它仍然不是完整智能体

我们只是给程序增加了：

• 一个输入循环。
• 多轮对话状态。
• 一个退出条件。

它仍然只能生成文本。即使模型回答“我帮你查询天气”，程序也没有提供任何查询天气的能力。

这里需要区分两个循环：

1. 用户对话循环：等待用户下一次输入。
2. 智能体执行循环：在一次用户任务内部，模型和工具可能来回运行多次。

只有第一个循环，只能得到一个聊天程序。

---

四、第三步：给模型提供工具

假设我们希望助手能够查询某个时区的当前时间。

思路分三层：

1. Python 函数（如 get_current_time）真正执行逻辑。
2. JSON Schema 工具定义（TOOLS）告诉模型工具叫什么、需要什么参数。
3. 白名单路由器（call_tool）只允许执行已注册的工具，不根据模型返回的字符串动态执行任意函数。

工具定义不会自动执行函数，它只是让模型知道：当需要当前时间时，可以返回一个结构化的 function_call。

完整实现与下一步的执行循环合并在 agent3.py 中。

---

五、第四步：加入真正的智能体执行循环

现在需要把模型调用、工具执行和结果反馈连接起来。run_agent() 在一次用户任务内部循环：模型决定是否调用工具 → 程序执行并返回 function_call_output → 模型根据真实结果生成回答。

一次任务内部可能发生：

用户问题
    ↓
模型请求 get_current_time
    ↓
程序验证参数并执行函数
    ↓
程序提交 function_call_output
    ↓
模型根据真实结果生成回答

运行：python agent3.py

完整代码（agent3.py）：

"""第三、四步：工具定义 + 单次任务智能体循环。"""

import json
import os
from datetime import datetime
from typing import Any
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError

from openai import OpenAI

client = OpenAI(
    base_url=os.getenv("OPENAI_BASE_URL"),
)
MODEL = os.getenv("OPENAI_MODEL", "gpt-5.6-sol")
MAX_TURNS = 8

TOOLS = [
    {
        "type": "function",
        "name": "get_current_time",
        "description": "查询指定 IANA 时区的当前时间。",
        "parameters": {
            "type": "object",
            "properties": {
                "timezone": {
                    "type": "string",
                    "description": "IANA 时区，例如 Asia/Singapore。",
                }
            },
            "required": ["timezone"],
            "additionalProperties": False,
        },
        "strict": True,
    }
]


def get_current_time(timezone: str) -> dict[str, str]:
    try:
        now = datetime.now(ZoneInfo(timezone))
    except ZoneInfoNotFoundError:
        return {
            "ok": "false",
            "error": f"未知时区：{timezone}",
        }

    return {
        "ok": "true",
        "timezone": timezone,
        "time": now.isoformat(timespec="seconds"),
    }


def call_tool(name: str, arguments: dict) -> dict:
    if name == "get_current_time":
        return get_current_time(**arguments)

    raise ValueError(f"不允许调用未知工具：{name}")


def run_agent(user_input: str) -> str:
    input_items: list[Any] = [
        {
            "role": "user",
            "content": user_input,
        }
    ]

    for _ in range(MAX_TURNS):
        response = client.responses.create(
            model=MODEL,
            instructions=(
                "你是一个可靠的助手。"
                "只有在确实需要外部能力时才调用工具；"
                "工具失败时应说明失败，不要编造结果。"
            ),
            input=input_items,
            tools=TOOLS,
        )

        input_items.extend(response.output)

        tool_calls = [
            item for item in response.output if item.type == "function_call"
        ]

        if not tool_calls:
            return response.output_text

        for tool_call in tool_calls:
            try:
                arguments = json.loads(tool_call.arguments)
                result = call_tool(tool_call.name, arguments)
            except (json.JSONDecodeError, TypeError, ValueError) as error:
                result = {
                    "ok": "false",
                    "error": str(error),
                }

            input_items.append(
                {
                    "type": "function_call_output",
                    "call_id": tool_call.call_id,
                    "output": json.dumps(result, ensure_ascii=False),
                }
            )

    raise RuntimeError(f"智能体超过最大轮次：{MAX_TURNS}")


if __name__ == "__main__":
    print(run_agent("新加坡现在几点？"))

这已经是一个最小工具型智能体。

它为什么比聊天程序更接近智能体

它具备了几个关键元素：

• 目标：完成用户当前任务。
• 状态：保存当前任务已经发生的消息和工具结果。
• 决策：模型根据上下文选择是否调用工具。
• 行动：应用执行经过允许的工具。
• 观察：工具结果重新进入上下文。
• 循环：模型可以基于新结果继续行动。
• 终止条件：没有工具调用、达到最大轮次或发生不可恢复错误。

模型没有直接执行 Python 函数。模型提出结构化请求，应用决定是否执行并负责返回结果。

---

六、第五步：把多轮对话和工具循环组合起来

前面的 run_agent() 只处理一次用户任务。如果希望在多轮对话中继续使用工具，需要把 Assistant 类在用户轮次之间保存 history。

运行：python agent4.py（交互式，exit 退出）

完整代码（agent4.py）：

"""第五步：多轮对话 + 工具循环。"""

import json
import os
from datetime import datetime
from typing import Any
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError

from openai import OpenAI

client = OpenAI(
    base_url=os.getenv("OPENAI_BASE_URL"),
)
MODEL = os.getenv("OPENAI_MODEL", "gpt-5.6-sol")
MAX_TURNS = 8

TOOLS = [
    {
        "type": "function",
        "name": "get_current_time",
        "description": "查询指定 IANA 时区的当前时间。",
        "parameters": {
            "type": "object",
            "properties": {
                "timezone": {
                    "type": "string",
                    "description": "IANA 时区，例如 Asia/Singapore。",
                }
            },
            "required": ["timezone"],
            "additionalProperties": False,
        },
        "strict": True,
    }
]


def get_current_time(timezone: str) -> dict[str, str]:
    try:
        now = datetime.now(ZoneInfo(timezone))
    except ZoneInfoNotFoundError:
        return {
            "ok": "false",
            "error": f"未知时区：{timezone}",
        }

    return {
        "ok": "true",
        "timezone": timezone,
        "time": now.isoformat(timespec="seconds"),
    }


def call_tool(name: str, arguments: dict) -> dict:
    if name == "get_current_time":
        return get_current_time(**arguments)

    raise ValueError(f"不允许调用未知工具：{name}")


class Assistant:
    def __init__(self) -> None:
        self.history: list[Any] = []

    def ask(self, user_input: str) -> str:
        self.history.append(
            {
                "role": "user",
                "content": user_input,
            }
        )

        for _ in range(MAX_TURNS):
            response = client.responses.create(
                model=MODEL,
                instructions=(
                    "你是一个可靠的个人助手。"
                    "需要实时信息时使用工具，不要编造工具结果。"
                ),
                input=self.history,
                tools=TOOLS,
            )

            self.history.extend(response.output)

            tool_calls = [
                item for item in response.output if item.type == "function_call"
            ]

            if not tool_calls:
                return response.output_text

            for tool_call in tool_calls:
                try:
                    arguments = json.loads(tool_call.arguments)
                    result = call_tool(tool_call.name, arguments)
                except (json.JSONDecodeError, TypeError, ValueError) as error:
                    result = {
                        "ok": "false",
                        "error": str(error),
                    }

                self.history.append(
                    {
                        "type": "function_call_output",
                        "call_id": tool_call.call_id,
                        "output": json.dumps(result, ensure_ascii=False),
                    }
                )

        raise RuntimeError("智能体未能在轮次限制内完成任务")


def chat() -> None:
    assistant = Assistant()

    while True:
        user_input = input("你：").strip()
        if user_input in {"exit", "quit"}:
            break
        if not user_input:
            continue

        print(f"助手：{assistant.ask(user_input)}")


if __name__ == "__main__":
    chat()

仍然存在的问题

history 只保存在当前 Python 进程中：

• 程序退出后历史消失。
• 历史会不断增长。
• 所有旧内容都发送给模型，成本和噪声逐渐增加。
• 对话历史不等于经过筛选的长期记忆。

---

七、第六步：加入长期记忆

如果用户说：

请记住，我更喜欢简洁的技术解释，代码示例使用 Python。

真正的「记住」不能只依赖当前上下文。应用需要：识别 → 持久化 → 检索 → 注入 → 更新或删除。

教学版用 user_memory.json 保存偏好，注册 save_preference 工具，并在 build_instructions() 里把记忆注入每次请求。规则：只有用户明确要求「记住」时才写入。

运行：python agent5.py（交互式，exit 退出）

完整代码（agent5.py）：

"""第六步：长期记忆（JSON 持久化 + 工具写入偏好）。"""

import json
import os
from datetime import datetime
from pathlib import Path
from typing import Any
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError

from openai import OpenAI

client = OpenAI(
    base_url=os.getenv("OPENAI_BASE_URL"),
)
MODEL = os.getenv("OPENAI_MODEL", "gpt-5.6-sol")
MAX_TURNS = 8
MEMORY_FILE = Path("user_memory.json")

TOOLS = [
    {
        "type": "function",
        "name": "get_current_time",
        "description": "查询指定 IANA 时区的当前时间。",
        "parameters": {
            "type": "object",
            "properties": {
                "timezone": {
                    "type": "string",
                    "description": "IANA 时区，例如 Asia/Singapore。",
                }
            },
            "required": ["timezone"],
            "additionalProperties": False,
        },
        "strict": True,
    },
    {
        "type": "function",
        "name": "save_preference",
        "description": "保存用户明确要求记住的偏好。仅在用户说「记住」时使用。",
        "parameters": {
            "type": "object",
            "properties": {
                "key": {
                    "type": "string",
                    "description": "偏好键，例如 response_style。",
                },
                "value": {
                    "type": "string",
                    "description": "偏好内容。",
                },
            },
            "required": ["key", "value"],
            "additionalProperties": False,
        },
        "strict": True,
    },
]


def load_memory() -> dict[str, str]:
    if not MEMORY_FILE.exists():
        return {}

    return json.loads(MEMORY_FILE.read_text(encoding="utf-8"))


def save_preference(key: str, value: str) -> dict[str, str]:
    memory = load_memory()
    memory[key] = value

    MEMORY_FILE.write_text(
        json.dumps(memory, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

    return {
        "ok": "true",
        "saved": key,
    }


def build_instructions() -> str:
    memory = load_memory()
    memory_text = json.dumps(memory, ensure_ascii=False)

    return (
        "你是一个可靠的个人助手。"
        "以下是用户明确保存的偏好："
        f"{memory_text}"
        "偏好只影响表达和默认选择，不能覆盖当前任务的明确要求。"
        "需要实时信息时使用工具，不要编造工具结果。"
        "只有用户明确要求「记住」时才调用 save_preference。"
    )


def get_current_time(timezone: str) -> dict[str, str]:
    try:
        now = datetime.now(ZoneInfo(timezone))
    except ZoneInfoNotFoundError:
        return {
            "ok": "false",
            "error": f"未知时区：{timezone}",
        }

    return {
        "ok": "true",
        "timezone": timezone,
        "time": now.isoformat(timespec="seconds"),
    }


def call_tool(name: str, arguments: dict) -> dict:
    if name == "get_current_time":
        return get_current_time(**arguments)
    if name == "save_preference":
        return save_preference(**arguments)

    raise ValueError(f"不允许调用未知工具：{name}")


class Assistant:
    def __init__(self) -> None:
        self.history: list[Any] = []

    def ask(self, user_input: str) -> str:
        self.history.append(
            {
                "role": "user",
                "content": user_input,
            }
        )

        for _ in range(MAX_TURNS):
            response = client.responses.create(
                model=MODEL,
                instructions=build_instructions(),
                input=self.history,
                tools=TOOLS,
            )

            self.history.extend(response.output)

            tool_calls = [
                item for item in response.output if item.type == "function_call"
            ]

            if not tool_calls:
                return response.output_text

            for tool_call in tool_calls:
                try:
                    arguments = json.loads(tool_call.arguments)
                    result = call_tool(tool_call.name, arguments)
                except (json.JSONDecodeError, TypeError, ValueError) as error:
                    result = {
                        "ok": "false",
                        "error": str(error),
                    }

                self.history.append(
                    {
                        "type": "function_call_output",
                        "call_id": tool_call.call_id,
                        "output": json.dumps(result, ensure_ascii=False),
                    }
                )

        raise RuntimeError("智能体未能在轮次限制内完成任务")


def chat() -> None:
    assistant = Assistant()

    while True:
        user_input = input("你：").strip()
        if user_input in {"exit", "quit"}:
            break
        if not user_input:
            continue

        print(f"助手：{assistant.ask(user_input)}")


if __name__ == "__main__":
    chat()

这才是长期记忆的本质

不是模型在内部永久记住了用户，而是应用完成了：

识别 → 持久化 → 检索 → 注入 → 更新或删除

JSON 文件只适合教学。真实产品通常还需要：

• 按用户隔离数据。
• 区分事实、偏好、任务和敏感信息。
• 设置来源、更新时间和过期策略。
• 对冲突记忆进行处理。
• 只检索当前任务相关的记忆。
• 提供查看、修改、导出和删除能力。

---

八、一个更强的助手还缺什么

到这里，我们已经拥有：

• 连续对话。
• 工具选择。
• 工具执行循环。
• 短期上下文。
• 简单长期偏好。

但从教学 Demo 到可靠产品，还缺少大量工程能力。

1. 权限与确认

查询时间是低风险只读操作；删除文件、发送邮件和付款则不是。

工具系统至少要区分：

• 可以直接执行的只读工具。
• 需要用户确认的写操作。
• 永远不允许模型调用的能力。

模型提出调用请求，不代表程序必须执行。

2. 预算与终止

需要限制：

• 最大循环轮次。
• 单次和累计 Token。
• 工具调用次数。
• 总运行时间。
• 可接受费用。

3. 错误与重试

网络超时可以有限重试；参数错误应反馈给模型修正；权限拒绝不应反复调用；产生副作用的操作需要幂等保护。

4. 上下文管理

对话不断增长后，需要选择、裁剪、总结或压缩旧内容，而不是无限追加历史。

5. 日志与评估

需要记录模型版本、提示词版本、工具调用、耗时、错误和费用，并用真实任务判断智能体是否完成目标。

---

九、再看 Claude Code 为什么强大

现在再看 Claude Code，会更容易理解它的本质。

它并不是因为“像人一样思考”才强大，而是因为围绕模型建立了一套完整运行系统：

• Query Loop 持续调度模型与工具。
• 文件读取、搜索、命令执行等工具扩展了行动范围。
• 权限系统控制哪些操作可以执行。
• 上下文注入让模型看到项目、环境和当前任务状态。
• Memory 和会话持久化跨轮次保留有用信息。
• Compaction 在上下文变长时压缩历史。
• Hooks、Skills、MCP 和子 Agent 提供扩展机制。
• REPL 和 SDK 把中间事件持续交给不同消费者。

如果增加经过约束的浏览器或电脑操作工具，智能体就能进一步完成网页和桌面任务。但能力越强，权限、确认、沙箱和审计越重要。

可以把一个成熟智能体概括为：

模型
+ 状态
+ 控制循环
+ 工具
+ 记忆
+ 权限
+ 终止条件
+ 可观测与评估

Claude Code 的价值，不只是提供了很多工具，而是把这些能力组织进了一套能够持续运行、处理失败并受到约束的系统。

---

十、总结

我们从一次普通 API 调用开始，逐步增加了：

1. 用户对话循环。
2. 多轮上下文。
3. 工具定义与白名单执行。
4. 模型—工具—结果的执行循环。
5. 会话历史。
6. 可持久化的用户偏好。
7. 权限、预算、错误、上下文和评估意识。

最关键的变化不是代码变长了，而是模型从一个文本生成组件，变成了运行系统中的决策组件。

这篇文章跑通的是教学版最小内核。后续文章将分别深入工具系统、上下文与记忆、权限与终止条件，再回到 Claude Code 的 Query Loop，理解这些能力在成熟 Agent 中如何组合。