Agent原理 实现（快速上手）

Agent原理

核心定义

1）LLM：被动应答，你问一句它答一句，不会主动做任何操作，不会主动调取外部资源。

2）Agent：主动闭环：思考Thought → 调用工具Action → 获取结果Observation，自主循环直到任务完成，不用人类分步指挥。

经典架构

ReAct 架构（最普及）Action → Observation 循环

1）Thought：大模型思考下一步需要做什么

2）Action：输出工具调用指令（调用代码、接口、检索、数据库）

3）Observation：拿到工具返回返回工具执行结果，把结果添加到上下文，输入大模型进行下一轮

循环往复，直到模型判断任务完成，输出最终答案。

注：Agent执行过程中，每次大模型输入包含：用户任务 + 历史上下文 + 工具函数执行结果。

优点：实现简单、动态适配未知场景；不需要预先写完整计划。

缺点：长任务容易跑偏、陷入循环；复杂多步骤任务容易遗漏步骤。

适用场景：通用助手、检索增强 RAG Agent、代码执行 Agent（Claude Code、Cursor 底层基础范式）

Plan-and-Solve / Plan & Execute（规划执行架构）

1）Planning 阶段：LLM 一次性生成完整任务步骤清单

2）Executor 阶段：按照清单依次执行工具调用

每一步执行结果反馈，可动态微调计划。

优点：长流程任务逻辑性更强，不容易丢步骤；便于人工审核计划。

缺点：任务未知、环境多变时，初始计划容易失效，灵活性不如 ReAct。

适用场景：标准化工作流、数据分析流水线、自动化运维任务。

Self-Ask（自我提问架构）

模型不断向自己提出子问题，通过工具回答子问题，最终汇总得到主问题答案。

示例：“武汉 7 月平均气温？” 自提问 1：武汉 7 月历史气温数据在哪里获取？→ 调用搜索 自提问 2：如何计算月平均值？

适用场景：信息搜集类、多事实推理任务；常作为 ReAct 内部增强策略。

Agent框架

LangGraph

LangGraph团队开发的一款开源 AI Agent 框架，主要用来帮开发者搭建、部署和管理那些比较复杂的生成式 AI Agent 工作流。

通信机制：Agent 之间不直接通信，所有读写操作围绕【全局共享 State】（Agent间不需要大量交互对话，Token开销低）

1）优点

2）缺点

AutoGen

AutoGen 是微软开发的开源多智能体 AI 框架‌，用于构建 AI 代理应用，支持多个智能体通过对话协作完成复杂任务。

通信机制：依靠 Manager/GroupChatManager 做消息转发，Agent 拥有独立上下文，无统一共享状态容器。（Agent间需要大量交互对话，Token开销高）

1）优点

2）缺点

单Agent实现

代码结构

代码目录

agent_demo │ ├─agent.py ├─prompt_template.py └─.env

核心代码

1）agent.py

import 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_url="https://kspmas.ksyun.com/v1/", api_key=ReActAgent.get_api_key(), ) def run(self, user_input: str): # 上下文消息列表，完整保存每一轮对话所有内容，每次对话都会把整个消息列表传给大模型 messages = [ {"role": "system", "content": self.render_system_prompt(react_system_prompt_template)}, {"role": "user", "content": f"<question>{user_input}</question>"} ] # 无限循环，持续执行ReAct思考 while True: # 请求模型，获取大模型返回文本，同时自动把模型回复存入messages content = self.call_model(messages) # 检测 Thought thought_match = re.search(r"<thought>(.*?)</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(r"<final_answer>(.*?)</final_answer>", content, re.DOTALL) return final_answer.group(1) # 检测 Action action_match = re.search(r"<action>(.*?)</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 = f"<observation>{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_system=self.get_operating_system_name(), tool_list=tool_list, file_list=file_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( model=self.model, messages=messages, ) 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", encoding="utf-8") as f: return f.read() def write_to_file(file_path, content): """将指定内容写入指定文件""" with open(file_path, "w", encoding="utf-8") as f: f.write(content.replace("\\n", "\n")) return "写入成功" def run_terminal_command(command): """用于执行终端命令""" import subprocess run_result = subprocess.run(command, shell=True, capture_output=True, text=True) return "执行成功" if run_result.returncode == 0 else run_result.stderr @click.command() @click.argument('project_directory', type=click.Path(exists=True, file_okay=False, dir_okay=True)) def main(project_directory): project_dir = os.path.abspath(project_directory) tools = [read_file, write_to_file, run_terminal_command] agent = ReActAgent(tools=tools, model="glm-5", project_directory=project_dir) task = input("请输入任务：") final_answer = agent.run(task) print(f"\n\n✅ Final Answer：{final_answer}") if __name__ == "__main__": main()

2）prompt_template.py

react_system_prompt_template = """ 你需要解决一个问题。为此，你需要将问题分解为多个步骤。对于每个步骤，首先使用 <thought> 思考要做什么，然后使用可用工具之一决定一个 <action>。接着，你将根据你的行动从环境/工具中收到一个 <observation>。持续这个思考和行动的过程，直到你有足够的信息来提供 <final_answer>。 所有步骤请严格使用以下 XML 标签格式输出： - <question> 用户问题 - <thought> 思考 - <action> 采取的工具操作 - <observation> 工具或环境返回的结果 - <final_answer> 最终答案 ⸻ 例子 1: <question>埃菲尔铁塔有多高？</question> <thought>我需要找到埃菲尔铁塔的高度。可以使用搜索工具。</thought> <action>get_height("埃菲尔铁塔")</action> <observation>埃菲尔铁塔的高度约为330米（包含天线）。</observation> <thought>搜索结果显示了高度。我已经得到答案了。</thought> <final_answer>埃菲尔铁塔的高度约为330米。</final_answer> ⸻ 例子 2: <question>帮我找一个简单的番茄炒蛋食谱，并看看家里的冰箱里有没有西红柿。</question> <thought>这个任务分两步。第一步，找到番茄炒蛋的食谱。第二步，检查冰箱里是否有西红柿。我先用 find_recipe 工具找食谱。</thought> <action>find_recipe(dish="番茄炒蛋")</action> <observation>简单的番茄炒蛋食谱：将2个鸡蛋打散，2个番茄切块。热油，先炒鸡蛋，盛出。再热油，炒番茄至软烂，加入鸡蛋，放盐调味即可。</observation> <thought>好的，我已经有食谱了。食谱需要西红柿。现在我需要用 check_fridge 工具看看冰箱里有没有西红柿。</thought> <action>check_fridge(item="西红柿")</action> <observation>冰箱检查结果：有3个西红柿。</observation> <thought>我找到了食谱，并且确认了冰箱里有西红柿。可以回答问题了。</thought> <final_answer>简单的番茄炒蛋食谱是：鸡蛋打散，番茄切块。先炒鸡蛋，再炒番茄，混合后加盐调味。冰箱里有3个西红柿。</final_answer> ⸻ 请严格遵守： - 你每次回答都必须包括两个标签，第一个是 <thought>，第二个是 <action> 或 <final_answer> - 输出 <action> 后立即停止生成，等待真实的 <observation>，擅自生成 <observation> 将导致错误 - 如果 <action> 中的某个工具参数有多行的话，请使用 \n 来表示，如：<action>write_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=

执行结果

写一个贪吃蛇游戏，使用HTML，css和js实现，代码分别放在不同的文件中

Agent常见架构