手把手用LangChain实现简易AutoGPT

手把手用LangChain实现简易AutoGPT

你有没有想过，让AI不只是回答问题，而是真正“动手做事”？比如你只说一句：“帮我分析下上个月的销售数据并写个报告”，它就能自己找文件、读表格、算指标、生成文档，最后把结果交给你——整个过程几乎不需要你插手。

这听起来像是科幻片里的场景，但其实正是AutoGPT的核心理念：一个能自主思考、规划任务、调用工具、持续迭代的AI智能体。它不再被动应答，而是主动推进目标，像人类一样“思考—行动—观察—再调整”。

今天，我们就来亲手打造一个轻量级但完整的AutoGPT原型，用的是目前最流行的AI应用开发框架——LangChain。这不是简单的调API，而是一次对智能体底层机制的深度拆解和重构。

你会看到：

• 如何让大模型“有条理地思考”
• 怎么设计一套可扩展的工具系统
• ReAct（Reason + Act）循环是如何驱动自主行为的
• Prompt工程如何决定Agent的“智商上限”

准备好了吗？我们从零开始，一步步构建这个会“自己想办法”的AI助手。

项目目标：做一个能“独立完成任务”的Agent

我们的目标很明确：用户输入一句话任务，比如“分析8月销售数据并生成报告”，Agent就要能：

1. 理解任务需求（关键词识别）
2. 拆解成可执行步骤（先看有哪些文件 → 再读Excel结构 → 提取关键数据 → 写报告）
3. 自主选择并调用合适的工具
4. 根据执行结果判断下一步动作
5. 直到任务完成，输出最终成果

整个流程形成闭环：感知 → 推理 → 行动 → 反馈 → 再推理。这才是真正意义上的“智能体”。

为了支撑这些操作，我们需要给Agent配备几个基础“能力模块”：

这些工具就是它的“手脚”。没有它们，再聪明的脑子也干不了活。

架构设计：SimpleAutoGPT主类

我们定义一个主类SimpleAutoGPT，作为整个系统的入口。它负责管理LLM、工具集、记忆系统和执行流程。

from langchain.chat_models import ChatOpenAI from langchain.schema import BaseOutputParser from langchain.memory import ConversationTokenBufferMemory from typing import List, Optional from pydantic import BaseModel, Field class Action(BaseModel): name: str = Field(description="工具名称") args: dict = Field(default_factory=dict, description="工具参数") class SimpleAutoGPT: """简易版AutoGPT实现""" def __init__( self, llm, tools: List, work_dir: str, prompts_path: str, main_prompt_file: str, output_parser: BaseOutputParser, max_thought_steps: int = 10, memory_retriever=None, ): self.llm = llm self.tools = tools self.work_dir = work_dir self.prompts_path = prompts_path self.main_prompt_file = main_prompt_file self.output_parser = output_parser self.max_thought_steps = max_thought_steps self.memory_retriever = memory_retriever self.tool_map = {tool.name: tool for tool in tools}

这里的关键点是：
-llm是大脑，负责推理决策
-tools是手脚，负责实际操作
-output_parser强制模型输出结构化动作，避免胡说八道
-memory_retriever支持长期记忆，记住过去的经验

核心逻辑：ReAct循环驱动自主行为

真正的“智能”不在于一次回答得多准，而在于能否根据反馈不断调整策略。这就是ReAct 模式：Reasoning + Acting。

它的运行节奏如下：

1. 思考（Reason）：基于当前上下文，决定下一步做什么
2. 行动（Act）：调用某个工具执行具体操作
3. 观察（Observe）：记录返回结果，更新记忆
4. 重复：回到第1步，直到任务完成

这种循环模式，才是AutoGPT“自主性”的本质。

下面是run方法的具体实现：

def run(self, task_description: str, verbose: bool = False) -> str: thought_count = 0 short_term_memory = ConversationTokenBufferMemory( llm=self.llm, max_token_limit=4000 ) short_term_memory.save_context({"input": "任务开始"}, {"output": "已接收任务"}) prompt_template = PromptTemplateBuilder( self.prompts_path, self.main_prompt_file ).build( tools=self.tools, output_parser=self.output_parser ).partial( work_dir=self.work_dir, task_description=task_description ) chain = prompt_template | self.llm | StrOutputParser() long_term_memory = VectorStoreRetrieverMemory(retriever=self.memory_retriever) \ if self.memory_retriever else None reply = "" while thought_count < self.max_thought_steps: if verbose: color_print(f"\n>>> Step {thought_count} >>>", "blue") action, response = self._step( chain=chain, short_term_memory=short_term_memory, long_term_memory=long_term_memory, task_description=task_description, verbose=verbose ) if action.name == "FINISH": if verbose: color_print("\n✅ 任务完成", "green") reply = self._final_response(short_term_memory, task_description) break observation = self._exec_action(action) if verbose: color_print(f"\n🔍 观察结果:\n{observation}", "yellow") short_term_memory.save_context( {"input": response}, {"output": f"执行结果:\n{observation}"} ) thought_count += 1 if not reply: reply = "抱歉，未能完成任务，请尝试更清晰的指令。" if long_term_memory: long_term_memory.save_context( {"input": task_description}, {"output": reply} ) return reply

