100行Python代码实现AI智能体核心循环:从ReAct模式到工具调用实战
1. 项目概述:为什么我们要亲手写一个AI智能体?
最近“AI智能体”这个概念火得不行,感觉不提Agent都不好意思说自己在搞AI。但说实话,很多文章和教程要么讲得太玄乎,堆砌一堆“自主性”、“工具使用”、“记忆”的术语;要么就是直接甩出一个庞大的框架,像LangChain、AutoGPT,代码动辄几千行,新手一看就懵,根本不知道核心机制到底是怎么转起来的。
这就像学开车,教练一上来就给你讲发动机缸内直喷、涡轮增压原理,然后直接把你塞进一辆F1赛车,告诉你“开吧”。结果大概率是连点火都不会。理解Agent,最好的方式就是自己动手,用最少的代码,实现它的核心引擎——那个驱动一切思考与行动的循环。
所以,我决定写这个系列,目标就一个:用100行左右的Python代码,带你从零构建一个具备完整“感知-思考-行动”循环的简易AI智能体。我们不依赖任何重型框架,就靠最基础的requests调用大模型API,加上清晰的逻辑,把Agent最本质的运作原理扒开给你看。当你亲手实现一遍这个循环,再去看那些复杂框架,就会有一种“哦,原来你就是在这些基础模块上搭积木”的豁然开朗感。
这个项目适合谁?如果你对AI应用开发感兴趣,听说过Agent但感觉云里雾里;或者你是个行动派,喜欢通过动手来理解概念;亦或是你厌倦了“调包”,想深入一层看看机制,那么这篇内容就是为你准备的。我们将从一张白纸开始,最终得到一个能根据目标自主调用工具(比如搜索天气)、并持续运行的智能体原型。
2. 智能体核心循环拆解:它到底在想什么?
在写代码之前,我们必须先搞清楚要构建什么。一个AI智能体,区别于简单的一次性问答模型,其核心在于持续的、目标导向的循环。这个循环通常被称为ReAct (Reason + Act)模式,或者是更经典的感知-思考-行动循环。
2.1 智能体的“状态机”模型
你可以把一个智能体想象成一个拥有内部状态的小机器人。它的核心是一个永不停止的循环(除非我们让它停止),每次循环都处理三件事:
- 感知:获取当前的环境信息或用户输入。这是我们给智能体的“刺激”。
- 思考:基于当前的目标、记忆(历史记录)和感知到的信息,进行推理,决定下一步该做什么。这是智能体的“大脑”。
- 行动:执行思考后决定的操作。这个操作可以是调用一个工具(如计算器、搜索引擎),也可以是生成一段最终答复给用户。
执行完行动后,行动的结果会作为新的“感知”输入,进入下一轮循环。如此周而复始,直到达成目标或满足停止条件。
2.2 循环中的关键组件
为了实现这个循环,我们需要在代码中定义几个核心组件:
- 目标:智能体需要完成的任务,比如“查询北京今天的天气并判断是否适合洗车”。
- 记忆:智能体需要记住之前的对话、自己的行动和行动结果。这是它进行连贯思考的基础。最简单的记忆就是保存整个对话历史。
- 工具集:智能体可以调用的外部函数。比如一个
get_weather函数,输入城市名,返回天气信息。工具扩展了智能体的能力边界。 - 推理引擎:这是最核心的部分,通常由一个大语言模型担任。它的职责是:根据当前的目标、记忆和可用的工具列表,分析现状,然后输出一个结构化的决策。这个决策通常包含两部分:“思考”过程和“行动”指令。
这个决策过程,就是让LLM按照我们设定的格式(比如JSON)来回答。例如:
思考:用户想了解北京天气并决定是否洗车。我需要先获取北京的天气信息。行动:调用
get_weather工具,参数为{"city": "北京"}。
2.3 与大模型的一次性问答有何不同?
你可能想问,这和直接问ChatGPT“北京天气怎么样,能洗车吗?”有什么区别?区别巨大!
- 被动应答 vs 主动规划:一次性问答是“你问,我答”。智能体是“你给我目标,我自行拆解步骤,调用工具,一步步完成”。后者需要自主规划能力。
- 静态 vs 动态:一次性问答上下文有限。智能体在循环中,上一次工具调用的结果会自动成为下一次推理的上下文,形成动态的工作流。
- 能力边界:纯LLM的知识可能过时,也无法执行具体操作(如计算、查询实时数据)。智能体通过工具调用,弥补了LLM的这部分短板。
理解了这些,我们的代码蓝图就清晰了:构建一个循环,在每次迭代中,将目标、记忆、工具描述拼装成提示词,送给LLM;解析LLM的回复,提取出“行动”指令;执行对应的工具;将工具结果和本次思考过程存入记忆,进入下一轮。
3. 100行代码实现核心循环
理论说再多不如一行代码。我们开始动手。确保你有一个可用的OpenAI API Key(或其他兼容OpenAI API格式的LLM服务密钥)。
3.1 环境准备与基础架构
首先,安装必要的库,我们只需要openai(或兼容库)和requests。
pip install openai requests然后,我们创建主程序文件simple_agent.py,并搭建基础结构。
import json import openai import requests # 1. 配置LLM客户端 (这里以OpenAI为例) openai.api_key = "你的API_KEY" MODEL = "gpt-3.5-turbo" # 或 "gpt-4" # 2. 定义工具集 def get_weather(city: str) -> str: """模拟获取天气的工具。实际应用中应接入真实API。""" # 这里为了演示,返回模拟数据。真实情况可以调用和风天气、OpenWeatherMap等API。 weather_data = { "北京": "晴,气温25度,湿度30%,北风2级。", "上海": "多云,气温28度,湿度65%,东南风3级。", "广州": "雷阵雨,气温30度,湿度85%,南风4级。" } return weather_data.get(city, f"未找到{city}的天气信息。") # 3. 工具描述,用于告诉LLM有什么工具可用 TOOLS = [ { "name": "get_weather", "description": "根据城市名称查询该城市的实时天气情况。", "parameters": { "type": "object", "properties": { "city": {"type": "string", "description": "城市名称,例如:北京、上海"} }, "required": ["city"] } } ] # 将工具函数映射到名字,方便调用 TOOL_FUNCTIONS = { "get_weather": get_weather }注意:这里的
get_weather函数是一个模拟工具。在生产环境中,你需要将其替换为真正的API调用,并做好错误处理。工具描述TOOLS的格式遵循了OpenAI的Function Calling规范,这能让LLM更好地理解如何调用工具。
3.2 构建智能体循环引擎
接下来是核心的Agent类,它封装了记忆、循环逻辑和与LLM的交互。
class SimpleAgent: def __init__(self, goal: str): self.goal = goal # 智能体的终极目标 self.memory = [] # 对话与行动历史 self.max_loops = 10 # 防止无限循环 def run(self): """启动智能体,运行核心循环。""" print(f"🎯 智能体目标:{self.goal}") print("-" * 40) for step in range(self.max_loops): print(f"\n🔄 循环第 {step + 1} 步:") # 1. 规划:让LLM根据目标、记忆和工具进行思考 llm_response = self._plan() # 2. 解析LLM的回复,判断是直接回答还是调用工具 action, action_input, thought = self._parse_response(llm_response) # 将本次“思考”存入记忆 self.memory.append({"role": "assistant", "content": thought}) if action == "final_answer": # 如果是最终答案,则输出并结束循环 print(f"💡 思考:{thought}") print(f"✅ 最终答案:{action_input}") print(f"\n🎉 目标达成!共用了 {step + 1} 步。") break elif action in TOOL_FUNCTIONS: # 3. 执行:调用工具 print(f"💡 思考:{thought}") print(f"🛠️ 行动:调用工具【{action}】,参数:{action_input}") tool_result = TOOL_FUNCTIONS[action](**action_input) print(f"📊 观察:工具返回结果 -> {tool_result}") # 将工具执行结果存入记忆,作为下一轮循环的输入 self.memory.append({"role": "user", "content": f"工具{action}返回的结果是:{tool_result}"}) else: print(f"⚠️ LLM返回了未知指令或格式错误:{llm_response}") break else: print(f"⛔ 已达到最大循环次数{self.max_loops},未完成目标。") def _plan(self): """构造提示词,调用LLM进行规划。""" # 构造系统提示,定义智能体的角色、目标和可用工具 system_prompt = f"""你是一个有帮助的AI智能体。你的目标是:{self.goal} 你可以使用以下工具: {json.dumps(TOOLS, indent=2, ensure_ascii=False)} 请严格按以下格式回应: 1. 首先进行“思考”,分析当前情况和下一步计划。 2. 然后决定是“调用工具”还是给出“最终答案”。 3. 如果是调用工具,请以JSON格式输出,包含“action”(工具名)和“action_input”(工具参数)。 4. 如果是最终答案,请以JSON格式输出,包含“action”: “final_answer” 和 “action_input”(你的答案文本)。 示例1(调用工具): 思考:用户想知道北京天气。我需要调用get_weather工具。 {{"action": "get_weather", "action_input": {{"city": "北京"}}}} 示例2(最终答案): 思考:我已经获得了北京的天气信息,是多云,可以洗车。 {{"action": "final_answer", "action_input": "北京今天多云,气温适宜,适合洗车。"}} 现在,请基于以下对话历史进行决策:""" messages = [{"role": "system", "content": system_prompt}] messages.extend(self.memory) # 加入历史记忆 try: response = openai.ChatCompletion.create( model=MODEL, messages=messages, temperature=0.1, # 低温度使输出更稳定、更遵循格式 max_tokens=500 ) return response.choices[0].message.content except Exception as e: return f"Error calling LLM: {e}" def _parse_response(self, response: str): """解析LLM的回复,提取思考、行动和输入。""" # 首先分离出“思考”部分和JSON部分 lines = response.strip().split('\n') thought = "" json_str = "" for line in lines: if line.startswith('思考:') or line.startswith('思考:'): thought = line[3:].strip() elif line.strip().startswith('{'): json_str = line # 有时JSON可能跨多行,这里简单处理。更健壮的做法是用正则匹配整个JSON块。 try: # 尝试直接解析这一行 data = json.loads(line) except json.JSONDecodeError: # 如果失败,尝试合并后续行直到找到完整的JSON pass # 如果json_str为空,尝试在整个response中查找JSON if not json_str: import re json_match = re.search(r'\{.*\}', response, re.DOTALL) if json_match: json_str = json_match.group() # 解析JSON try: data = json.loads(json_str) action = data.get("action", "") action_input = data.get("action_input", {}) return action, action_input, thought except (json.JSONDecodeError, AttributeError): # 如果解析失败,尝试另一种常见格式:LLM可能直接说了最终答案 if "final" in response.lower() or "答案" in response: return "final_answer", response, "LLM直接给出了最终回复。" return "error", {}, f"无法解析LLM回复:{response}"3.3 运行你的第一个智能体
现在,让我们创建一个智能体实例并运行它。
if __name__ == "__main__": # 定义一个目标 agent_goal = "查询北京今天的天气,并告诉我是否适合洗车。" # 创建智能体 agent = SimpleAgent(agent_goal) # 运行! agent.run()保存并运行这个脚本。你应该能看到类似下面的输出,清晰地展示了智能体“思考-行动-观察”的循环过程:
🎯 智能体目标:查询北京今天的天气,并告诉我是否适合洗车。 ---------------------------------------- 🔄 循环第 1 步: 💡 思考:用户想了解北京天气并决定是否洗车。我需要先获取北京的天气信息。 🛠️ 行动:调用工具【get_weather】,参数:{'city': '北京'} 📊 观察:工具返回结果 -> 晴,气温25度,湿度30%,北风2级。 🔄 循环第 2 步: 💡 思考:我已经获得了北京的天气信息。今天是晴天,气温25度,湿度较低,风力不大。这种天气非常适合洗车,因为阳光充足,水分蒸发快,且没有雨水和沙尘的担忧。 ✅ 最终答案:北京今天天气晴朗,气温25度,湿度低,风力小。这种天气条件非常适合洗车,可以快速晾干且不易沾染灰尘。 🎉 目标达成!共用了 2 步。看,一个简易但功能完整的AI智能体就诞生了!它在第一轮循环中“思考”出需要调用天气工具,执行后获得结果;在第二轮循环中,它基于新的记忆(天气结果)进行“思考”,判断出适合洗车,并给出了“最终答案”,循环结束。
4. 核心机制深度解析与优化
代码跑通了,但里面有很多细节值得深究。理解这些细节,是你从“能用”到“精通”的关键。
4.1 提示词工程:如何让LLM乖乖听话?
智能体的“思考”质量,极大程度上取决于我们给LLM的提示词。上面的system_prompt是一个经典的结构:
- 角色与目标定义:明确告诉LLM“你是谁”、“你要干什么”。
- 工具描述:以结构化的方式(JSON)清晰列出工具的名称、描述和参数。这利用了LLM对结构化数据的理解能力。
- 输出格式约束:这是最关键的一步。我们强制要求LLM以“思考:...”和JSON块的形式回复。通过提供清晰的示例,可以极大地提高LLM遵循格式的概率。
temperature参数设为较低值(如0.1),也是为了减少输出的随机性,让它的行为更可控。 - 上下文注入:将
self.memory(历史记录)作为对话历史传入,让LLM拥有“记忆”能力。
实操心得:让LLM输出严格格式的JSON有时会失败,它可能在JSON外加引号或添加无关文本。因此,
_parse_response函数中的解析逻辑需要有一定的容错性,比如使用正则表达式re.search(r'\{.*\}', response, re.DOTALL)来提取可能的JSON对象。更高级的做法是使用LLM的“函数调用”功能,但这超出了我们100行代码的极简范畴。
4.2 记忆管理:上下文长度的博弈
我们的memory只是一个简单的列表,每次循环都全部发送给LLM。这存在两个问题:
- 上下文长度限制:所有主流LLM都有上下文窗口限制(如4K、8K、128K tokens)。如果对话历史很长,很快就会超出限制。
- 效率与成本:发送大量历史token会增加API调用成本和延迟。
解决方案:
- 摘要式记忆:不要存储完整的对话历史,而是定期(或每次行动后)让LLM对之前的历史进行摘要,只保留摘要和最近几次交互。这能显著压缩上下文。
- 向量记忆:将历史信息转换为向量存入数据库(如ChromaDB)。每次需要回忆时,根据当前问题检索最相关的历史片段。这适合处理超长记忆,但实现更复杂。
- 滑动窗口:只保留最近N轮对话,丢弃老的。这是最简单粗暴但也最常用的方法,适合短期任务。
在我们的简易版中,由于循环步数少(max_loops=10),直接使用完整记忆是可行的。但在复杂任务中,你必须考虑记忆管理策略。
4.3 工具调用:安全性与错误处理
我们的TOOL_FUNCTIONS映射直接执行了函数。在真实场景中,这非常危险,尤其是当工具涉及文件操作、系统命令或网络请求时。
必须加入的安全与健壮性措施:
- 参数验证与清洗:在工具函数内部,务必检查输入参数。例如,
get_weather(city)应该检查city是否为字符串,并可能过滤掉危险字符。 - 权限隔离:为智能体创建一个具有最小必要权限的执行环境。绝对不要让它以高级权限运行。
- 异常捕获:工具执行可能会失败(网络错误、API限流等)。必须在
try...except块中调用工具,并将友好的错误信息返回给智能体,让它能据此调整策略。 - 工具结果格式化:工具返回的结果应该简洁、信息丰富。冗长或混乱的结果会干扰LLM的下一次推理。
# 增强版的工具调用示例 def safe_tool_call(tool_name, tool_args): if tool_name not in TOOL_FUNCTIONS: return f"错误:未知工具 '{tool_name}'。" tool_func = TOOL_FUNCTIONS[tool_name] # 1. 参数验证(简单示例) if tool_name == "get_weather": if not isinstance(tool_args.get("city"), str): return "错误:参数'city'必须是字符串。" # 可以加入城市名白名单等 # 2. 执行并捕获异常 try: result = tool_func(**tool_args) return str(result) # 确保返回字符串 except Exception as e: return f"调用工具'{tool_name}'时发生错误:{e}"5. 常见问题与扩展思路
在实际编写和运行这类智能体时,你肯定会遇到一些典型问题。
5.1 智能体陷入死循环或无效行动
- 现象:智能体反复调用同一个工具,或者在不同工具间来回切换,无法得出最终答案。
- 原因:
- 目标不明确:LLM不理解什么时候算“任务完成”。需要在提示词中明确停止条件,例如“当你拥有足够信息给出最终建议时,请使用
final_answer”。 - 工具结果模糊:工具返回的信息不足以让LLM做出决策。需要优化工具输出,使其更结构化、更具信息量。
- 思维链断裂:LLM的“思考”部分可能逻辑混乱。可以尝试在提示词中要求更详细的推理步骤,或者使用更强大的模型(如GPT-4)。
- 目标不明确:LLM不理解什么时候算“任务完成”。需要在提示词中明确停止条件,例如“当你拥有足够信息给出最终建议时,请使用
- 解决:
- 设置
max_loops硬性限制,防止无限消耗资源。 - 在
_plan方法的提示词中加入“反思”要求,例如:“请检查当前是否已达成目标,如果已达成,请给出最终答案。” - 在记忆中加入“强制停止”信号。例如,如果连续三步行动没有产生新的有效信息,可以人工注入一条消息:“系统提示:你似乎陷入了循环。请重新评估你的计划。”
- 设置
5.2 如何增加更多工具?
扩展性是我们这个架构的优点。增加新工具只需两步:
- 定义工具函数:编写具体的Python函数。
- 注册到工具集:将工具的描述字典加入
TOOLS列表,并将函数名映射加入TOOL_FUNCTIONS字典。
例如,增加一个计算器工具:
def calculator(expression: str) -> str: """计算数学表达式。注意:使用eval有安全风险,此处仅作演示。""" try: # 警告:在生产环境中,直接eval用户输入是极度危险的! # 应使用安全的表达式解析库,如 `asteval` result = eval(expression) return f"{expression} = {result}" except Exception as e: return f"计算错误:{e}" # 更新TOOLS和TOOL_FUNCTIONS TOOLS.append({ "name": "calculator", "description": "计算一个数学表达式的结果,例如:'3 + 5 * 2'。", "parameters": { "type": "object", "properties": { "expression": {"type": "string", "description": "数学表达式字符串"} }, "required": ["expression"] } }) TOOL_FUNCTIONS["calculator"] = calculator现在,你的智能体就具备了计算能力。你可以给它一个目标:“计算(15 + 7) * 3的值,然后告诉我这个值除以2是多少。” 它会自主规划,先调用一次计算器,再用结果调用第二次。
5.3 从“玩具”到“可用”的进阶方向
我们这个100行的智能体是一个完美的教学原型和起点。基于它,你可以向多个方向深化:
- 集成真实工具:将
get_weather替换为真正的天气API,增加搜索引擎、数据库查询、邮件发送等工具。 - 采用专业框架:理解核心循环后,可以学习LangChain的
Agent、Tool、Memory模块,它们提供了工业级的实现,处理了流式输出、复杂记忆、工具路由等大量细节。 - 实现多智能体协作:创建多个
SimpleAgent实例,让它们分别扮演不同角色(如规划者、执行者、评审者),并通过共享内存或消息队列进行通信,解决更复杂的问题。 - 加入验证与反思:在每次行动后,不是直接进入下一轮,而是增加一个“验证”步骤,检查行动结果是否合理,或让LLM对本次行动进行自我批评和反思,从而提升决策质量。
通过这100行代码,我们亲手点亮了AI智能体的“引擎”。它不再是一个黑盒概念,而是一个由清晰循环、明确状态和可控工具组成的可理解、可调试、可扩展的程序。这个简单的循环,是构建一切复杂智能体应用的基石。当你下次看到那些功能炫酷的Agent项目时,希望你能会心一笑:它的核心,或许就是从这样一个循环开始的。
