从理论到代码:150行Python实现一个AI Agent核心逻辑

作者:da吃一鲸8862026.11.04 11:24浏览量:0

简介:本文从LLM与Agent的本质差异切入,带你在不依赖任何特定厂商框架的前提下,用约150行Python代码从零实现一个具备“规划-记忆-工具调用”能力的极简AI Agent。文章适合有Python基础、希望深入理解Agent工作原理的开发者,通过手写代码理解其核心调度逻辑,为后续接入生产级平台(如云服务商Agent框架)打下坚实基础。

agent">一、教程目标:理解并亲手搭建一个微型Agent

本教程的目标是帮助你在不依赖任何第三方Agent框架的前提下,用约150行标准Python代码实现一个具备基本推理、记忆和工具调用能力的AI Agent。这个Agent能够接收自然语言指令,自主决定调用哪些本地工具(如计算器、文件读取器、时间查询器等),并循环执行“思考-行动-观察”的流程,直至得出最终答案。

通过本教程,你将掌握:

  1. LLM(大语言模型)与Agent的根本区别;
  2. Agent核心循环(ReAct模式)的底层实现原理;
  3. 如何用通用代码封装工具注册与调用机制;
  4. 如何扩展Agent以适配不同业务场景。

二、适用场景与读者对象

本教程适合以下读者:

  • 有一定Python基础,了解基本函数和类定义的开发者;
  • 对AI Agent感兴趣,但不满足于仅调用现成框架、希望理解其内核原理的技术人员;
  • 需要为团队搭建轻量级内部自动化工具,但不想引入重量级依赖的工程师。

典型应用场景包括:内部工单自动回复、日志分析辅助、定时报告生成、简单数据查询与计算等。

三、前置准备

开始前,你需要准备:

  1. Python 3.8及以上版本环境;
  2. 一个可调用的LLM API客户端(本教程使用抽象接口,不绑定任何具体厂商,你可以根据自身环境替换为实际可用的SDK调用代码);
  3. 基础开发工具:文本编辑器或IDE。

需要注意的是,本教程不涉及具体云服务商的接入细节,仅从算法和调度逻辑层面还原Agent的骨架。若要接入生产环境,可结合主流云平台提供的模型服务或自建模型服务完成。

四、核心理论:LLM与Agent的差异

先明确一个关键概念:LLM本质是一个“静态大脑”。你给它一段文本,它生成下一段文本。它没有手、没有脚,不主动获取新信息,也不记得你上周问过它什么。

Agent则是一个“闭环系统”。它在大脑(LLM)之外,加装了:

  • 感知层:从外界接收任务指令;
  • 行动层:调用工具(如执行代码、读取文件、发送网络请求)来改变环境状态或获取新信息;
  • 记忆层:存储短期对话历史或长期业务知识。

Agent的核心工作循环可概括为:分析当前状态 -> 决策下一步动作 -> 执行动作 -> 观察结果 -> 更新状态 -> 重复上述过程,直到达成目标。这个模式常被称为“思考-行动-观察”循环。

五、设计一个极简Agent架构

在动笔写代码前,我们先定义Agent的四个核心组件:

  1. LLM接口(Brain):负责接收当前任务描述和已观察到的上下文,返回Agent应采取的下一步动作文本。动作文本可以是“执行某个工具”或“回答用户”。
  2. 工具注册表(Hands):一个字典,键为工具名,值为可调用函数。Agent根据LLM的输出选择调用哪个工具。
  3. 记忆组件(Memory):维护一个消息列表,记录用户原话、Agent的思考过程、工具返回结果。这相当于短时工作记忆。
  4. 调度循环(Body):在一个while循环中,反复调用LLM决策、解析动作、执行工具、追加观察结果,直到LLM认定任务已完成并输出最终答案。

六、逐步实现:150行Python代码拆解

步骤1:定义LLM抽象接口

我们不直接绑定任何具体模型服务,而是定义一个函数,接收消息列表,返回文本。这样方便你在本地或云端替换。

  1. # llm_interface.py
  2. def call_llm(messages: list) -> str:
  3. """
  4. 模拟LLM调用接口。
  5. 实际使用中,您可将此函数替换为调用云端模型服务或本地模型的代码。
  6. messages: 形如 [{"role": "user", "content": "..."}, {"role": "assistant", "content": "..."}]
  7. """
  8. # 此处仅作示意,具体实现需替换为真实模型API调用
  9. # 例如:response = your_model_client.chat(messages)
  10. # return response
  11. pass

