Agent原理 实现(快速上手)

📅 2026/7/29 3:25:27
Agent原理  实现(快速上手)
Agent原理核心定义1LLM被动应答你问一句它答一句不会主动做任何操作不会主动调取外部资源。2Agent主动闭环思考Thought → 调用工具Action → 获取结果Observation自主循环直到任务完成不用人类分步指挥。经典架构ReAct 架构最普及Action → Observation 循环1Thought大模型思考下一步需要做什么2Action输出工具调用指令调用代码、接口、检索、数据库3Observation拿到工具返回返回工具执行结果把结果添加到上下文输入大模型进行下一轮循环往复直到模型判断任务完成输出最终答案。注Agent执行过程中每次大模型输入包含用户任务 历史上下文 工具函数执行结果。优点实现简单、动态适配未知场景不需要预先写完整计划。缺点长任务容易跑偏、陷入循环复杂多步骤任务容易遗漏步骤。适用场景通用助手、检索增强 RAG Agent、代码执行 AgentClaude Code、Cursor 底层基础范式Plan-and-Solve / Plan Execute规划执行架构1Planning 阶段LLM 一次性生成完整任务步骤清单2Executor 阶段按照清单依次执行工具调用每一步执行结果反馈可动态微调计划。优点长流程任务逻辑性更强不容易丢步骤便于人工审核计划。缺点任务未知、环境多变时初始计划容易失效灵活性不如 ReAct。适用场景标准化工作流、数据分析流水线、自动化运维任务。Self-Ask自我提问架构模型不断向自己提出子问题通过工具回答子问题最终汇总得到主问题答案。示例“武汉 7 月平均气温” 自提问 1武汉 7 月历史气温数据在哪里获取→ 调用搜索 自提问 2如何计算月平均值适用场景信息搜集类、多事实推理任务常作为 ReAct 内部增强策略。Agent框架LangGraphLangGraph团队开发的一款开源 AI Agent 框架主要用来帮开发者搭建、部署和管理那些比较复杂的生成式 AI Agent 工作流。通信机制Agent 之间不直接通信所有读写操作围绕【全局共享 State】Agent间不需要大量交互对话Token开销低1优点流程高度可控适合工程化落地生产服务原生支持循环、分支、多路并行、断点续跑状态结构化方便日志、监控、回放测试天然适配ReAct循环推理2缺点需要设计规范的State数据结构AutoGenAutoGen 是微软开发的开源多智能体 AI 框架‌用于构建 AI 代理应用支持多个智能体通过对话协作完成复杂任务。通信机制依靠 Manager/GroupChatManager 做消息转发Agent 拥有独立上下文无统一共享状态容器。Agent间需要大量交互对话Token开销高1优点快速搭建多角色Agent模型开箱即用GroupChat非常适合探索型任务、代码调试、多Agent辩论Agent可以自主协商分工不需要预先写死流程内置代码沙箱执行能力2缺点流程不可控极易出现无限对话、任务跑偏缺少持久化断点机制大量对话消息叠加token成本高生产长会话场景可观测性差难以定位异常不适合强约束、确定性要求高的自动化运维 或 工具Agent单Agent实现代码结构代码目录agent_demo │ ├─agent.py ├─prompt_template.py └─.env核心代码1agent.pyimport ast import inspect import os import re from string import Template from typing import List, Callable, Tuple import click from dotenv import load_dotenv from openai import OpenAI import platform from prompt_template import react_system_prompt_template class ReActAgent: def __init__(self, tools: List[Callable], model: str, project_directory: str): self.tools {func.__name__: func for func in tools} self.model model self.project_directory project_directory self.client OpenAI( base_urlhttps://kspmas.ksyun.com/v1/, api_keyReActAgent.get_api_key(), ) def run(self, user_input: str): # 上下文消息列表完整保存每一轮对话所有内容每次对话都会把整个消息列表传给大模型 messages [ {role: system, content: self.render_system_prompt(react_system_prompt_template)}, {role: user, content: fquestion{user_input}/question} ] # 无限循环持续执行ReAct思考 while True: # 请求模型获取大模型返回文本同时自动把模型回复存入messages content self.call_model(messages) # 检测 Thought thought_match re.search(rthought(.*?)/thought, content, re.DOTALL) if thought_match: thought thought_match.group(1) print(f\n\n Thought: {thought}) # 检测模型是否输出 Final Answer如果是的话直接返回 if final_answer in content: final_answer re.search(rfinal_answer(.*?)/final_answer, content, re.DOTALL) return final_answer.group(1) # 检测 Action action_match re.search(raction(.*?)/action, content, re.DOTALL) if not action_match: raise RuntimeError(模型未输出 action) action action_match.group(1) tool_name, args self.parse_action(action) print(f\n\n Action: {tool_name}({, .join(args)})) # 只有终端命令才需要询问用户其他的工具直接执行 should_continue input(f\n\n是否继续Y/N) if tool_name run_terminal_command else y if should_continue.lower() ! y: print(\n\n操作已取消。) return 操作被用户取消 try: observation self.tools[tool_name](*args) except Exception as e: observation f工具执行错误{str(e)} print(f\n\n Observation{observation}) # 将观测结果包装observation标签以 user 角色追加到上下文下一轮循环模型会读取该结果继续推理 obs_msg fobservation{observation}/observation messages.append({role: user, content: obs_msg}) def get_tool_list(self) - str: 生成工具列表字符串包含函数签名和简要说明 tool_descriptions [] for func in self.tools.values(): name func.__name__ signature str(inspect.signature(func)) doc inspect.getdoc(func) tool_descriptions.append(f- {name}{signature}: {doc}) return \n.join(tool_descriptions) def render_system_prompt(self, system_prompt_template: str) - str: 渲染系统提示模板替换变量 tool_list self.get_tool_list() # file_list , .join( # os.path.abspath(os.path.join(self.project_directory, f)) # for f in os.listdir(self.project_directory) # ) # 新建文件的保存目录 file_list os.path.abspath(self.project_directory) return Template(system_prompt_template).substitute( operating_systemself.get_operating_system_name(), tool_listtool_list, file_listfile_list ) staticmethod def get_api_key() - str: Load the API key from an environment variable. # 加载环境变量 load_dotenv() # 读取环境变量 api_key os.getenv(OPENAI_API_KEY) if not api_key: raise ValueError(未找到 OPENROUTER_API_KEY 环境变量请在 .env 文件中设置。) return api_key def call_model(self, messages): print(\n\n正在请求模型请稍等...) response self.client.chat.completions.create( modelself.model, messagesmessages, ) content response.choices[0].message.content messages.append({role: assistant, content: content}) return content def parse_action(self, code_str: str) - Tuple[str, List[str]]: match re.match(r(\w)\((.*)\), code_str, re.DOTALL) if not match: raise ValueError(Invalid function call syntax) func_name match.group(1) args_str match.group(2).strip() # 手动解析参数特别处理包含多行内容的字符串 args [] current_arg in_string False string_char None i 0 paren_depth 0 while i len(args_str): char args_str[i] if not in_string: if char in [, ]: in_string True string_char char current_arg char elif char (: paren_depth 1 current_arg char elif char ): paren_depth - 1 current_arg char elif char , and paren_depth 0: # 遇到顶层逗号结束当前参数 args.append(self._parse_single_arg(current_arg.strip())) current_arg else: current_arg char else: current_arg char if char string_char and (i 0 or args_str[i - 1] ! \\): in_string False string_char None i 1 # 添加最后一个参数 if current_arg.strip(): args.append(self._parse_single_arg(current_arg.strip())) return func_name, args def _parse_single_arg(self, arg_str: str): 解析单个参数 arg_str arg_str.strip() # 如果是字符串字面量 if (arg_str.startswith() and arg_str.endswith()) or \ (arg_str.startswith() and arg_str.endswith()): # 移除外层引号并处理转义字符 inner_str arg_str[1:-1] # 处理常见的转义字符 inner_str inner_str.replace(\\, ).replace(\\, ) inner_str inner_str.replace(\\n, \n).replace(\\t, \t) inner_str inner_str.replace(\\r, \r).replace(\\\\, \\) return inner_str # 尝试使用 ast.literal_eval 解析其他类型 try: return ast.literal_eval(arg_str) except (SyntaxError, ValueError): # 如果解析失败返回原始字符串 return arg_str def get_operating_system_name(self): os_map { Darwin: macOS, Windows: Windows, Linux: Linux } return os_map.get(platform.system(), Unknown) def read_file(file_path): 用于读取文件内容 with open(file_path, r, encodingutf-8) as f: return f.read() def write_to_file(file_path, content): 将指定内容写入指定文件 with open(file_path, w, encodingutf-8) as f: f.write(content.replace(\\n, \n)) return 写入成功 def run_terminal_command(command): 用于执行终端命令 import subprocess run_result subprocess.run(command, shellTrue, capture_outputTrue, textTrue) return 执行成功 if run_result.returncode 0 else run_result.stderr click.command() click.argument(project_directory, typeclick.Path(existsTrue, file_okayFalse, dir_okayTrue)) def main(project_directory): project_dir os.path.abspath(project_directory) tools [read_file, write_to_file, run_terminal_command] agent ReActAgent(toolstools, modelglm-5, project_directoryproject_dir) task input(请输入任务) final_answer agent.run(task) print(f\n\n✅ Final Answer{final_answer}) if __name__ __main__: main()2prompt_template.pyreact_system_prompt_template 你需要解决一个问题。为此你需要将问题分解为多个步骤。对于每个步骤首先使用 thought 思考要做什么然后使用可用工具之一决定一个 action。接着你将根据你的行动从环境/工具中收到一个 observation。持续这个思考和行动的过程直到你有足够的信息来提供 final_answer。 所有步骤请严格使用以下 XML 标签格式输出 - question 用户问题 - thought 思考 - action 采取的工具操作 - observation 工具或环境返回的结果 - final_answer 最终答案 ⸻ 例子 1: question埃菲尔铁塔有多高/question thought我需要找到埃菲尔铁塔的高度。可以使用搜索工具。/thought actionget_height(埃菲尔铁塔)/action observation埃菲尔铁塔的高度约为330米包含天线。/observation thought搜索结果显示了高度。我已经得到答案了。/thought final_answer埃菲尔铁塔的高度约为330米。/final_answer ⸻ 例子 2: question帮我找一个简单的番茄炒蛋食谱并看看家里的冰箱里有没有西红柿。/question thought这个任务分两步。第一步找到番茄炒蛋的食谱。第二步检查冰箱里是否有西红柿。我先用 find_recipe 工具找食谱。/thought actionfind_recipe(dish番茄炒蛋)/action observation简单的番茄炒蛋食谱将2个鸡蛋打散2个番茄切块。热油先炒鸡蛋盛出。再热油炒番茄至软烂加入鸡蛋放盐调味即可。/observation thought好的我已经有食谱了。食谱需要西红柿。现在我需要用 check_fridge 工具看看冰箱里有没有西红柿。/thought actioncheck_fridge(item西红柿)/action observation冰箱检查结果有3个西红柿。/observation thought我找到了食谱并且确认了冰箱里有西红柿。可以回答问题了。/thought final_answer简单的番茄炒蛋食谱是鸡蛋打散番茄切块。先炒鸡蛋再炒番茄混合后加盐调味。冰箱里有3个西红柿。/final_answer ⸻ 请严格遵守 - 你每次回答都必须包括两个标签第一个是 thought第二个是 action 或 final_answer - 输出 action 后立即停止生成等待真实的 observation擅自生成 observation 将导致错误 - 如果 action 中的某个工具参数有多行的话请使用 \n 来表示如actionwrite_to_file(/tmp/test.txt, a\nb\nc)/action - 工具参数中的文件路径请使用绝对路径不要只给出一个文件名。比如要写 write_to_file(/tmp/test.txt, 内容)而不是 write_to_file(test.txt, 内容) ⸻ 本次任务可用工具 ${tool_list} ⸻ 环境信息 操作系统${operating_system} 当前目录下文件列表${file_list} 3.env#.env OPENAI_API_KEY执行结果写一个贪吃蛇游戏使用HTMLcss和js实现代码分别放在不同的文件中Agent常见架构