1. 项目概述:为什么我们要“徒手”造一个Agent?
最近AI Agent的概念火得不行,各种框架层出不穷,LangChain、LangGraph、AutoGen……功能强大,生态繁荣。但对于很多刚入门的开发者,或者只是想快速验证一个简单想法的朋友来说,这些框架有时显得过于“重型”了。依赖繁多,概念复杂,一个简单的“思考-行动”循环,可能被层层抽象包裹,让人看不清本质。
我这个项目,就是想做一次“减法”。只用大约50行Python代码,不引入任何外部Agent框架,实现一个具备ReAct(Reasoning and Acting)思维链能力的最小化智能体。这就像学编程时,我们总会自己写一遍“Hello World”,而不是直接调用一个复杂的库。通过亲手搭建这个最简模型,你能彻底理解Agent运作的核心机制:它如何理解问题、规划步骤、调用工具并从中学习。这对于后续无论是选用成熟框架进行二次开发,还是设计更复杂的多智能体系统,都至关重要。你会发现,剥开华丽的外衣,Agent的内核其实清晰而优雅。
2. 核心设计:拆解ReAct与最小化架构
2.1 ReAct范式:智能体的“思考-行动”循环
ReAct,即“推理(Reasoning)+ 行动(Acting)”,是让大语言模型(LLM)具备工具使用和与环境交互能力的经典范式。它的工作流是一个循环:
- 思考(Think):LLM分析当前状况(用户问题、已有观察、可用工具),规划下一步该做什么。输出通常以“Thought: ...”开头。
- 行动(Act):根据思考的结果,LLM决定调用哪个工具,并生成符合工具要求的调用指令。输出以“Action: ...”开头,后跟工具名和输入参数。
- 观察(Observe):执行工具调用,获取结果(可能是成功的数据,也可能是错误信息)。这个结果作为“Observation: ...”反馈给LLM。
- 循环:LLM接收到观察结果后,进入下一轮“思考”,判断问题是否已解决,若未解决则继续“行动”,直至得出最终答案。
这个循环的精妙之处在于,它将LLM的推理过程“外化”了,使得模型的思考路径变得可追踪、可调试。我们的50行代码,就是对这个循环最直接的实现。
2.2 最小化架构设计
为了极致精简,我们的架构只包含三个核心部分:
- LLM客户端:负责与一个大语言模型API(如OpenAI GPT、DeepSeek、通义千问等)通信。我们将其封装为一个简单的函数,接收提示词(Prompt),返回模型的文本回复。
- 工具集(Tools):一组Python函数,每个函数代表Agent可以执行的一个具体动作。例如,一个计算器函数、一个网络搜索函数(模拟)、一个查询数据库的函数等。每个工具都需要有清晰的名称、描述和参数定义,以便LLM理解何时以及如何调用它。
- ReAct循环引擎:这是最核心的部分,一个
while循环,不断执行“思考->行动->观察”的流程,并解析LLM的输出,直到LLM说出“Final Answer:”。
我们故意不设计复杂的记忆管理、不引入工作流编排、也不做工具的动态注册发现。所有工具在循环开始前就静态定义好。这样,代码的每一行都在为最核心的循环服务,没有任何冗余。
3. 代码实现:逐行解析50行智能体
下面,我将分模块展示并详解这50行左右的核心代码。我们假设使用OpenAI的ChatCompletion API,并准备了两个简单的工具:一个计算器和一个能返回当前时间的模拟工具。
# 导入必要的库 import openai import json import datetime # 1. 配置LLM客户端 client = openai.OpenAI(api_key='your-api-key-here') # 请替换为你的API Key model_name = "gpt-3.5-turbo" # 也可以使用 gpt-4 或其他兼容API的模型 def call_llm(prompt): """调用LLM的简单封装函数""" response = client.chat.completions.create( model=model_name, messages=[{"role": "user", "content": prompt}], temperature=0, # 温度设为0,保证输出的确定性,便于解析 stream=False ) return response.choices[0].message.content.strip()注意:这里使用
temperature=0是为了让Agent的行为在调试阶段更稳定、可复现。在实际复杂任务中,可以适当调高以增加创造性,但会降低工具调用的准确性。
# 2. 定义工具集 def calculator(expression): """计算一个数学表达式。例如:calculator('3 + 5 * 2')""" try: # 警告:使用eval有安全风险,此处仅用于演示。生产环境应使用更安全的表达式解析库(如ast.literal_eval)或自定义解析器。 result = eval(expression) return str(result) except Exception as e: return f"计算错误:{e}" def get_current_time(query): """返回当前的日期和时间。输入参数通常被忽略。""" now = datetime.datetime.now() return now.strftime("%Y-%m-%d %H:%M:%S") # 工具字典,将工具名映射到函数对象和其描述 TOOLS = { "calculator": { "function": calculator, "description": "用于计算数学表达式。输入应为一个可被Python eval函数理解的字符串,如 '3 + 5 * 2'。" }, "get_current_time": { "function": get_current_time, "description": "获取当前的日期和时间。输入参数通常可以为空或任意文本。" } }实操心得:工具的描述(
description)至关重要!LLM完全依赖这段文本来理解工具的用途和输入格式。描述要尽可能精确、无歧义。例如,明确说明输入是“数学表达式字符串”,而不是模糊的“计算问题”。
# 3. 构建系统提示词(System Prompt) def build_system_prompt(): """构建指导LLM行为的系统提示词""" tool_descriptions = "\n".join([f"- {name}: {info['description']}" for name, info in TOOLS.items()]) prompt = f""" 你是一个智能助手,可以通过调用工具来解决问题。请遵循严格的ReAct格式进行回应: 格式说明: Thought: 你需要在这里思考当前情况,分析问题,并决定下一步做什么。 Action: 当你需要调用工具时,请严格按照以下格式输出:`Action: <tool_name>[[<input>]]` 例如:Action: calculator[[3 + 5]] Observation: 工具调用后的结果会在这里提供给你。 ... (这个 Thought/Action/Observation 循环可以重复多次) Final Answer: 当你认为已经有了最终答案时,请以此开头输出最终结果。 你可以使用的工具: {tool_descriptions} 现在,开始回答用户的问题。请确保每一步都严格遵守上述格式。 """ return prompt这段提示词是Agent的“宪法”,它规定了LLM的输出格式和行为准则。清晰的格式定义(如Action: tool_name[[input]])是后续能够用简单字符串匹配来解析输出的关键。
# 4. ReAct循环引擎核心 def run_agent(user_query, max_steps=10): """执行ReAct循环的主函数""" system_prompt = build_system_prompt() # 初始化对话历史,包含系统指令 conversation_history = [{"role": "system", "content": system_prompt}] # 添加用户问题 conversation_history.append({"role": "user", "content": user_query}) step = 0 while step < max_steps: step += 1 # 4.1 思考与行动:调用LLM获取下一步指令 # 将整个对话历史作为上下文发送给LLM prompt_for_llm = "\n".join([f"{msg['role']}: {msg['content']}" for msg in conversation_history]) llm_response = call_llm(prompt_for_llm) print(f"\n--- Step {step} ---") print(f"LLM Response:\n{llm_response}") # 4.2 解析LLM的响应 if "Final Answer:" in llm_response: final_answer = llm_response.split("Final Answer:")[-1].strip() print(f"\n✅ 任务完成!最终答案:{final_answer}") return final_answer # 解析 Action 行 action_line = None for line in llm_response.split('\n'): if line.startswith('Action:'): action_line = line break if not action_line: # 如果LLM没有输出Action,可能是格式错误或它想直接回答但没写Final Answer print("⚠️ LLM响应中未找到有效的Action行。将其作为观察加入历史,让LLM继续思考。") conversation_history.append({"role": "assistant", "content": llm_response}) conversation_history.append({"role": "user", "content": "请严格按照要求的格式(Thought/Action/Observation)进行回应。如果你有答案,请使用'Final Answer:'开头。"}) continue # 提取工具名和输入参数 try: # 预期格式:Action: calculator[[3 + 5]] action_part = action_line.replace('Action:', '').strip() tool_name = action_part.split('[')[0].strip() input_str = action_part.split('[[', 1)[1].rsplit(']]', 1)[0].strip() print(f"解析结果 -> 工具: {tool_name}, 输入: {input_str}") except IndexError: error_msg = f"无法解析Action行: {action_line}" print(f"❌ {error_msg}") conversation_history.append({"role": "assistant", "content": llm_response}) conversation_history.append({"role": "user", "content": f"Action格式解析失败。{error_msg} 请检查格式是否为'Action: tool_name[[input]]'。"}) continue # 4.3 执行工具调用(观察) if tool_name not in TOOLS: observation = f"错误:未知工具 '{tool_name}'。可用工具:{list(TOOLS.keys())}" else: tool_func = TOOLS[tool_name]["function"] try: observation = tool_func(input_str) print(f"工具执行结果: {observation}") except Exception as e: observation = f"工具执行时出错: {e}" # 4.4 将本轮循环的完整响应和观察结果加入历史,用于下一轮推理 conversation_history.append({"role": "assistant", "content": llm_response}) conversation_history.append({"role": "user", "content": f"Observation: {observation}"}) # 循环结束仍未得到最终答案 print(f"\n❌ 达到最大步数限制({max_steps}),任务未完成。") return None这就是整个Agent的核心循环。让我们拆解一下while循环里的关键步骤:
- 组装提示:将包含所有历史消息(系统指令、用户问题、之前的思考、行动、观察)的完整上下文发送给LLM。
- 解析响应:用字符串查找和分割这种“笨”但极其可靠的方法,从LLM的回复中提取
Final Answer或Action指令。这里没有用复杂的正则表达式,是为了保持代码的极简和可读性。 - 工具分发与执行:根据解析出的工具名,从
TOOLS字典中找到对应的函数并执行。 - 更新历史:将LLM的本轮回复和工具执行的观察结果,以固定的格式追加到对话历史中。这是实现多轮推理的关键,它让LLM在下一轮调用时,能“看到”自己之前的思考和行动结果。
# 5. 启动Agent if __name__ == "__main__": # 测试几个问题 test_queries = [ "今天星期几?现在几点了?", "请计算一下 (15 + 7) * 3 的值是多少?", "先查一下现在的时间,然后基于这个时间,计算100小时后的日期和时间。" ] for query in test_queries: print(f"\n{'='*50}") print(f"用户问题: {query}") print(f"{'='*50}") result = run_agent(query)运行这段代码,你会看到Agent一步步思考、调用工具、获取结果,最终给出答案的完整过程。例如,对于第三个复合问题,它的输出可能类似于:
用户问题: 先查一下现在的时间,然后基于这个时间,计算100小时后的日期和时间。 ================================================== --- Step 1 --- LLM Response: Thought: 用户需要先知道当前时间,然后计算100小时后的时间。我应该先调用get_current_time工具。 Action: get_current_time[[]] 解析结果 -> 工具: get_current_time, 输入: 工具执行结果: 2023-10-27 14:30:00 --- Step 2 --- LLM Response: Thought: 我得到了当前时间是2023-10-27 14:30:00。现在需要计算100小时后的时间。这需要日期时间的计算,我可以利用calculator工具吗?但calculator主要用于数学表达式。或许我需要先理解100小时是多少天零多少小时...实际上,我可以将当前时间转换为时间戳(秒数),加上100*3600秒,再转换回来。但我没有直接的时间计算工具。让我思考...我可以让用户提供这个计算,或者我手动估算?等等,我可以使用calculator计算100*3600得到增加的秒数,但将日期时间字符串转换为时间戳并进行计算,超出了当前工具的能力。我需要更精确的方法。或许我应该直接给出基于当前时间的估算?但用户要求“计算”。我可能需要承认工具的限制。 Action: calculator[[100 * 3600]] 解析结果 -> 工具: calculator, 输入: 100 * 3600 工具执行结果: 360000 --- Step 3 --- LLM Response: Thought: 我计算出100小时是360000秒。但我无法直接将这个秒数加到日期时间字符串上。我现有的工具无法完成精确的日期时间加减。我应该向用户说明这一点,并提供一个基于简单算术的估算:100小时大约是4天又4小时(因为24*4=96,100-96=4)。所以从2023-10-27 14:30:00开始,加上4天是2023-10-31 14:30:00,再加上4小时是2023-10-31 18:30:00。这是一个估算。 Final Answer: 当前时间是2023-10-27 14:30:00。100小时后大约是2023-10-31 18:30:00(此为估算,未考虑夏令时等复杂因素)。请注意,我的工具库无法进行精确的日期时间加减运算。 ✅ 任务完成!最终答案:当前时间是2023-10-27 14:30:00。100小时后大约是2023-10-31 18:30:00(此为估算,未考虑夏令时等复杂因素)。请注意,我的工具库无法进行精确的日期时间加减运算。这个过程完美展示了ReAct的威力:LLM意识到工具能力的不足,尝试用已有工具(计算秒数)部分解决问题,并在最终答案中诚实地说明了局限性和估算方法。
4. 关键细节与深度优化点
虽然核心代码只有50行,但其中蕴含了许多设计和权衡。以下是几个关键的优化和思考方向:
4.1 提示词工程:稳定输出的基石
系统提示词的质量直接决定了Agent的成败。除了定义格式,还有几个技巧:
- 明确边界:在提示词中强调“只能使用上述工具”,可以有效减少LLM“幻想”出不存在工具的情况。
- 示例的力量(Few-Shot):在系统提示词中加入一两个完整的ReAct循环示例,能极大地提升LLM输出格式的稳定性。例如,在工具描述后加上:
示例: 用户:3的5次方是多少? 助手:Thought: 用户需要计算幂运算。我可以使用calculator工具。 Action: calculator[[3 ** 5]] Observation: 243 Thought: 我得到了结果243。 Final Answer: 3的5次方等于243。 - 错误处理指引:可以提示LLM,如果工具调用失败或返回错误,应该如何处理(例如,“如果Observation显示错误,请分析原因并尝试其他方法或给出解释”)。
4.2 工具设计:能力与安全的平衡
- 输入验证与清洗:在工具函数内部,务必对输入进行验证。例如,在
calculator函数中,我们使用了危险的eval。在生产环境中,这是不可接受的。应该替换为安全的替代方案,比如使用ast.literal_eval处理数字和简单运算,或者使用numexpr、pandas.eval等受限的表达式求值库,甚至完全自己实现一个四则运算解析器。import ast import operator def safe_calculator(expression): """一个相对安全的计算器(示例,仍有限制)""" allowed_operators = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv, ast.Pow: operator.pow, ast.USub: operator.neg} def eval_node(node): if isinstance(node, ast.Num): # 注意:Python 3.8+ 中 ast.Num 已弃用,需用 ast.Constant return node.n elif isinstance(node, ast.BinOp): left_val = eval_node(node.left) right_val = eval_node(node.right) op_func = allowed_operators.get(type(node.op)) if op_func is None: raise ValueError(f"不允许的操作符: {type(node.op)}") return op_func(left_val, right_val) elif isinstance(node, ast.UnaryOp): operand_val = eval_node(node.operand) op_func = allowed_operators.get(type(node.op)) return op_func(operand_val) else: raise ValueError(f"不支持的表达式结构: {type(node)}") try: tree = ast.parse(expression, mode='eval') return str(eval_node(tree.body)) except (SyntaxError, ValueError, TypeError) as e: return f"计算错误或表达式不安全: {e}" - 工具描述的颗粒度:描述越详细,LLM调用越准确。可以包括参数类型、返回格式、常见错误示例等。
4.3 循环控制与错误恢复
- 最大步数限制:
max_steps是必要的安全阀,防止Agent陷入死循环。 - 解析鲁棒性:我们的解析逻辑比较简单。更健壮的做法是使用更精确的正则表达式,或者要求LLM以JSON等结构化格式输出,但这会增加提示词的复杂性和对模型的要求。
- 错误反馈循环:当解析失败或工具调用出错时,我们将错误信息以
Observation的形式反馈给LLM。这给了LLM自我纠正的机会,是构建稳健Agent的重要机制。
5. 常见问题与实战调试技巧
在实际运行中,你可能会遇到以下典型问题:
5.1 LLM不按格式输出
- 症状:LLM的回复是自由文本,没有“Thought:”、“Action:”等前缀。
- 排查:
- 检查系统提示词:确保格式指令清晰、无歧义。加入Few-Shot示例效果立竿见影。
- 降低Temperature:确保调用LLM时
temperature=0,以获得最确定性的输出。 - 检查上下文:确保每次调用LLM时,完整的对话历史(包括之前所有正确的格式回合)都被发送了过去。LLM会根据历史来延续风格。
- 应急处理:就像我们代码里做的,如果没找到
Action行,就把LLM的回复当作Observation反馈回去,并提醒它遵守格式。通常经过一两轮纠正,LLM就会回到正轨。
5.2 工具调用错误或参数不对
- 症状:
Observation里是工具返回的错误信息。 - 排查:
- 看工具输入:打印出解析得到的
input_str,看是否符合工具函数的要求。LLM可能会生成多余的空格、引号或解释性文字。 - 优化工具描述:在工具描述中,用“输入必须是一个纯数字表达式”或“输入应为‘城市名’的格式”来严格约束。
- 在提示词中示例:在系统提示词的示例部分,展示一个参数传递正确的工具调用案例。
- 看工具输入:打印出解析得到的
- 进阶技巧:你可以让LLM在
Thought阶段先“说出”它打算传递给工具的准确参数,这样在调试时更容易定位问题。
5.3 Agent陷入循环或逻辑混乱
- 症状:Agent反复调用同一个工具,或者思考过程明显偏离正轨。
- 排查:
- 观察Thought内容:仔细阅读LLM的
Thought,看它的推理逻辑在哪里出现了断裂或误解。 - 简化任务:用一个极其简单的问题(如“1+1等于几?”)测试,看基础循环是否正常。如果正常,说明复杂任务可能超出了当前提示词和工具集的设计。
- 检查历史累积:在长时间循环后,对话历史会变得很长。有些模型有上下文长度限制,可能导致最早的指令被“遗忘”。可以考虑只保留最近几轮的交互,或者对历史进行摘要(但这会显著增加复杂度)。
- 观察Thought内容:仔细阅读LLM的
- 解决:根据
Thought中的错误推理,针对性调整系统提示词。例如,如果LLM总想调用不存在的工具,就在提示词里强调“严禁使用列表外的工具”。如果LLM在得到答案后还不停止,就强调“一旦得到确信的答案,必须立即以‘Final Answer:‘开头输出”。
5.4 性能与成本考量
- 每次循环都是一次API调用:ReAct循环的每一步都需要调用一次LLM。对于复杂任务,步数可能很多,这意味着更高的延迟和API成本。
- 优化方向:
- 任务分解:对于非常复杂的问题,可以尝试在外部先将其分解成几个子问题,然后让Agent逐个解决,而不是完全依赖Agent自己规划所有步骤。
- 设置超时和步数限制:务必设置合理的
max_steps,并在实际应用中设置超时机制。 - 选择性价比高的模型:对于工具调用这类格式要求高但推理深度相对可控的任务,
gpt-3.5-turbo通常比gpt-4更具性价比,且速度更快。
通过这个仅50行代码的项目,我们亲手搭建了一个AI Agent的“心脏”。它简陋,但完整;它直接,却清晰地揭示了智能体与工具交互的本质。当你下次再使用LangChain这样的强大框架时,你会更清楚底层那些AgentExecutor、Tool类到底在帮你管理什么。更重要的是,当你有某个非常定制化、轻量级的场景时,你完全有能力甩开框架,用这几十行代码快速构建一个专属于你的、高效的小智能体。这就是理解核心原理的价值所在。