步骤2:定义工具注册机制

工具是Agent的“手脚”。每个工具都是一个普通Python函数。我们用装饰器将其注册到一个全局字典中。

  1. # tools.py
  2. TOOL_MAP = {}
  3. def register_tool(name: str):
  4. def decorator(func):
  5. TOOL_MAP[name] = func
  6. return func
  7. return decorator
  8. @register_tool("calculator")
  9. def calculate(expression: str) -> str:
  10. """计算数学表达式,例如 '3+5*2'"""
  11. try:
  12. # 注意:生产环境中应使用更安全的eval替代方案(如ast.literal_eval)
  13. result = eval(expression)
  14. return f"计算结果是: {result}"
  15. except Exception as e:
  16. return f"计算出错: {str(e)}"
  17. @register_tool("get_current_time")
  18. def get_time() -> str:
  19. """返回当前时间"""
  20. from datetime import datetime
  21. return f"当前时间是: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
  22. @register_tool("read_file")
  23. def read_file(file_path: str) -> str:
  24. """读取指定路径的文本文件内容(用于演示,生产环境需加权限控制)"""
  25. try:
  26. with open(file_path, 'r', encoding='utf-8') as f:
  27. content = f.read()
  28. return f"文件内容:\n{content[:500]}" # 截断避免过长
  29. except Exception as e:
  30. return f"读取失败: {str(e)}"

步骤3:实现Agent主类

Agent类负责组装记忆、调用LLM、解析动作、执行工具。

  1. # agent.py
  2. import json
  3. from typing import List, Dict
  4. class Agent:
  5. def __init__(self, llm_function):
  6. self.llm_function = llm_function
  7. self.messages: List[Dict[str, str]] = []
  8. self.max_iterations = 5 # 防止无限循环
  9. def add_user_message(self, content: str):
  10. self.messages.append({"role": "user", "content": content})
  11. def run(self, task: str) -> str:
  12. self.add_user_message(task)
  13. for i in range(self.max_iterations):
  14. # 1. 调用LLM决策下一步
  15. response = self.llm_function(self.messages)
  16. self.messages.append({"role": "assistant", "content": response})
  17. # 2. 判断是否结束
  18. if "FINAL_ANSWER:" in response:
  19. final_answer = response.split("FINAL_ANSWER:")[1].strip()
  20. return final_answer
  21. # 3. 解析工具调用(格式约定为:ACTION: 工具名 | 参数JSON)
  22. if "ACTION:" in response:
  23. action_line = response.split("ACTION:")[1].split("|")[0].strip()
  24. param_line = response.split("|")[1].strip() if "|" in response else "{}"
  25. tool_name = action_line
  26. try:
  27. params = json.loads(param_line)
  28. except json.JSONDecodeError:
  29. params = {}
  30. # 4. 执行工具
  31. if tool_name in TOOL_MAP:
  32. tool_func = TOOL_MAP[tool_name]
  33. try:
  34. observation = tool_func(**params) if isinstance(params, dict) else tool_func(params)
  35. except Exception as e:
  36. observation = f"工具执行失败: {str(e)}"
  37. else:
  38. observation = f"未知工具: {tool_name},可用的工具有: {list(TOOL_MAP.keys())}"
  39. # 5. 将观察结果加入记忆
  40. self.messages.append({"role": "user", "content": f"观察结果: {observation}"})
  41. return "达到最大迭代次数,未完成目标。"
  42. # 全局注册表引入(在实际模块中应有更规范的处理)
  43. from tools import TOOL_MAP

步骤4:构建Prompt模板(引导LLM输出规范)

为了让LLM真正驱动Agent,我们需要通过系统提示词约束其输出格式。这是Agent效果的灵魂。

  1. system_prompt = """
  2. 你是一个智能Agent,你有以下工具可用:
  3. - calculator(expression): 计算数学表达式
  4. - get_current_time(): 获取当前时间
  5. - read_file(file_path): 读取文件内容
  6. 你必须严格遵循以下输出格式:
  7. 1. 当你需要调用工具时,输出一行:
  8. ACTION: 工具名 | {"参数1": "值1", "参数2": "值2"}
  9. 2. 当你已经获取足够信息能回答用户时,输出一行:
  10. FINAL_ANSWER: 最终的答案
  11. 请记住:你的目标是尽可能帮助用户解决问题。如果工具调用失败,请基于观察结果调整你的方案。
  12. 请开始你的任务,先输出你的思考过程(用一句话描述),然后输出动作。
  13. """

