# 疯狂熬夜三个月,我终于把AI Agent做成了:这不是你以为的「套壳」
疯狂熬夜三个月,我终于把AI Agent做成了:这不是你以为的「套壳」
大家好,我是摘星,今天我们来聊聊一个让我又爱又恨的技术方向——AI Agent。
过去三个月,我几乎每天都在和AI Agent打交道。做过的项目包括:自动写周报的Agent、处理客户投诉的分类Agent、甚至还有一个帮我回复邮件的Agent(虽然后来被我关掉了,因为它总用过于热情的语气回复客户)。
踩了无数坑之后,我发现一件事:市面上90%的"AI Agent教程"都在教你调API,真正能让你理解底层逻辑的少之又少。今天这篇文章,我把三个月踩坑总结的经验全部摊开,包括技术架构、核心挑战、以及如何在实际项目中落地。
核心概念:AI Agent到底是什么?
很多人第一次听到"AI Agent"这个词,第一反应是:不就是接了个API吗?套个壳就叫Agent了?
如果你也这么想,那接下来的内容可能会颠覆你的认知。
AI Agent(人工智能智能体)的本质,是让大模型具备自主规划、工具调用、长期记忆和环境交互的能力。简单来说,传统的AI调用是"一问一答"模式——你问,它答,结束。而Agent模式下,大模型可以:
- 自主规划任务步骤:把一个复杂目标拆解成多个子任务
- 调用外部工具:搜索、计算、读写文件、执行代码
- 长期记忆:跨会话积累上下文和学习经验
- 环境交互:感知环境变化并调整行为
用一个类比来解释:如果传统AI是一个只会背书的实习生,那AI Agent就是一个能独立完成项目的项目经理。它不只是回答问题,而是能够主动发现问题、解决问题、交付结果。
技术架构:拆解AI Agent的四大核心模块
一个完整的AI Agent通常由四个核心模块组成:
┌─────────────────────────────────────────────────────────┐
│ AI Agent │
├─────────────────────────────────────────────────────────┤
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Planning │ │ Memory │ │ Tools │ │
│ │ 规划模块 │ │ 记忆模块 │ │ 工具模块 │ │
│ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │
│ │ │ │ │
│ └────────────────┼────────────────┘ │
│ │ │
│ ┌──────▼──────┐ │
│ │ Core │ │
│ │ LLM Engine │ │
│ └─────────────┘ │
└─────────────────────────────────────────────────────────┘
规划模块(Planning):负责将复杂任务拆解为可执行的步骤。常见实现包括:
- ReAct(Reasoning + Acting):让模型在推理过程中交替进行思考和行动
- CoT(Chain of Thought):通过链式思考逐步推导
- MoA(Mixture of Agents):多Agent协作分工
记忆模块(Memory):分为三种类型:
- 短期记忆(Short-term):当前会话的上下文
- 长期记忆(Long-term):跨会话积累的知识
- 工作记忆(Working):任务执行过程中的临时信息
工具模块(Tools):Agent调用外部能力的接口:
- 搜索工具(Web Search)
- 代码执行(Code Interpreter)
- 文件读写(File System)
- API调用(External APIs)
核心引擎(Core LLM):负责理解、推理、决策的大脑。
深度拆解:ReAct范式与AutoGPT的核心原理
理解了架构之后,我们来看两个最核心的技术实现:ReAct范式和AutoGPT。
ReAct:让大模型学会"边想边做"
ReAct(Reasoning + Acting)是我在实战中使用最频繁的框架。它的核心思想是:让大模型在推理过程中交替进行"思考"和"行动"。
传统的CoT(Chain of Thought)只让模型"想",不让它"做"。而ReAct让模型在每个推理步骤后追加一个行动,通过观察行动结果来调整下一步推理。
# ReAct核心逻辑伪代码
def react_loop(query, agent):
history = []
observation = ""
while not finished:
# 思考步骤:基于当前状态生成推理
thought = agent.think(query, history, observation)
# 行动步骤:决定下一步动作
action = agent.decide_action(thought)
# 执行动作并获取结果
observation = agent.execute(action)
# 记录历史
history.append({
"thought": thought,
"action": action,
"observation": observation
})
# 检查是否达成目标
if agent.is_goal_achieved(observation):
finished = True
return agent.generate_response(history)
关键在于decide_action函数。它需要让模型学会:
- 当前状态是什么?
- 有哪些可用工具?
- 下一步最合适的行动是什么?
这需要精心设计的prompt工程。让我分享一个我常用的ReAct prompt模板:
SYSTEM_PROMPT = """你是一个专业的AI助手,擅长将复杂任务拆解并逐步执行。
在每个步骤中,你需要:
1. 思考(THINK):分析当前情况,明确已知信息和目标
2. 行动(ACT):从以下可选行动中选择最合适的:
- search[关键词]:使用搜索引擎查找信息
- calculate[表达式]:进行数学计算
- read[文件路径]:读取本地文件内容
- write[内容]:写入文件
- execute[代码]:执行代码
- finish:任务完成,生成最终回答
每次只选择一个行动。行动执行后,你会获得观察结果(OBSERVATION)。
基于观察结果,继续下一步的思考和行动。
示例:
用户: 北京今天气温多少度?
THINK: 用户想知道北京今天的气温。我需要搜索最新的北京天气信息。
ACT: search[北京今天气温]
OBSERVATION: 搜索结果显示北京今天气温15-23度,多云。
THINK: 获得了北京今天的气温信息,可以回答用户了。
ACT: finish[北京今天气温15-23度,多云]"""
def think(self, query, history, observation):
prompt = f"""当前任务:{query}
历史记录:
{self._format_history(history)}
最新观察:{observation}
请进行THINK和ACT:
"""
response = self.llm.generate(prompt)
return response
AutoGPT:从"自动"到"自主"的跨越
AutoGPT是2023年最具影响力的开源Agent项目之一。它的核心创新在于引入了自主目标分解和递归任务执行。
传统的ReAct需要人类指定每个步骤,而AutoGPT可以让模型自己生成子目标,形成一个任务树:
class AutoGPT:
def __init__(self, model, tools):
self.model = model
self.tools = tools
selfgoals = []
def run(self, initial_goal):
self.goals = [Task(initial_goal, priority=1)]
completed_tasks = []
while self.goals:
# 按优先级排序
self.goals.sort(key=lambda x: x.priority)
# 取出最高优先级任务
current_task = self.goals.pop(0)
# 执行任务
result = self.execute_task(current_task)
# 分析结果,生成新子任务
subtasks = self.analyze_and_decompose(result)
# 将子任务加入队列
for subtask in subtasks:
self.goals.append(subtask)
# 标记完成
completed_tasks.append(current_task)
return self.generate_final_output(completed_tasks)
def execute_task(self, task):
"""执行单个任务,支持递归分解"""
prompt = f"""任务:{task.description}
目标:{task.goal}
请执行任务。如果任务复杂,可以分解为多个子任务。
子任务格式:subtask[描述]|priority[优先级]
执行结果:"""
response = self.model.generate(prompt)
# 解析子任务
if "subtask[" in response:
subtasks = self.parse_subtasks(response)
return {"status": "decomposed", "subtasks": subtasks}
else:
return {"status": "completed", "result": response}
AutoGPT的另一个关键特性是目标回溯:如果某个子任务失败,它会自动分析原因并尝试替代方案。这使得整个系统具有很强的容错能力。
实战:从零构建一个自动写周报的Agent
光说不练假把式。接下来,我带大家从零构建一个自动写周报的Agent。这个Agent需要:
- 读取本周的工作记录(从钉钉/飞书导出)
- 分析并分类工作内容
- 生成结构化的周报
第一步:定义工具集
from dataclasses import dataclass
from typing import List, Dict, Callable
@dataclass
class Tool:
name: str
description: str
function: Callable
parameters: Dict
class ToolRegistry:
def __init__(self):
self.tools: List[Tool] = []
def register(self, name: str, description: str, func: Callable,
params: List[str]):
self.tools.append(Tool(name, description, func, params))
def get_tool(self, name: str) -> Tool:
for tool in self.tools:
if tool.name == name:
return tool
raise ValueError(f"Tool {name} not found")
def list_tools(self) -> str:
return "n".join([
f"- {t.name}: {t.description} (参数: {', '.join(t.parameters)})"
for t in self.tools
])
# 注册工具
registry = ToolRegistry()
registry.register(
name="read_file",
description="读取本地文件内容",
func=read_file_content,
params=["file_path"]
)
registry.register(
name="search_web",
description="搜索互联网获取信息",
func=search_internet,
params=["query"]
)
registry.register(
name="save_file",
description="保存内容到本地文件",
func=save_to_file,
params=["content", "file_path"]
)
第二步:实现Agent核心
import json
from typing import List, Dict, Tuple
class WeeklyReportAgent:
def __init__(self, llm, tool_registry: ToolRegistry):
self.llm = llm
self.tools = tool_registry
self.conversation_history = []
self.short_term_memory = []
def system_prompt(self) -> str:
tools_desc = self.tools.list_tools()
return f"""你是一个专业的周报撰写助手。
## 你的能力
你可以通过以下工具完成复杂任务:
{tools_desc}
## 工作流程
1. 首先收集信息:读取工作记录文件
2. 分析整理:分类本周工作内容
3. 生成周报:按照标准格式撰写
## 输出格式
周报需要包含:
- 本周完成(3-5条)
- 进行中(1-3条)
- 下周计划(3-5条)
- 心得总结(1条)
## 注意事项
- 数据必须来自实际文件,不要编造
- 语言简洁专业,突出价值
- 遇到不确定的信息,主动搜索核实"""
def think(self, user_input: str) -> str:
"""生成思考和行动"""
context = self.build_context()
prompt = f"""{self.system_prompt()}
当前任务:{user_input}{context}
请进行THINK,分析当前状态,然后决定ACT(使用工具或完成)。
"""
response = self.llm.generate(prompt)
self.conversation_history.append({
"role": "assistant",
"content": response
})
return response
def act(self, thought: str) -> Tuple[str, any]:
"""执行行动"""
# 解析行动
if "read_file" in thought:
# 提取文件路径
import re
match = re.search(r'read_file[([^]]+)]', thought)
if match:
file_path = match.group(1)
result = self.tools.get_tool("read_file").function(file_path)
return f"已读取文件 {file_path}", result
elif "save_file" in thought:
import re
match = re.search(r'save_file[([^]]+)]', thought)
if match:
content = match.group(1)
# 实际实现需要从上下文中获取内容
return "保存文件", "success"
elif "finish" in thought or "完成" in thought:
return "FINISH", None
return "继续执行", None
def run(self, initial_task: str, max_iterations: int = 10):
"""运行Agent主循环"""
current_input = initial_task
iteration = 0
while iteration < max_iterations:
# 思考
thought = self.think(current_input)
# 提取思考内容
think_content = self.extract_think(thought)
# 执行行动
action_result = self.act(thought)
# 记录到记忆
self.short_term_memory.append({
"iteration": iteration,
"thought": think_content,
"action_result": action_result
})
# 检查是否完成
if action_result[0] == "FINISH":
return self.generate_final_report()
# 更新输入
current_input = f"观察结果:{action_result[0]}"
iteration += 1
return "执行超时,请检查任务设置"
def extract_think(self, response: str) -> str:
"""提取THINK部分"""
if "THINK:" in response:
return response.split("THINK:")[1].split("ACT:")[0]
return response
def generate_final_report(self) -> str:
"""生成最终周报"""
# 汇总所有记忆
context = "n".join([
f"步骤{m['iteration']}: {m['thought']}"
for m in self.short_term_memory
])
prompt = f"""基于以下执行记录,生成最终周报:
{context}
请按照标准格式输出周报内容。
"""
return self.llm.generate(prompt)
第三步:完整运行示例
# 初始化
llm = OpenAIClient(model="gpt-4o")
registry = ToolRegistry()
register_work_tools(registry)
agent = WeeklyReportAgent(llm, registry)
# 读取工作记录文件
# 假设工作记录在 weekly_logs.txt 中
task = "请读取 weekly_logs.txt 文件,分析本周工作内容,生成周报"
# 运行Agent
report = agent.run(task)
print(report)
实际运行效果会像这样:
=== Agent 执行日志 ===
步骤 0: THINK - 用户需要生成周报。我需要先读取工作记录文件。
步骤 0: ACT - read_file[weekly_logs.txt]
观察结果:已读取文件,包含15条工作记录
步骤 1: THINK - 已获得工作记录,需要分析并分类
步骤 1: ACT - 分析完成,识别出:项目A开发、代码Review、周会参与等
步骤 1: ACT - finish[生成周报...]
=== 最终周报 ===
## 本周完成
1. 完成用户模块API开发(3天)
2. 完成代码Review 8次
3. 参加产品评审会1次
## 进行中
1. 性能优化方案设计
## 下周计划
1. 上线用户模块
2. 开始订单模块开发
## 心得
本周重点完成了核心功能开发,对模块化设计有了更深的理解。
技术对比:主流Agent框架横评
市场上主流的Agent框架有很多,我做了一个横向对比:
| 框架 | 核心特性 | 适合场景 | 学习成本 |
|---|---|---|---|
| LangChain | 功能全面,生态丰富 | 复杂业务流程 | 中等 |
| AutoGPT | 自主性强,递归执行 | 探索性任务 | 较高 |
| BabyAGI | 轻量级,易上手 | 快速原型 | 低 |
| MetaGPT | 多Agent协作 | 软件开发 | 中等 |
| CrewAI | 多Agent编排 | 企业应用 | 中等 |
我个人的经验是:
- 快速原型:用BabyAGI,它的概念简单,上手快
- 复杂业务:用LangChain,它的灵活性最强
- 软件开发:用MetaGPT,它的多Agent协作模式很适合代码生成
核心挑战:为什么你的Agent总是"跑偏"?
做了三个月Agent,我踩过最多的坑就是:Agent跑到一半开始"幻觉",偏离目标。
这里总结几个核心问题和解决方案:
问题一:上下文窗口耗尽
当任务复杂、步骤多的时候,上下文会快速膨胀,最终超过模型的上下文窗口限制。
解决方案:实施记忆压缩和任务摘要
def compress_context(memory: List[Dict], max_items: int = 10) -> List[Dict]:
"""压缩记忆,保留关键信息"""
if len(memory) <= max_items:
return memory
# 保留最近的记忆
recent = memory[-max_items//2:]
# 对早期记忆进行摘要
early = memory[:-max_items//2]
summary = summarize_memories(early)
return [
{"type": "summary", "content": summary},
*recent
]
def summarize_memories(memories: List[Dict]) -> str:
"""将多条记忆压缩为摘要"""
prompt = f"""将以下工作记录压缩为一个简洁的摘要:
{chr(10).join([m['thought'] for m in memories])}
要求:
- 保留关键决策和行动
- 移除重复信息
- 长度控制在100字以内"""
return llm.generate(prompt)
问题二:工具调用失败
网络不稳定、文件不存在、API超时等问题都可能导致工具调用失败。
解决方案:实施重试机制和降级策略
import time
from functools import wraps
def retry_with_backoff(max_retries=3, initial_delay=1):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
delay = initial_delay
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_retries - 1:
raise
time.sleep(delay)
delay *= 2 # 指数退避
return wrapper
return decorator
@retry_with_backoff(max_retries=3)
def robust_execute(tool_name: str, params: Dict):
try:
tool = registry.get_tool(tool_name)
return tool.function(**params)
except ToolExecutionError as e:
# 降级策略:使用替代方案
if tool_name == "search_web":
return {"fallback": True, "content": "搜索不可用,请手动提供信息"}
raise
问题三:目标偏离
Agent在多步骤执行后,容易忘记最初的 목표。
解决方案:实施目标提醒和中间检查点
def check_alignment(current_step: str, original_goal: str) -> bool:
"""检查当前步骤是否与原目标对齐"""
prompt = f"""当前执行步骤:{current_step}
原始目标:{original_goal}
判断执行是否偏离目标。如果偏离,返回原因和建议调整。
如果正常,返回"ALIGNED"。"""
response = llm.generate(prompt)
if "ALIGNED" in response:
return True
else:
return False
# 在每个主要步骤后进行检查
def execute_with_checkpoint(agent, step: str, goal: str):
result = execute_step(step)
if not check_alignment(result, goal):
# 重新对齐
correction = realign_goal(goal, result)
return correction
return result
架构图:复杂Agent系统的完整设计
下面是一个完整的复杂Agent系统架构图,展示了从用户输入到最终输出的完整流程:
工具层
记忆模块
问答类
任务类
创作类
是
成功
失败
否
用户输入
意图识别层
意图类型判断
问答Agent
任务规划层
创作Agent
任务分解器
子任务队列
队列非空?
执行子任务
结果评估
标记完成
重新规划
结果汇总
输出层
短期记忆
上下文管理
长期记忆
工作记忆
搜索引擎
工具调度器
代码执行
文件操作
API调用
用户输出
说明:这是一个典型的复杂Agent系统架构。用户输入首先经过意图识别层,判断是问答、任务还是创作类需求。任务类需求会进入任务规划层,通过任务分解器生成子任务队列,由执行器逐个执行并评估结果。记忆模块管理三类记忆,为各层提供上下文支持。工具层提供搜索、代码执行、文件操作等能力。
未来展望:Agent将如何改变我们的工作方式?
做了三个月的Agent开发,我的感受是:我们正处在一个转折点。
五年前,大家讨论的是"AI能不能做某件事"。现在讨论的是"AI能不能自主做好某件事"。再过五年,讨论的将是"AI能不能自主发现并完成某件事"。
这个演进路径正在加速。
从我自己的实践来看,当前Agent最成熟的落地场景是:
- 信息收集与整理:自动搜索、汇总、分类信息
- 流程自动化:把重复性工作流程化
- 代码辅助:自动生成、Review、调试
- 文档处理:总结、翻译、改写
而最难突破的场景是:
- 跨组织协调:涉及多方利益和复杂决策
- 长期规划:需要数周甚至数月的目标管理
- 创意工作:真正具有原创性的内容创作
未来五年,我认为Agent会在以下方面有突破:
- 多模态Agent:不只是文字,还包括图像、视频、音频的统一理解
- 自主学习:从用户反馈中持续优化,不再"每次都是新的开始"
- 群体智能:多个Agent协作完成复杂项目
写给正在入局AI的你
写了这么多,最后想说几句掏心窝的话。
这三个月的Agent开发经历,让我深刻认识到一件事:AI Agent不是"调API套壳"那么简单。它涉及到大模型推理、提示工程、系统架构、错误处理等多个维度的综合能力。
但同时,它也没有很多人想象的那么难。关键在于:
- 从小做起:先做一个能跑通的简单Agent,再逐步迭代
- 聚焦场景:不要试图做一个"万能Agent",找到最适合的场景切入
- 重视评估:Agent的输出质量参差不齐,必须建立有效的评估体系
最最重要的是:不要害怕失败。我的第一个Agent跑了三天,最后产出的周报简直不忍直视。但正是这些失败,让我真正理解了Agent的边界在哪里。
好了,今天的分享就到这里。如果你在AI Agent开发中有什么问题,或者想了解某个具体方向的实现细节,欢迎在评论区留言。
我是摘星,我们下期见。
参考链接
- LangChain Official Documentation: https://www.langchain.com/
- AutoGPT GitHub Repository: https://github.com/Significant-Gravitas/AutoGPT
- ReAct Paper: “ReAct: Synergizing Reasoning and Acting in Language Models”
- MetaGPT Official Site: https://arxiv.org/abs/2308.00352
注:因网络原因无法获取实时搜索结果,部分技术细节基于近期公开资料整理,如有疏漏欢迎指正。