可以看到，这个循环非常接近人类解决问题的方式：每走一步都回头看一眼结果，再决定怎么继续。

小贴士：max_thought_steps设置为10是为了防止无限循环。实践中可以根据任务复杂度动态调整。

Prompt工程：决定Agent“智商”的关键

很多人低估了Prompt的作用，以为只是个格式模板。但在智能体系统中，Prompt就是操作系统。它决定了Agent会不会思考、怎么思考、往哪个方向走。

我们采用模块化设计，把主Prompt拆成多个子组件，便于维护和调试。

主框架：main.txt

你是强大的AI助手，可以根据任务目标自主规划并调用工具解决问题。 请严格遵循以下流程进行思考与行动。 【任务目标】 {task_description} 【工作目录】 所有相关文件位于：{work_dir} 【可用工具】 {tools} 【历史记忆】 {long_term_memory} 【当前上下文】 {short_term_memory} 【输出格式要求】 1. 先按以下结构输出你的思考过程： {thought_instructions} 2. 然后输出你要执行的动作，格式如下： {format_instructions}

思考指令：thought_instructions.txt

这是最关键的“思维引导”部分，直接决定Agent是否有条理：

概念识别： - 识别任务中的关键实体和需求，例如“销售额”、“时间范围”、“文档类型”等。 任务拆解： - 将主任务分解为若干可执行的子任务，每个子任务对应一次工具调用。 - 明确每个子任务所需的数据源和依赖关系。 反思： - 回顾之前的执行记录，判断是否有遗漏或错误。 - 是否已获得足够的信息来推进下一步？ - 上一次调用是否失败？如果是，原因是什么？是否可以换一种方式？ 推理： - 分析当前信息缺口，确定下一个最有效的动作。 - 判断是否需要先获取元信息（如列出文件、查看表头）。 计划： - 明确下一步要使用的工具及其参数。 - 若任务已完成，应选择 FINISH 工具。

你会发现，这套指令其实在模仿人类专家的工作方式：先理解问题，再制定计划，过程中不断复盘和调整。

模板加载器：自动组装Prompt链

为了让结构更灵活，我们封装了一个PromptTemplateBuilder类，支持.txt和.json混合加载：

from langchain.prompts import load_prompt from langchain_core.prompts import PipelinePromptTemplate import os def _load_file(path: str) -> str: with open(path, 'r', encoding='utf-8') as f: return f.read() class PromptTemplateBuilder: def __init__(self, prompt_path: str, prompt_file: str): self.prompt_path = prompt_path self.prompt_file = prompt_file def build(self, tools=None, output_parser=None): main_path = os.path.join(self.prompt_path, self.prompt_file) main_prompt = load_prompt(main_path) variables = main_prompt.input_variables partial_vars = {} sub_prompts = [] for var in variables: txt_path = os.path.join(self.prompt_path, f"{var}.txt") json_path = os.path.join(self.prompt_path, f"{var}.json") if os.path.exists(json_path): sub_builder = PromptTemplateBuilder(self.prompt_path, f"{var}.json") sub_template = sub_builder.build(tools, output_parser) sub_prompts.append((var, sub_template)) elif os.path.exists(txt_path): content = _load_file(txt_path) if var == "tools" and tools: from langchain.tools.render import render_text_description partial_vars[var] = render_text_description(tools) elif var == "format_instructions" and output_parser: partial_vars[var] = output_parser.get_format_instructions() else: partial_vars[var] = content if sub_prompts: main_prompt = PipelinePromptTemplate( final_prompt=main_prompt, pipeline_prompts=sub_prompts ) return main_prompt.partial(**partial_vars)

这样做的好处是，未来新增提示词时只需加个文件，无需改动代码逻辑。

实现思考步骤_step

这一步的核心是调用大模型生成结构化动作，并确保输出符合预期格式。

def _step(self, chain, short_term_memory, long_term_memory, task_description, verbose): inputs = { "short_term_memory": _format_memory(short_term_memory), "long_term_memory": _format_memory(long_term_memory) if long_term_memory else "", } response = "" for chunk in chain.stream(inputs): if verbose: color_print(chunk, "cyan", end="") response += chunk try: action = self.output_parser.parse(response) except Exception as e: action = Action(name="FINISH", args={"response": f"解析失败：{str(e)}"}) return action, response

辅助函数用于统一处理记忆内容：

def _format_memory(memory): if hasattr(memory, "load_memory_variables"): return memory.load_memory_variables({}).get("history", "") return str(memory)

使用流式输出（stream）的好处是可以实时看到Agent的“内心独白”，方便调试。

