百行代码手搓AI Agent:从零实现思考-行动-观察循环
1. 项目概述:为什么我们要亲手“搓”一个Agent?
最近AI Agent这个概念火得不行,各种框架和平台层出不穷,功能一个比一个炫酷。但不知道你有没有这种感觉:看别人的Demo很过瘾,轮到自己想搞点定制化功能,或者想理解底层到底是怎么转起来的,就有点无从下手了。文档看了一堆,API调来调去,感觉还是在“用”工具,而不是在“造”工具。这就像开车很熟练,但打开引擎盖却一脸茫然。
所以,就有了这个“百行代码从零手搓Agent”的想法。它的核心目标不是要造一个能比肩AutoGPT或LangChain的工业级框架,而是 通过极简的代码,穿透层层抽象,亲手实现一个Agent最核心的“思考-行动-观察”循环 。这个过程的价值在于,当你亲手用百来行代码让一个程序具备了“根据目标,自主调用工具并持续执行”的能力后,你对LLM(大语言模型)如何与外部世界交互、任务如何被分解、状态如何管理这些核心概念的理解,会变得无比扎实。
这个项目适合谁呢?首先是所有对AI Agent感兴趣,但被复杂框架“劝退”的开发者。其次,是已经会用一些Agent框架,但想深入理解其原理,以期能更好地进行故障排查和深度定制的朋友。最后,它也适合作为教学案例,用最直观的方式展示智能体的运作本质。我们不会依赖任何特定的Agent框架,核心交互对象就是一个LLM的API(比如OpenAI的GPT,或国内的一些大模型API),加上一点清晰的Python逻辑。
2. 核心设计思路:拆解Agent的“大脑”与“手脚”
在动手写代码之前,我们必须先想清楚一个能自主工作的Agent到底由哪些基本部件构成。如果把Agent看作一个智能体,那么它的核心循环可以抽象为一个经典的“感知-思考-行动”模型,在编程上,我们通常称之为 “思考(Think)- 行动(Act)- 观察(Observe)”循环 。
2.1 大脑:LLM作为推理引擎
Agent的“大脑”毫无疑问是大语言模型。但在这里,它的角色不是一个简单的问答机,而是一个 推理引擎和决策中心 。它的核心任务是:
- 理解目标 :解析我们给出的最终目标(例如,“查一下北京今天的天气,然后如果下雨就提醒我带伞”)。
- 规划与分解 :将宏大、模糊的目标分解成一系列可执行的具体子任务(“第一步:获取北京天气信息;第二步:判断是否包含‘雨’;第三步:如果判断为真,则生成提醒”)。
- 选择工具 :决定当前步骤应该使用哪个“工具”(函数)来执行。这需要LLM根据对工具功能描述的理解来做出选择。
- 总结与迭代 :根据工具执行后返回的结果(观察),决定下一步该做什么:是继续下一个子任务,还是任务已完成可以终止。
为了让LLM能更好地履行这些职责,我们需要通过 系统提示词(System Prompt) 来对它进行“角色设定”和“能力约束”。这是整个Agent设计中至关重要的一环。
2.2 手脚:工具(Tools)作为执行单元
Agent的“手脚”就是它能够调用的各种工具。一个工具本质上就是一个 函数 ,它有着明确的功能描述、输入参数和输出结果。例如:
get_weather(city: str) -> str:工具描述为“获取指定城市的当前天气信息”。输入是城市名,输出是天气字符串。send_message(content: str) -> str:工具描述为“发送一条消息”。输入是消息内容,输出是发送状态。
工具的设计原则是“单一职责”和“可靠”。它们应该只做一件明确的事情,并且尽可能保证执行成功(做好异常处理)。Agent的大脑(LLM)不关心工具内部如何实现,它只关心工具的描述、调用格式和返回结果。
2.3 记忆与状态:维持对话的连续性
Agent在执行多步任务时,需要有“记忆”能力,记住之前发生了什么。这通常通过维护一个 对话历史(Message History) 来实现。这个历史记录了用户初始请求、LLM的每次思考、每次工具调用及其结果。每次LLM进行下一轮推理时,都会将完整的历史记录作为上下文喂给它,这样它就能知道任务进展到了哪一步,之前得到了什么信息。
2.4 循环控制器:驱动整个流程的引擎
这是我们将用代码实现的核心逻辑。它负责:
- 初始化大脑(LLM客户端)和工具集。
- 接收用户目标,并将其与系统提示词一起,作为对话历史的开端。
- 进入循环:将当前对话历史发给LLM,请求其下一步决策。
- 解析LLM的响应。响应中应包含明确的指令,如“调用工具A,参数是X”或“最终答案是Y”。
- 如果是指令调用工具,则找到对应工具函数,传入参数执行,并将执行结果以“观察”的形式追加到对话历史。
- 如果是最终答案,则跳出循环,返回结果。
- 为了防止死循环,还需要设置一个最大循环次数。
这个控制器,就是将大脑、手脚和记忆串联起来的“神经系统”。
3. 手搓Agent:分步实现与核心代码解析
接下来,我们抛开所有框架,仅使用 openai 这个官方库(或其他兼容API的库)和Python标准库,用大约百行代码实现上述设计。
3.1 环境准备与基础定义
首先,确保你有一个可用的LLM API密钥。这里我们以OpenAI API为例,但请注意,其设计模式是通用的,可轻松替换为其他提供类似聊天完成接口的模型服务。
pip install openai
然后,我们开始定义最基础的结构。
# agent_core.py
import json
import openai
from typing import Dict, Any, Callable, List
# 1. 定义工具类型:一个工具是一个字典,包含名称、描述、函数本身和参数schema
Tool = Dict[str, Any]
# 2. 初始化OpenAI客户端(请替换your_api_key)
client = openai.OpenAI(api_key="your-api-key-here")
# 3. 定义消息历史类型
MessageHistory = List[Dict[str, str]]
3.2 打造Agent的“工具箱”
我们实现一个简单的工具注册和管理机制。为了能让LLM理解工具,我们需要用自然语言清晰地描述它,并结构化地定义其参数。
class Toolbox:
def __init__(self):
self._tools: Dict[str, Tool] = {}
def register(self, func: Callable, name: str, description: str, args_schema: Dict[str, Any]) -> None:
"""注册一个工具"""
self._tools[name] = {
"function": func,
"description": description,
"args_schema": args_schema
}
def get_tool(self, name: str) -> Tool:
"""根据名称获取工具"""
return self._tools.get(name)
def get_tools_description(self) -> str:
"""生成给LLM看的工具描述文本,这是提示词的关键部分"""
descriptions = []
for name, tool in self._tools.items():
args_desc = ", ".join([f"{k}: {v}" for k, v in tool['args_schema'].items()])
descriptions.append(f"- {name}: {tool['description']} 参数: ({args_desc})")
return "\n".join(descriptions)
# 实例化一个工具箱
toolbox = Toolbox()
现在,我们来定义两个具体的工具函数并注册它们。
# 模拟工具1:获取天气
def get_weather(city: str) -> str:
# 这里本应调用真实天气API,为演示我们模拟返回
weather_data = {
"北京": "晴,气温15-25℃,西北风2级",
"上海": "多云,气温18-28℃,东南风1级",
"广州": "雷阵雨,气温23-32℃,南风3级",
}
return weather_data.get(city, f"未找到{city}的天气信息")
# 模拟工具2:发送消息
def send_message(content: str) -> str:
# 模拟发送动作,例如记录到日志或调用通知接口
print(f"[消息发送] {content}")
return "消息发送成功"
# 注册工具到工具箱
toolbox.register(
func=get_weather,
name="get_weather",
description="获取指定城市的当前天气信息",
args_schema={"city": "字符串,城市名称"}
)
toolbox.register(
func=send_message,
name="send_message",
description="发送一条文本消息",
args_schema={"content": "字符串,消息内容"}
)
注意 :
args_schema这里我们用了简化的字典描述。在更复杂的实现中,可以使用JSON Schema来提供更精确的定义,这能极大提高LLM调用工具时的参数解析成功率。对于百行代码的版本,自然语言描述已足够清晰。
3.3 构建系统提示词:给LLM设定规则
系统提示词是Agent的“宪法”,它定义了LLM在本次对话中的行为准则。一个好的提示词能显著提升Agent的可靠性和准确性。
def create_system_prompt(tools_description: str) -> str:
prompt = f"""
你是一个智能助手,可以调用工具来帮助用户完成任务。
你必须遵循以下规则:
1. 你的目标是根据用户的请求,规划并执行步骤,最终给出答案。
2. 你可以调用以下工具:
{tools_description}
3. 你的每次响应必须是严格的JSON格式,包含两个字段:
- `thought`: 你的思考过程,分析当前情况和下一步计划。
- `action`: 你的行动指令。它有两种可能:
a) 调用工具:格式为 `{{"name": "工具名", "args": {{"参数名": "参数值"}}}}`
b) 最终回答:格式为 `{{"final_answer": "给用户的最终回复内容"}}`
4. 除非用户请求已完全被满足,否则你不应直接给出`final_answer`。你必须通过调用工具来获取信息。
5. 你会收到之前的对话历史,其中包含你之前的`thought`、`action`以及工具执行后的`observation`。请基于完整历史进行下一步决策。
现在,开始处理用户的请求。请始终以JSON格式响应。
"""
return prompt.strip()
这个提示词做了几件关键事:
- 明确角色和目标 :让LLM进入“任务执行者”状态。
- 公开工具清单 :将工具箱的描述嵌入,让LLM知道它能用什么。
- 强制结构化输出 :要求返回JSON,这便于我们程序化地解析。这是实现可靠交互的关键,避免了LLM自由文本输出带来的解析困难。
- 规定决策逻辑 :强调了“非最终答案则必须调用工具”的原则,防止LLM偷懒直接猜测。
3.4 实现核心控制器:运行思考-行动循环
这是最核心的部分,我们将上述所有组件串联起来。
def run_agent(user_request: str, max_steps: int = 10) -> str:
"""
运行Agent的主函数。
:param user_request: 用户的初始请求
:param max_steps: 最大循环步数,防止无限循环
:return: 最终给用户的答案
"""
# 1. 准备系统提示词和初始消息历史
system_prompt = create_system_prompt(toolbox.get_tools_description())
history: MessageHistory = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_request}
]
for step in range(max_steps):
print(f"\n=== 步骤 {step + 1} ===")
# 2. 调用LLM进行“思考”
try:
response = client.chat.completions.create(
model="gpt-3.5-turbo", # 也可使用 gpt-4 获得更好推理能力
messages=history,
temperature=0.1, # 低温度使输出更确定、更遵循指令
response_format={"type": "json_object"} # 要求返回JSON,GPT-3.5-turbo-1106及更新版本支持
)
llm_output = response.choices[0].message.content
print(f"LLM原始输出: {llm_output}")
except Exception as e:
return f"调用LLM API时出错: {e}"
# 3. 解析LLM的响应
try:
result = json.loads(llm_output)
thought = result.get("thought", "")
action = result.get("action", {})
print(f"思考: {thought}")
except json.JSONDecodeError:
return f"无法解析LLM的JSON输出: {llm_output}"
# 4. 判断行动类型并执行
if "final_answer" in action:
# 任务完成,返回最终答案
final_answer = action["final_answer"]
print(f"任务完成!最终答案: {final_answer}")
return final_answer
elif "name" in action and "args" in action:
# 需要调用工具
tool_name = action["name"]
tool_args = action["args"]
print(f"执行动作: 调用工具 `{tool_name}`, 参数 `{tool_args}`")
# 查找工具
tool = toolbox.get_tool(tool_name)
if not tool:
observation = f"错误:未知的工具 '{tool_name}'。"
else:
# 执行工具函数
try:
func = tool["function"]
# 这里假设args是一个字典,直接作为关键字参数传入
observation = func(**tool_args)
except Exception as e:
observation = f"错误:调用工具'{tool_name}'时发生异常 - {e}"
print(f"工具执行结果: {observation}")
# 5. 将本次思考和观察追加到历史,以便下一轮循环
history.append({"role": "assistant", "content": llm_output})
history.append({"role": "user", "content": f"Observation: {observation}"})
else:
# LLM的响应格式不符合预期
return f"LLM返回的动作格式无效: {action}"
# 循环达到最大步数仍未结束
return f"任务未在{max_steps}步内完成,可能陷入循环。最后的历史记录:\n{json.dumps(history, indent=2, ensure_ascii=False)}"
3.5 让我们跑起来:一个完整的执行示例
写一个主函数来测试我们手搓的Agent。
if __name__ == "__main__":
# 测试一个多步任务
user_query = "查一下北京今天的天气,如果下雨就提醒我带伞,如果不下雨就告诉我天气不错。"
print(f"用户请求: {user_query}")
print("-" * 50)
final_result = run_agent(user_query, max_steps=5)
print("\n" + "=" * 50)
print(f"Agent运行结束。结果: {final_result}")
当你运行这段代码时,会在控制台看到类似以下的输出(具体内容因模型随机性略有不同):
用户请求: 查一下北京今天的天气,如果下雨就提醒我带伞,如果不下雨就告诉我天气不错。
--------------------------------------------------
=== 步骤 1 ===
LLM原始输出: {"thought": "用户想了解北京的天气,并根据是否下雨给出不同的建议。我需要先调用get_weather工具获取天气信息。", "action": {"name": "get_weather", "args": {"city": "北京"}}}
思考: 用户想了解北京的天气,并根据是否下雨给出不同的建议。我需要先调用get_weather工具获取天气信息。
执行动作: 调用工具 `get_weather`, 参数 `{'city': '北京'}`
工具执行结果: 晴,气温15-25℃,西北风2级
=== 步骤 2 ===
LLM原始输出: {"thought": "工具返回北京的天气是晴天,没有下雨。根据用户要求,如果不下雨就回复天气不错。现在可以给出最终答案了。", "action": {"final_answer": "北京今天的天气是:晴,气温15-25℃,西北风2级。天气不错,不用带伞。"}}
思考: 工具返回北京的天气是晴天,没有下雨。根据用户要求,如果不下雨就回复天气不错。现在可以给出最终答案了。
任务完成!最终答案: 北京今天的天气是:晴,气温15-25℃,西北风2级。天气不错,不用带伞。
==================================================
Agent运行结束。结果: 北京今天的天气是:晴,气温15-25℃,西北风2级。天气不错,不用带伞。
看,一个能够理解复杂指令、自主调用工具、并根据结果进行判断和回复的Agent,就这样在百行左右的代码里跑起来了!它完成了“感知(用户请求)-思考(规划步骤)-行动(调用天气工具)-再感知(观察结果)-再思考(判断是否下雨)-最终行动(给出答案)”的完整循环。
4. 关键细节、优化与避坑指南
虽然核心流程已经跑通,但要让这个“手搓”的Agent更健壮、更实用,还需要考虑很多细节。这里分享一些从实践中总结的关键点和避坑技巧。
4.1 结构化输出的稳定性保障
我们要求LLM返回JSON,但LLM有时会“不听话”,在JSON外添加额外解释,或者格式错误。除了使用 response_format={"type": "json_object"} 这个强有力的约束(注意模型需支持),还可以增加一个后处理校验和修复层。
import re
def parse_llm_output(llm_text: str) -> Dict[str, Any]:
"""尝试从LLM输出中解析出JSON对象。"""
# 方法1:直接解析
try:
return json.loads(llm_text)
except json.JSONDecodeError:
pass
# 方法2:尝试提取被```json ```包裹的内容
json_match = re.search(r'```json\n(.*?)\n```', llm_text, re.DOTALL)
if json_match:
try:
return json.loads(json_match.group(1))
except:
pass
# 方法3:尝试提取最像JSON的一段(简易版,生产环境需更健壮)
# 这里可以尝试更复杂的启发式方法...
# 如果都失败,返回一个明确的错误结构
return {"error": f"无法解析为JSON: {llm_text[:200]}"}
然后在 run_agent 中,用 parse_llm_output 代替直接的 json.loads ,并对解析错误的情况进行更优雅的处理,比如让Agent重试或给出更明确的错误提示。
4.2 工具调用的参数验证与转换
我们的工具函数期望特定类型的参数(如 city 是字符串),但LLM生成的参数值可能是 "北京" ,也可能是 "城市:北京" 。直接传入可能导致函数调用失败。因此,在调用工具前,应增加一层参数清洗和验证。
def invoke_tool_safely(tool: Tool, llm_args: Dict) -> str:
"""安全地调用工具,处理参数转换和异常。"""
func = tool["function"]
schema = tool["args_schema"] # 这里可以升级为更详细的schema
cleaned_args = {}
# 简单的参数清洗示例:去除空格,转换类型(这里假设都是字符串)
for expected_arg in schema.keys():
raw_value = llm_args.get(expected_arg)
if raw_value is None:
return f"错误:缺少必要参数 '{expected_arg}'。"
# 确保是字符串,并去除首尾空格
cleaned_args[expected_arg] = str(raw_value).strip()
try:
return func(**cleaned_args)
except TypeError as e:
return f"错误:参数不匹配 - {e}"
except Exception as e:
return f"错误:工具执行过程中出错 - {e}"
4.3 管理对话历史的长度与成本
随着循环进行,对话历史会越来越长。这带来两个问题:1) 可能超过模型的上下文窗口限制;2) 每次API调用都携带冗长历史,成本增加。常见的优化策略是 历史摘要(History Summarization) 。在每轮或每几轮之后,可以请求LLM对当前对话历史进行一个简短的摘要,然后用这个摘要替换掉之前的大部分原始历史,只保留最近几轮完整的交互。这能有效控制上下文长度。
def summarize_history(history: MessageHistory, client) -> str:
"""请求LLM生成对话历史的摘要。"""
summary_prompt = [
{"role": "system", "content": "请将以下对话历史浓缩成一个简洁的摘要,保留任务目标、关键决策和当前状态。"},
{"role": "user", "content": json.dumps(history, ensure_ascii=False)}
]
try:
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=summary_prompt,
temperature=0,
max_tokens=200
)
return response.choices[0].message.content
except:
return "摘要生成失败。"
在 run_agent 循环中,可以判断历史长度,当超过某个阈值(如总token数超过2000)时,触发摘要生成,然后用 [系统提示词] + [摘要] + [最近2轮完整交互] 组成新的历史。
4.4 处理复杂任务与循环失控
我们的Agent目前是线性的“思考-行动”循环。对于更复杂的、需要回溯或尝试不同路径的任务(比如“在网上找一家评分高于4.5的意大利餐厅,并预订”),简单的循环可能不够。此时可以引入更高级的概念,如 子目标(Sub-goal) 和 任务队列(Task Queue) 。LLM在思考时,不仅可以决定下一步动作,还可以将一个复杂动作分解成多个子任务放入队列,然后依次执行。这需要更复杂的状态机来管理。
另外,一定要设置 max_steps 并做好监控。有时LLM可能会在两个工具间来回调用陷入死循环,或者在某个判断逻辑上“鬼打墙”。除了设置步数限制,还可以在历史中检测重复的模式,一旦发现就中断并报错。
4.5 选择与适配不同的模型
我们的代码基于OpenAI的ChatCompletion接口。如果你想换成其他模型,比如国内的通义千问、文心一言,或者开源的Llama 3的API服务,主要需要修改两部分:
- 客户端初始化 :替换
openai.OpenAI为对应模型的客户端。 - API调用参数 :调整
client.chat.completions.create中的参数名,确保其符合目标API的规范。大部分提供类似功能的API,其请求和响应格式都是相似的。
核心的Agent循环逻辑、工具管理、提示词设计都是通用的,这体现了我们“手搓”的价值——理解了本质,迁移和适配就变得非常容易。
5. 从原型到实用:扩展思路与进阶方向
这个百行代码的Agent是一个完美的起点和教学工具。基于它,你可以向多个方向扩展,构建真正实用的应用。
5.1 工具生态的扩展
真正的生产力来自于丰富的工具。你可以将任何API、任何函数封装成工具。
- 网络搜索 :封装Serper、Google Search API。
- 文件操作 :读取本地文档(TXT、PDF、Word)、写入文件。
- 代码执行 :在一个安全的沙箱环境中执行Python代码片段(需极其谨慎!)。
- 数据库查询 :连接数据库,将自然语言转换为SQL并执行。
- 软件操作 :通过Selenium控制浏览器,或通过RPA库操作桌面软件。
关键在于为每个工具编写清晰、准确、无歧义的描述,并设计好参数schema。
5.2 记忆与知识库的增强
当前的“记忆”仅是对话历史,是短暂的。对于需要长期记忆或专业知识的任务,可以引入向量数据库(如Chroma、Weaviate)。
- 将相关文档(产品手册、公司制度、个人笔记)切片、编码成向量并存储。
- 当用户提问时,将问题也编码成向量,在向量数据库中检索最相关的文档片段。
- 将这些片段作为“上下文”连同问题一起喂给LLM,让Agent实现“基于知识库的问答”。
这相当于给Agent配备了一个外部长期记忆库。
5.3 多智能体协作
单个Agent能力有限。可以创建多个具有不同专长(如“研究员”、“写手”、“审核员”)的Agent,让它们通过一个“协调者”或简单的消息队列进行协作。例如,一个Agent负责搜索信息,另一个负责撰写报告草稿,第三个负责润色和检查。这需要设计Agent间的通信协议和协作流程。
5.4 可视化与调试界面
在开发复杂Agent时,一个能可视化展示“思考链”的界面无比重要。你可以记录下每一轮的 thought , action , observation ,并将其输出到控制台、日志文件,或者用一个简单的Web界面展示出来。这能帮你快速定位Agent是在哪一步“想歪了”,是工具描述不清?还是提示词有歧义?抑或是LLM本身推理的局限性?
亲手实现这个百行Agent的最大收获,绝不是这百行代码本身,而是在实现过程中,你被迫去思考并厘清的每一个细节:LLM如何被引导、工具如何被描述、状态如何流转、错误如何被处理。这些认知,是无论阅读多少篇框架文档都无法完全获得的。当你再去看那些成熟的Agent框架时,你会一眼看穿它们华丽外衣下的骨架,无非是更健壮、更高效、功能更丰富的“思考-行动-观察”循环管理器。这时,你就不再只是一个框架的使用者,而是一个真正的理解者和创造者。
更多推荐

所有评论(0)