步骤5:实例化并启动Agent

  1. # main.py
  2. from agent import Agent
  3. def simple_llm_chat(messages):
  4. """
  5. 这里仅提供一个模拟LLM响应的示例。
  6. 实际使用时,请将messages传给真实的模型API,
  7. 并确保返回的文本符合上述格式约束。
  8. """
  9. # 模拟一个简单的行为:如果用户问时间,就调用工具;如果用户问计算,就调用计算器。
  10. last_user_msg = [m for m in reversed(messages) if m['role'] == 'user'][0]['content']
  11. if '时间' in last_user_msg:
  12. return "我需要查询时间。\nACTION: get_current_time | {}"
  13. elif '计算' in last_user_msg or '加' in last_user_msg:
  14. # 这里简化提取表达式
  15. return f"我需要进行计算。\nACTION: calculator | {{\"expression\": \"{last_user_msg.split('计算')[1].strip()}\"}}"
  16. else:
  17. # 如果已有观察结果,返回最终答案
  18. if any(m['content'].startswith('观察结果') for m in messages):
  19. obs = [m['content'] for m in messages if m['content'].startswith('观察结果')][-1]
  20. return f"FINAL_ANSWER: 根据查询,{obs}"
  21. return "FINAL_ANSWER: 抱歉,我暂时无法处理这个请求。"
  22. agent = Agent(llm_function=simple_llm_chat)
  23. # 设置系统提示词
  24. agent.messages.append({"role": "system", "content": system_prompt})
  25. # 运行任务
  26. result = agent.run("现在几点了?")
  27. print(result)
  28. result2 = agent.run("帮我计算 123 * 456 的结果。")
  29. print(result2)

七、验证结果与常见问题排查

7.1 如何验证Agent正常工作?

  • 观察日志:在Agent循环中打印每次的response和observation,确认流程正确推进;
  • 测试简单场景:先测试单一工具调用(如查时间),再测试需要多步推理的复杂任务;
  • 检查记忆增长:确认messages列表长度随迭代增加,且包含正确的角色交替序列。

7.2 常见问题排查

问题现象 可能原因 解决思路
LLM总是输出FINAL_ANSWER,不调用工具 系统提示词约束不够强硬 加强提示词中的格式要求,增加“除非调用工具,否则无法完成任务”等措辞
工具调用后,Agent无法生成最终答案 观察结果未正确拼接到提示中 检查观察结果是否作为user消息追加;尝试在系统提示词中加入“结合观察结果来回答”的指令
工具参数解析错误 LLM返回的JSON格式不规范 在提示词中给出具体的JSON示例;使用更宽松的解析逻辑(如正则提取第一个花括号)
达到最大迭代次数 任务过于复杂或提示词引导不足 增加max_iterations;分解任务;优化工具设计以降低调用次数

八、优化建议:从演示走向生产

  1. 增强安全性与健壮性

    • 工具函数中严禁直接使用eval(),应改用安全表达式求值库或抽象语法树解析;
    • 对文件读取、网络请求等工具做权限校验和路径白名单限制。
  2. 引入持久化记忆
    当前Agent只在单次运行内记忆。若需长期跨会话记忆,可将messages序列化存储到数据库或云数据库服务中。

  3. 支持动态工具发现
    可以将工具注册表提升为服务发现机制,允许Agent在运行中动态加载新工具,便于扩展。

  4. 使用真实生产级LLM服务
    将call_llm替换为云服务商提供的模型API调用代码,并配置合适的温度、最大token数等参数,以平衡速度与准确性。

  5. 增加日志与监控
    记录每次Agent调用的输入、输出、工具执行耗时,便于追踪异常和性能调优。

九、总结

本教程通过约150行Python代码,完整搭建了一个具备“规划-记忆-工具调用”能力的AI Agent核心骨架。你已理解:

  • LLM仅提供智力支持,Agent通过循环调度让智力转化为行动;
  • 工具注册表是Agent扩展能力的核心接口;
  • 记忆列表是维持上下文连贯性的关键;
  • 提示词工程是控制LLM行为模式最直接的手段。

在此基础上,你可以进一步探索更复杂的任务规划算法(如任务分解)、记忆压缩技术、多Agent协作模式等。若希望快速在生产环境落地,可评估主流云服务商提供的Agent编排平台,将本教程中手写的调度逻辑交给平台托管,以换取更高的稳定性和可观测性。但无论使用何种框架,其内核依然是你现在掌握的这套循环逻辑。