简介:本文从LLM与Agent的本质差异切入,带你在不依赖任何特定厂商框架的前提下,用约150行Python代码从零实现一个具备“规划-记忆-工具调用”能力的极简AI Agent。文章适合有Python基础、希望深入理解Agent工作原理的开发者,通过手写代码理解其核心调度逻辑,为后续接入生产级平台(如云服务商Agent框架)打下坚实基础。
本教程的目标是帮助你在不依赖任何第三方Agent框架的前提下,用约150行标准Python代码实现一个具备基本推理、记忆和工具调用能力的AI Agent。这个Agent能够接收自然语言指令,自主决定调用哪些本地工具(如计算器、文件读取器、时间查询器等),并循环执行“思考-行动-观察”的流程,直至得出最终答案。
通过本教程,你将掌握:
本教程适合以下读者:
典型应用场景包括:内部工单自动回复、日志分析辅助、定时报告生成、简单数据查询与计算等。
开始前,你需要准备:
需要注意的是,本教程不涉及具体云服务商的接入细节,仅从算法和调度逻辑层面还原Agent的骨架。若要接入生产环境,可结合主流云平台提供的模型服务或自建模型服务完成。
先明确一个关键概念:LLM本质是一个“静态大脑”。你给它一段文本,它生成下一段文本。它没有手、没有脚,不主动获取新信息,也不记得你上周问过它什么。
Agent则是一个“闭环系统”。它在大脑(LLM)之外,加装了:
Agent的核心工作循环可概括为:分析当前状态 -> 决策下一步动作 -> 执行动作 -> 观察结果 -> 更新状态 -> 重复上述过程,直到达成目标。这个模式常被称为“思考-行动-观察”循环。
在动笔写代码前,我们先定义Agent的四个核心组件:
我们不直接绑定任何具体模型服务,而是定义一个函数,接收消息列表,返回文本。这样方便你在本地或云端替换。
# llm_interface.pydef call_llm(messages: list) -> str:"""模拟LLM调用接口。实际使用中,您可将此函数替换为调用云端模型服务或本地模型的代码。messages: 形如 [{"role": "user", "content": "..."}, {"role": "assistant", "content": "..."}]"""# 此处仅作示意,具体实现需替换为真实模型API调用# 例如:response = your_model_client.chat(messages)# return responsepass
工具是Agent的“手脚”。每个工具都是一个普通Python函数。我们用装饰器将其注册到一个全局字典中。
# tools.pyTOOL_MAP = {}def register_tool(name: str):def decorator(func):TOOL_MAP[name] = funcreturn funcreturn decorator@register_tool("calculator")def calculate(expression: str) -> str:"""计算数学表达式,例如 '3+5*2'"""try:# 注意:生产环境中应使用更安全的eval替代方案(如ast.literal_eval)result = eval(expression)return f"计算结果是: {result}"except Exception as e:return f"计算出错: {str(e)}"@register_tool("get_current_time")def get_time() -> str:"""返回当前时间"""from datetime import datetimereturn f"当前时间是: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"@register_tool("read_file")def read_file(file_path: str) -> str:"""读取指定路径的文本文件内容(用于演示,生产环境需加权限控制)"""try:with open(file_path, 'r', encoding='utf-8') as f:content = f.read()return f"文件内容:\n{content[:500]}" # 截断避免过长except Exception as e:return f"读取失败: {str(e)}"
Agent类负责组装记忆、调用LLM、解析动作、执行工具。
# agent.pyimport jsonfrom typing import List, Dictclass Agent:def __init__(self, llm_function):self.llm_function = llm_functionself.messages: List[Dict[str, str]] = []self.max_iterations = 5 # 防止无限循环def add_user_message(self, content: str):self.messages.append({"role": "user", "content": content})def run(self, task: str) -> str:self.add_user_message(task)for i in range(self.max_iterations):# 1. 调用LLM决策下一步response = self.llm_function(self.messages)self.messages.append({"role": "assistant", "content": response})# 2. 判断是否结束if "FINAL_ANSWER:" in response:final_answer = response.split("FINAL_ANSWER:")[1].strip()return final_answer# 3. 解析工具调用(格式约定为:ACTION: 工具名 | 参数JSON)if "ACTION:" in response:action_line = response.split("ACTION:")[1].split("|")[0].strip()param_line = response.split("|")[1].strip() if "|" in response else "{}"tool_name = action_linetry:params = json.loads(param_line)except json.JSONDecodeError:params = {}# 4. 执行工具if tool_name in TOOL_MAP:tool_func = TOOL_MAP[tool_name]try:observation = tool_func(**params) if isinstance(params, dict) else tool_func(params)except Exception as e:observation = f"工具执行失败: {str(e)}"else:observation = f"未知工具: {tool_name},可用的工具有: {list(TOOL_MAP.keys())}"# 5. 将观察结果加入记忆self.messages.append({"role": "user", "content": f"观察结果: {observation}"})return "达到最大迭代次数,未完成目标。"# 全局注册表引入(在实际模块中应有更规范的处理)from tools import TOOL_MAP
为了让LLM真正驱动Agent,我们需要通过系统提示词约束其输出格式。这是Agent效果的灵魂。
system_prompt = """你是一个智能Agent,你有以下工具可用:- calculator(expression): 计算数学表达式- get_current_time(): 获取当前时间- read_file(file_path): 读取文件内容你必须严格遵循以下输出格式:1. 当你需要调用工具时,输出一行:ACTION: 工具名 | {"参数1": "值1", "参数2": "值2"}2. 当你已经获取足够信息能回答用户时,输出一行:FINAL_ANSWER: 最终的答案请记住:你的目标是尽可能帮助用户解决问题。如果工具调用失败,请基于观察结果调整你的方案。请开始你的任务,先输出你的思考过程(用一句话描述),然后输出动作。"""
# main.pyfrom agent import Agentdef simple_llm_chat(messages):"""这里仅提供一个模拟LLM响应的示例。实际使用时,请将messages传给真实的模型API,并确保返回的文本符合上述格式约束。"""# 模拟一个简单的行为:如果用户问时间,就调用工具;如果用户问计算,就调用计算器。last_user_msg = [m for m in reversed(messages) if m['role'] == 'user'][0]['content']if '时间' in last_user_msg:return "我需要查询时间。\nACTION: get_current_time | {}"elif '计算' in last_user_msg or '加' in last_user_msg:# 这里简化提取表达式return f"我需要进行计算。\nACTION: calculator | {{\"expression\": \"{last_user_msg.split('计算')[1].strip()}\"}}"else:# 如果已有观察结果,返回最终答案if any(m['content'].startswith('观察结果') for m in messages):obs = [m['content'] for m in messages if m['content'].startswith('观察结果')][-1]return f"FINAL_ANSWER: 根据查询,{obs}"return "FINAL_ANSWER: 抱歉,我暂时无法处理这个请求。"agent = Agent(llm_function=simple_llm_chat)# 设置系统提示词agent.messages.append({"role": "system", "content": system_prompt})# 运行任务result = agent.run("现在几点了?")print(result)result2 = agent.run("帮我计算 123 * 456 的结果。")print(result2)
| 问题现象 | 可能原因 | 解决思路 |
|---|---|---|
| LLM总是输出FINAL_ANSWER,不调用工具 | 系统提示词约束不够强硬 | 加强提示词中的格式要求,增加“除非调用工具,否则无法完成任务”等措辞 |
| 工具调用后,Agent无法生成最终答案 | 观察结果未正确拼接到提示中 | 检查观察结果是否作为user消息追加;尝试在系统提示词中加入“结合观察结果来回答”的指令 |
| 工具参数解析错误 | LLM返回的JSON格式不规范 | 在提示词中给出具体的JSON示例;使用更宽松的解析逻辑(如正则提取第一个花括号) |
| 达到最大迭代次数 | 任务过于复杂或提示词引导不足 | 增加max_iterations;分解任务;优化工具设计以降低调用次数 |
增强安全性与健壮性
eval(),应改用安全表达式求值库或抽象语法树解析;引入持久化记忆
当前Agent只在单次运行内记忆。若需长期跨会话记忆,可将messages序列化存储到数据库或云数据库服务中。
支持动态工具发现
可以将工具注册表提升为服务发现机制,允许Agent在运行中动态加载新工具,便于扩展。
使用真实生产级LLM服务
将call_llm替换为云服务商提供的模型API调用代码,并配置合适的温度、最大token数等参数,以平衡速度与准确性。
增加日志与监控
记录每次Agent调用的输入、输出、工具执行耗时,便于追踪异常和性能调优。
本教程通过约150行Python代码,完整搭建了一个具备“规划-记忆-工具调用”能力的AI Agent核心骨架。你已理解:
在此基础上,你可以进一步探索更复杂的任务规划算法(如任务分解)、记忆压缩技术、多Agent协作模式等。若希望快速在生产环境落地,可评估主流云服务商提供的Agent编排平台,将本教程中手写的调度逻辑交给平台托管,以换取更高的稳定性和可观测性。但无论使用何种框架,其内核依然是你现在掌握的这套循环逻辑。