工具定义：赋予Agent动手能力

接下来我们用 LangChain 的StructuredTool定义一系列实用工具。

from langchain.tools import StructuredTool import os import pandas as pd def list_files(directory: str) -> str: """列出目录下所有文件""" try: files = os.listdir(directory) return "\n".join(files) except Exception as e: return f"错误: {e}" def inspect_excel(file_path: str, n: int = 3) -> str: """查看Excel前n行""" df = pd.read_csv(file_path) if file_path.endswith(".csv") else pd.read_excel(file_path) return f"列名: {list(df.columns)}\n前{n}行:\n{df.head(n).to_string()}" def search_web(query: str) -> str: """模拟网络搜索""" return f"[模拟搜索结果] 关于 '{query}' 的相关信息..." def generate_doc(content: str, filename: str): """生成文档""" with open(filename, 'w', encoding='utf-8') as f: f.write(content) return f"文档已保存至 {filename}"

然后封装为Tool对象：

tools = [ StructuredTool.from_function( func=list_files, name="ListDirectory", description="列出指定目录下的所有文件" ), StructuredTool.from_function( func=inspect_excel, name="InspectExcel", description="查看Excel或CSV文件的结构和部分内容" ), StructuredTool.from_function( func=search_web, name="SearchWeb", description="联网搜索相关信息" ), StructuredTool.from_function( func=generate_doc, name="GenerateDocument", description="生成一份文档" ), StructuredTool.from_function( func=lambda: None, name="FINISH", description="任务已完成，请使用此工具结束流程" ), ]

注意：FINISH工具虽然不做事，但它是一个重要的“终止信号”，告诉系统可以收工了。

行动执行_exec_action

根据模型输出的动作调用对应工具：

def _exec_action(self, action: Action) -> str: tool = self.tool_map.get(action.name) if not tool: return f"❌ 错误：未找到工具 '{action.name}'" try: result = tool.run(action.args) return str(result) except Exception as e: return f"🛠️ 工具执行出错: {type(e).__name__}: {str(e)}"

这里做了基本的异常捕获，防止某个工具崩溃导致整个流程中断。

运行示例：让Agent真正动起来

现在我们把所有模块组装起来，跑一个真实案例。

from langchain_openai import ChatOpenAI from langchain.schema.output_parser import PydanticOutputParser from langchain_community.vectorstores import Chroma from langchain.embeddings import OpenAIEmbeddings from langchain.schema import Document # 加载LLM llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.3) # 输出解析器 parser = PydanticOutputParser(pydantic_object=Action) # 向量数据库（用于长期记忆） db = Chroma.from_documents([Document(page_content="")], OpenAIEmbeddings()) retriever = db.as_retriever(k=1) # 创建Agent agent = SimpleAutoGPT( llm=llm, tools=tools, work_dir="./data", prompts_path="./prompts", main_prompt_file="main.json", output_parser=parser, max_thought_steps=8, memory_retriever=retriever ) # 交互式运行 def chat_with_agent(): while True: user_input = input("👨: ") if user_input.lower() in ["quit", "exit"]: break response = agent.run(user_input, verbose=True) print(f"🤖: {response}\n") chat_with_agent()

假设输入：

“请分析 data/2023年8月销售记录.xlsx 中的销售额，并生成一份总结报告”

典型执行路径可能是：

1. 调用ListDirectory发现文件存在
2. 使用InspectExcel查看结构，确认有“金额”字段
3. 可能通过SearchWeb获取统计方法（模拟）
4. 最终调用GenerateDocument输出报告
5. 执行FINISH结束任务

整个过程完全由Agent自主决策，无需人工干预。

常见问题与应对策略

特别提醒：如果你发现Agent总是乱调工具，大概率是Prompt不够清晰。越复杂的任务，越需要精细的思维引导。

写在最后：这只是开始

我们刚刚搭建的是一个最小可行的AutoGPT原型，但它已经具备了智能体的核心要素：

• ✅ 自主任务分解能力
• ✅ ReAct 循环机制
• ✅ 模块化Prompt工程
• ✅ 多工具集成架构
• ✅ 短期+长期记忆管理

这不仅是技术练习，更是对未来AI工作方式的一次预演。

接下来你可以继续深化：

• 加入Python REPL，让它能运行脚本分析数据
• 引入Plan-and-Execute模式，减少无效试探
• 实现失败重试与容错机制
• 使用GPT-4提升推理稳定性
• 添加可视化执行轨迹日志

完整项目代码已打包上传至CSDN资源站：

🔗 下载地址

包含：
-/prompts/：所有Prompt模板
-/data/：示例数据集
-main.py：主程序入口
-requirements.txt：依赖列表

如果你也在探索AI Agent的边界，欢迎一起交流。毕竟，未来的AI不是一个人造出来，而是一群人共同进化出来的。

我是同学小张，专注AI实战分享，下期见！