手写 AI Agent 1
本文是《手写 AI Agent》系列教程的第 1 篇,建议按顺序阅读。
一、概述
Agent(智能体)是当下 AI 领域最热门的概念之一。但很多人第一次接触 Agent 时,会选择直接调用 LangChain、AutoGPT 等框架,调通 demo 之后,面对报错却无从下手。
问题出在哪里?框架把底层逻辑封装得太好了。你不知道工具调用的顺序是怎么决定的,也不知道报错信息为什么被吞掉。出了问题,只能去翻框架源码。
所以,手写 Agent 的意义不是”造轮子”,而是理解轮子怎么转。
本文是《手写 AI Agent》系列的第一篇。读完本文,你会得到一个最小可用的 Agent 类,约 60 行代码,可以直接运行。
二、核心概念
Agent 的本质,可以用一个公式概括:
Agent = LLM + 循环 + 消息历史
也就是说,一个最小可用的 Agent 只需要四样东西:
封装状态和行为。状态是消息历史,行为是对话。
持续与 LLM 交互。用户说完一句,Agent 处理完,再等待下一句。
记住对话上下文。没有它,Agent 每次对话都是重新开始。
连接 LLM。目前主流提供商(OpenAI、Anthropic、DeepSeek)都支持同一套消息格式。
下面是一段示例,展示消息格式:
messages = [
{"role": "system", "content": "你是一个有用的助手"},
{"role": "user", "content": "你好"},
{"role": "assistant", "content": "你好!有什么可以帮你的?"},
]
三个角色:
system
:系统提示,定义 Agent 的行为和身份 user
:用户输入 assistant
:AI 的回复
消息历史是 Agent 的”短期记忆”。每次对话都追加进去,LLM 才能看到上下文。
三、代码实现
下面是完整的实现代码,保存为 agent.py 即可运行:
"""
最小可用 Agent —— 让 AI 说第一句话
从零手写 AI Agent 课程 · 第 1 章
"""
import os
from typing import Optional
from openai import OpenAI
class Agent:
"""最小可用的 AI Agent"""
def__init__(
self,
system_prompt: str = "你是一个有用的 AI 助手。",
model: str = "gpt-4o-mini",
api_key: Optional[str] = None,
):
# 初始化 LLM 客户端
self.client = OpenAI(api_key=api_key or os.environ.get("OPENAI_API_KEY"))
self.model = model
# 消息历史(初始包含 system prompt)
self.messages = [
{"role": "system", "content": system_prompt}
]
defchat(self, user_input: str) -> str:
"""与 Agent 对话一次"""
# 1. 添加用户消息到历史
self.messages.append({"role": "user", "content": user_input})
# 2. 调用 LLM
response = self.client.chat.completions.create(
model=self.model,
messages=self.messages,
)
# 3. 获取回复文本
reply = response.choices[0].message.content
# 4. 添加助手回复到历史
self.messages.append({"role": "assistant", "content": reply})
return reply
defreset(self):
"""重置对话历史,保留 system prompt"""
self.messages = [self.messages[0]]
defchat_stream(self, user_input: str) -> str:
"""流式对话,逐字输出"""
self.messages.append({"role": "user", "content": user_input})
response = self.client.chat.completions.create(
model=self.model,
messages=self.messages,
stream=True,
)
full_reply = ""
for chunk in response:
if chunk.choices[0].delta.content:
text = chunk.choices[0].delta.content
print(text, end="", flush=True)
full_reply += text
print()
self.messages.append({"role": "assistant", "content": full_reply})
return full_reply
if __name__ == "__main__":
agent = Agent(
system_prompt="你是一个简洁的 AI 助手。回答要简短、准确。",
model="gpt-4o-mini",
)
print("Agent 已启动。输入 'quit' 退出\n")
whileTrue:
user_input = input("你: ").strip()
if user_input.lower() in ("quit", "exit", "q"):
break
reply = agent.chat(user_input)
print(f"\nAgent: {reply}\n")
代码的逻辑很简单,可以拆成三步理解:
创建 Agent 类时,传入 system prompt 和模型名称。构造函数里做两件事:创建 OpenAI 客户端,初始化消息列表(第一条是 system 消息)。
chat() 方法接收用户输入,追加到消息历史,然后调用 LLM。拿到回复后,再把回复也追加到消息历史。这样下一轮对话时,LLM 就能看到完整的上下文。 第三步,流式输出。chat_stream() 与 chat() 的区别在于 stream=True。LLM 不会等全部内容生成完再返回,而是每生成一个字就传回一个字。用户体验上,就像是有人在打字。四、注意事项
运行代码之前,有几点需要注意。
代码会从环境变量读取 OPENAI_API_KEY。运行前确保已设置:
export OPENAI_API_KEY="sk-..."
如果没有 OpenAI 的 key,也可以用 Ollama 本地运行。把客户端初始化改成:
client = OpenAI(
base_url="http://localhost:11434/v1",
api_key="ollama"
)
需要安装 openai 包:
pip install openai
如果频繁调用 API,可能会触发 RateLimitError。等几分钟再试,或者检查你的 API 套餐额度。
如果你每次运行都重新创建 Agent() 实例,历史会丢失。这是正常行为。想要保留对话,需要在同一个实例上连续调用 chat()。
五、总结
本文内容可以总结为以下几点:
-
Agent = LLM + 循环 + 消息历史 Agent
类封装了状态(消息历史)和行为(对话) -
消息格式是统一的,所有 LLM 提供商都兼容 -
流式输出提升用户体验,实现成本很低
这个实现已经能跑通。但它的局限也很明显:只能对话,不能读写文件、不能执行命令。下一篇文章会加入工具调用,让它能做事。
六、参考链接
-
OpenAI API 文档 -
Python 官方文档 -
本系列源码:关注后回复 “Agent1” 获取
(完)
本文属于《手写 AI Agent》系列教程的第 1 篇。
如果你对本文的代码感兴趣,可以在公众号回复 “Agent1″,获取完整源码和课后练习。
下一篇文章会讲工具调用——给 Agent 装上手脚。
夜雨聆风