1. 项目概述从“Agen”看个人化AI代理的构建思路最近在GitHub上看到一个名为“Agen”的项目作者是Anjuan555。这个项目名本身就很值得玩味——“Agen”很容易让人联想到“Agent”代理但又少了一个“t”。这或许暗示着它并非一个功能完备的通用代理框架而更像是一个起点、一个原型或者一个高度个人化的实现。作为一名长期关注自动化工具和效率提升的从业者我本能地对这类项目产生了兴趣。它不像那些动辄宣称要颠覆行业的明星开源项目更像是一个技术爱好者为了解决自己日常工作中的特定痛点而动手搭建的一个“小工具”。这类项目往往更接地气其设计思路和实现细节对于我们理解如何将大语言模型LLM真正落地到具体、微小的场景中有着独特的参考价值。简单来说Agen项目探索的是如何利用现有的大语言模型API比如OpenAI的GPT系列构建一个能够执行特定、连贯任务的“代理”。这里的“代理”不是指那种拥有长期记忆、能自主规划复杂任务的强人工智能体而更像是一个被精心设计好流程的“自动化脚本增强版”。它接收一个目标然后按照预设或动态生成的步骤调用不同的工具可能是搜索、读写文件、调用其他API去逐步完成。这个过程的“大脑”就是大语言模型它负责理解任务、分解步骤、判断下一步该做什么。Agen的价值在于它提供了一个非常轻量级的脚手架让你可以快速试验“LLM工具”这种模式解决那些规则模糊但又有一定模式的重复性工作。这个项目适合谁呢我认为有三类人可能会从中受益。第一类是希望深入理解AI代理底层工作机制的开发者通过阅读和修改这个相对简洁的代码你能清晰地看到提示词工程、工具调用、状态管理的整个链条是如何串联起来的。第二类是有明确自动化需求但觉得现有自动化平台如Zapier, Make不够灵活或希望将AI深度集成到工作流中的技术型用户。第三类则是像我一样的“工具控”享受亲手打造并优化一个能为自己所用的智能助手的过程哪怕它最初的功能非常单一。接下来我将深入拆解这个项目的设计思路、核心实现并分享如何基于它构建一个实用的个人自动化代理。2. 核心架构与设计哲学解析2.1 轻量级与模块化为什么“小”即是美打开Agen的代码仓库第一个直观感受就是其代码结构的简洁。它没有追求大而全的框架式设计没有复杂的依赖注入、也没有抽象出多层继承体系。这种“轻量级”的选择恰恰是其核心设计哲学的体现快速验证想法降低上手门槛。在AI代理领域很多概念尚未定型过早进行过度设计反而会束缚手脚。Agen选择将核心功能——与大语言模型的交互、工具的执行、任务状态的推进——封装成几个清晰的函数或类它们之间的耦合度很低。这种模块化带来的最大好处是可插拔性。例如如果你想更换底层的LLM提供商从OpenAI切换到Claude或国内的其他模型你通常只需要修改模型调用那一小部分代码而任务逻辑、工具定义等上层结构几乎不用动。同样如果你想增加一个新的工具比如让代理能够发送邮件你只需要按照已有的模式定义一个发送邮件的函数并将其注册到工具列表中即可。这种设计鼓励实验和迭代你可以从一个最简单的“文件总结代理”开始逐步为它添加网络搜索、数据提取、内容改写等能力每一步的改动都局限在很小的范围内风险可控。注意这种轻量级设计也意味着它缺乏企业级应用所需的一些特性如分布式任务队列、完善的错误恢复机制、细粒度的权限控制等。因此它更适用于个人或小团队内的自动化场景作为生产力工具而非核心业务系统。2.2 基于LLM的任务分解与调度引擎Agen的核心“智能”来源于大语言模型对任务的分解与调度能力。这并不是一个复杂的规划算法其本质是一个精心设计的提示词Prompt循环。工作流程通常如下目标输入用户给出一个自然语言描述的目标例如“帮我总结今天项目会议记录中的行动项并发送给相关同事”。初始规划Agen会将这个目标连同可用的工具列表如read_file,search_web,send_email一起构成一个提示词发送给LLM。提示词会要求LLM将大目标分解成一系列具体的、可执行的步骤。例如步骤1读取meeting_notes.txt文件步骤2提取所有以“Action:”开头的句子步骤3整理成表格步骤4获取相关同事的邮箱列表步骤5起草邮件正文步骤6发送邮件。步骤执行与循环Agen拿到LLM输出的步骤列表后开始执行第一步。执行完成后它将当前状态已完成的步骤、结果、剩余目标再次组合成提示词询问LLM“下一步该做什么”。LLM根据当前状态决定是继续执行下一个既定步骤还是需要调整计划比如发现文件不存在需要先搜索文件然后输出下一个动作指令。终止判断循环持续进行直到LLM判断目标已达成或遇到无法克服的障碍输出“任务完成”或“任务失败”的信号。这个流程的关键在于状态提示词的设计。你需要让LLM清晰地知道它已经做了什么、得到了什么结果、还有什么没做、以及它能用什么工具。Agen的代码中通常会有一个_build_agent_prompt之类的函数这里就是核心战场。你需要反复调试这个提示词以确保LLM能做出合理的决策。例如在提示词中明确要求LLM“在决定下一步时请严格基于当前已知信息不要假设未获取的数据”可以有效减少LLM的“幻觉”避免它跳过一个未执行的搜索步骤而去直接撰写需要搜索结果的报告。2.3 工具系统的设计与扩展实践工具Tools是代理的手和脚。Agen项目中的工具本质上就是Python函数。一个典型的工具定义包含三部分函数实现、自然语言描述、参数模式。例如一个网络搜索工具可能这样定义def search_web(query: str) - str: 使用搜索引擎查询信息。 # 实际调用SerpAPI或SearXNG等搜索API的代码 results call_search_api(query) return results # 在代理中注册工具 agent.register_tool({ name: search_web, description: 当需要获取最新的、未知的或实时信息时使用此工具。输入一个搜索查询字符串。, function: search_web, parameters: {query: {type: string, description: 搜索关键词}} })描述Description是工具的灵魂。LLM并不理解你的代码它完全依靠这段自然语言描述来决定在什么情况下调用这个工具以及如何调用。因此描述必须精确、无歧义并说明使用场景和输入格式。比如“读写文件”这个描述就太模糊了更好的描述是“读取指定路径的文本文件内容。输入参数应为文件的绝对路径字符串。适用于需要获取本地文档信息的场景。”扩展工具是发挥代理威力的关键。除了文件操作、网络搜索、HTTP请求这些通用工具你可以针对特定领域创建高度定制化的工具。例如如果你是电商运营可以创建一个“查询商品库存”的工具它内部调用公司内部的库存管理系统API如果你是研究人员可以创建一个“在arXiv上搜索论文”的工具。将这些工具注册给代理后你就可以用自然语言指挥它“查一下上个月销量Top 10的商品当前库存如果任何一款库存低于安全水位就发邮件提醒采购部。”代理会自动规划先调用某个内部API获取商品列表再循环调用库存查询工具最后判断并调用邮件发送工具。实操心得在定义工具时务必让函数具有鲁棒性和明确的错误处理。因为LLM可能会生成不合法的参数比如一个不存在的文件路径。你的工具函数应该能捕获异常并返回一个结构化的错误信息如“错误文件/xxx/yyy.txt不存在”而不是直接抛出异常导致整个代理崩溃。这个错误信息会被反馈给LLM它就有可能调整计划比如改为先搜索这个文件。3. 从零构建一个个人邮件处理代理为了彻底理解Agen这类项目的精髓最好的方式就是动手构建一个。下面我将以创建一个“智能邮件处理代理”为例展示从环境搭建到核心功能实现的完整过程。这个代理的目标是定期检查邮箱根据邮件内容自动分类、总结并对特定类型的邮件如会议邀请、待办事项进行预处理。3.1 环境准备与基础框架搭建首先我们需要一个干净的Python环境。建议使用venv或conda创建独立环境。# 创建并激活虚拟环境 python -m venv agen_venv source agen_venv/bin/activate # Linux/Mac # agen_venv\Scripts\activate # Windows # 安装核心依赖 pip install openai # 或其他LLM SDK如anthropic, qianfan等 pip install python-dotenv # 用于管理API密钥接下来我们创建项目的基本结构。虽然可以直接借鉴Agen的源码但为了更透彻的理解我们尝试自己组织核心模块。my_email_agent/ ├── .env # 存储API密钥等敏感信息 ├── main.py # 主程序入口 ├── agent_core.py # 代理核心逻辑状态机、提示词构建 ├── tools.py # 所有工具函数的定义 ├── email_handler.py # 邮件相关的专用工具和逻辑 └── config.py # 配置文件在agent_core.py中我们构建最简化的代理循环骨架# agent_core.py import openai import json from typing import List, Dict, Any class SimpleAgent: def __init__(self, model: str gpt-4, tools: List[Dict] None): self.model model self.tools tools or [] self.conversation_history [] # 记录与LLM的交互历史 def run(self, initial_goal: str) - str: 运行代理直到任务完成或失败。 current_state {goal: initial_goal, completed_steps: [], findings: } for _ in range(10): # 设置最大步数防止无限循环 # 1. 构建提示词 prompt self._build_prompt(current_state) self.conversation_history.append({role: user, content: prompt}) # 2. 调用LLM获取下一步指令 response self._call_llm(self.conversation_history) self.conversation_history.append({role: assistant, content: response}) # 3. 解析LLM的响应期望是JSON格式包含action和parameters try: instruction json.loads(response) action instruction.get(action) # 例如 call_tool params instruction.get(parameters, {}) except json.JSONDecodeError: # 如果LLM没有返回JSON可能是最终答复或错误 if 任务完成 in response or final answer in response.lower(): return f任务成功结束。最终结果{response} current_state[findings] f\nLLM返回了非JSON响应{response} continue # 4. 执行动作这里以调用工具为例 if action call_tool: tool_name params.get(tool_name) tool_args params.get(arguments, {}) tool_func self._get_tool(tool_name) if tool_func: result tool_func(**tool_args) current_state[findings] f\n执行工具[{tool_name}]结果{result} current_state[completed_steps].append(f调用工具 {tool_name}) else: current_state[findings] f\n错误未找到工具 {tool_name} elif action finalize: return f任务完成。总结{params.get(summary, )} else: current_state[findings] f\n未知指令{action} # 5. 检查目标是否达成简化逻辑 # 在实际应用中这里可以加入更复杂的判断逻辑 return 任务因达到最大步数而终止。 def _build_prompt(self, state: Dict) - str: 构建当前状态下的提示词。这是核心中的核心。 tools_desc \n.join([f- {t[name]}: {t[description]} for t in self.tools]) prompt f 你是一个智能助理。你的目标是{state[goal]}。 到目前为止你已经完成了以下步骤 {chr(10).join(state[completed_steps]) if state[completed_steps] else 暂无。} 你目前掌握的信息和发现 {state[findings] if state[findings] else 暂无。} 你可以使用以下工具 {tools_desc} 请根据当前目标和已有信息决定下一步做什么。你必须以严格的JSON格式回复格式如下 {{action: call_tool, parameters: {{tool_name: 工具名, arguments: {{...}}}}}} 或者如果你认为目标已达成可以回复 {{action: finalize, parameters: {{summary: 任务总结}}}} 请只输出JSON不要有其他任何内容。 return prompt def _call_llm(self, messages: List[Dict]) - str: 调用OpenAI API示例。实际使用时请处理错误和重试。 client openai.OpenAI(api_keyos.getenv(OPENAI_API_KEY)) response client.chat.completions.create( modelself.model, messagesmessages, temperature0.1, # 低温度保证输出稳定适合结构化指令 ) return response.choices[0].message.content def _get_tool(self, name: str): 根据名称查找工具函数。 for tool in self.tools: if tool[name] name: return tool[function] return None这个骨架虽然简单但包含了代理最核心的循环构建提示 - LLM决策 - 执行动作 - 更新状态。_build_prompt函数是控制LLM行为的关键你需要不断优化它来提升代理的可靠性。3.2 邮件工具集的具体实现有了核心框架接下来实现邮件处理相关的工具。我们在tools.py和email_handler.py中编写。首先我们需要安全地读取邮箱。这里使用imaplib和email标准库。为了安全邮箱凭证存放在.env文件中。# email_handler.py import imaplib import email from email.header import decode_header import os from dotenv import load_dotenv import time load_dotenv() class EmailClient: def __init__(self): self.imap_server os.getenv(IMAP_SERVER, imap.gmail.com) self.email_address os.getenv(EMAIL_ADDRESS) self.password os.getenv(EMAIL_PASSWORD) # 对于Gmail建议使用应用专用密码 self.connection None def connect(self): 连接到IMAP服务器。 try: self.connection imaplib.IMAP4_SSL(self.imap_server) self.connection.login(self.email_address, self.password) print(邮箱连接成功) return True except Exception as e: print(f邮箱连接失败: {e}) return False def fetch_unread_emails(self, limit10): 获取未读邮件列表。 if not self.connection: if not self.connect(): return [] self.connection.select(INBOX) status, messages self.connection.search(None, UNSEEN) if status ! OK: return [] email_ids messages[0].split() emails [] for eid in email_ids[-limit:]: # 取最新的limit封 status, msg_data self.connection.fetch(eid, (RFC822)) if status OK: raw_email msg_data[0][1] msg email.message_from_bytes(raw_email) # 解析发件人、主题、正文 subject, encoding decode_header(msg[Subject])[0] if isinstance(subject, bytes): subject subject.decode(encoding if encoding else utf-8, errorsignore) from_ msg.get(From) # 提取纯文本正文 body if msg.is_multipart(): for part in msg.walk(): content_type part.get_content_type() content_disposition str(part.get(Content-Disposition)) if content_type text/plain and attachment not in content_disposition: try: body part.get_payload(decodeTrue).decode() except: pass break else: content_type msg.get_content_type() if content_type text/plain: try: body msg.get_payload(decodeTrue).decode() except: pass emails.append({ id: eid.decode(), from: from_, subject: subject, body_preview: body[:500] ... if len(body) 500 else body, # 预览 body_full: body }) return emails def mark_as_read(self, email_id): 将邮件标记为已读。 if self.connection: self.connection.store(email_id, FLAGS, \\Seen) def disconnect(self): if self.connection: self.connection.close() self.connection.logout() # 在tools.py中暴露为代理可用的工具 from email_handler import EmailClient _email_client EmailClient() def tool_fetch_unread_emails(limit: int 5) - str: 获取收件箱中未读邮件的列表。返回一个包含邮件摘要发件人、主题、正文预览的格式化字符串。 参数limit最多获取的邮件数量默认为5。 emails _email_client.fetch_unread_emails(limit) if not emails: return 当前没有未读邮件。 result f共找到 {len(emails)} 封未读邮件\n for i, eml in enumerate(emails): result f\n--- 邮件 {i1} ---\n result f发件人{eml[from]}\n result f主题{eml[subject]}\n result f内容预览{eml[body_preview]}\n # 注意这里不标记为已读由后续决策决定 return result def tool_get_email_details(email_index: int) - str: 获取指定索引从1开始的未读邮件的完整内容。需要先调用fetch_unread_emails获取列表。 参数email_index邮件在最近未读列表中的序号1, 2, 3...。 # 这里需要一个缓存机制来存储最近一次fetch的结果简化起见我们重新fetch emails _email_client.fetch_unread_emails(limit10) if 1 email_index len(emails): email_detail emails[email_index - 1] # 可以在这里标记为已读 _email_client.mark_as_read(email_detail[id]) return f邮件详情\n发件人{email_detail[from]}\n主题{email_detail[subject]}\n完整内容\n{email_detail[body_full]} else: return f错误未找到索引为 {email_index} 的未读邮件。请确认索引范围1-{len(emails)}。现在我们在main.py中将所有部分组装起来# main.py import os from dotenv import load_dotenv from agent_core import SimpleAgent import tools load_dotenv() def main(): # 1. 定义工具集 my_tools [ { name: fetch_unread_emails, description: 检查并获取收件箱中的未读邮件列表。返回邮件的发件人、主题和内容预览。可选参数limit默认5控制获取数量。, function: tools.tool_fetch_unread_emails }, { name: get_email_details, description: 获取指定序号基于fetch_unread_emails返回的列表的邮件的完整内容。参数email_index是从1开始的整数。, function: tools.tool_get_email_details }, # 未来可以添加更多工具如分析邮件内容、自动回复、分类归档等 ] # 2. 初始化代理 agent SimpleAgent(modelgpt-4, toolsmy_tools) # 3. 运行代理 goal 请检查我的未读邮件找出所有看起来像是会议邀请的邮件并为我提取出会议时间、地点和主题。 print(f开始执行任务{goal}) result agent.run(goal) print(\n *50) print(代理执行结果) print(result) if __name__ __main__: main()运行这个程序代理就会开始工作。它会先调用fetch_unread_emails工具获取邮件列表然后LLM会根据邮件预览判断哪些可能是会议邀请并决定调用get_email_details获取疑似邮件的完整内容最后再从完整内容中提取信息。整个过程完全由LLM驱动我们只需要提供目标和工具。3.3 提示词工程与代理行为调优初始版本的代理很可能表现不佳。LLM可能无法准确识别会议邀请或者提取信息格式混乱。这时就需要进行提示词工程调优。我们回到_build_prompt函数针对邮件处理场景进行强化。原始的提示词比较通用。我们可以为邮件分析任务设计一个更专业、约束更强的提示词。修改agent_core.py中的_build_prompt方法或者为不同任务类型创建不同的提示词模板。例如在邮件处理任务中我们可以在提示词中加入领域知识def _build_prompt_for_email_analysis(self, state): tools_desc ... prompt f 你是一个专业的邮件助理擅长识别和提取会议信息。 **你的终极目标**{state[goal]} **当前进展** 已完成步骤{state[completed_steps]} 现有发现{state[findings]} **你可以使用的工具** {tools_desc} **特别指令** 1. 会议邀请邮件通常包含关键词如“邀请”、“Invitation”、“Meeting”、“会议”、“Calendar”、“日程”、“时间”、“地点”、“Zoom”、“腾讯会议”等。 2. 提取信息时请严格按照以下JSON格式组织即使某些字段未找到也要包含空值 {{meetings: [ {{from: 发件人, subject: 邮件主题, time: 会议时间, location_or_link: 会议地点或链接, topic: 会议主题摘要}} ]}} 3. 你的工作流程应该是a) 获取未读邮件列表b) 逐一检查疑似会议邀请的邮件详情c) 提取信息并汇总。 4. 如果一封邮件明显不是会议邀请如促销广告、个人聊天请忽略它。 请根据当前状态决定下一步行动。你的回复必须是严格的JSON格式只能是以下两种之一 1. 调用工具{{action: call_tool, parameters: {{tool_name: ..., arguments: {{...}}}}}} 2. 最终答复{{action: finalize, parameters: {{summary: 任务总结, extracted_data: {{...}}}}}} 现在请思考并输出你的下一步JSON指令。 return prompt通过加入领域关键词、输出格式范例和工作流程引导我们可以极大地提升LLM在特定任务上的表现。这就是提示词工程的核心通过提供上下文、范例和约束将LLM的通用能力“塑造”成解决特定问题的专家能力。4. 性能优化、安全考量与扩展方向4.1 提升代理的可靠性与效率一个玩具级的代理和真正可用的代理之间隔着无数个细节的优化。1. 状态管理与上下文长度优化LLM有上下文窗口限制。我们的conversation_history会随着对话轮次增长。很快我们会遇到Token超限的问题。解决方案是选择性记忆。不要将完整的工具执行结果可能很长都塞进历史。而是进行总结。例如在current_state[“findings”]中我们存储的是工具返回结果的摘要而非全文。我们可以让LLM自己来总结或者写一个简单的摘要函数。def summarize_tool_result(result: str, max_length200) - str: 对过长的工具执行结果进行摘要。 if len(result) max_length: return result # 简单截取更好的方式是用LLM进行摘要 return result[:max_length] ...已截断2. 错误处理与重试机制网络可能波动API可能限速工具可能出错。代理必须有韧性。在_call_llm和工具函数外围添加重试逻辑和异常捕获至关重要。import tenacity from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def _call_llm_with_retry(self, messages): try: return self._call_llm(messages) except openai.RateLimitError: print(遇到速率限制等待后重试...) raise # tenacity会捕获并重试 except Exception as e: print(f调用LLM失败: {e}) return f系统错误{e} # 返回错误信息让代理知晓3. 验证与确认机制对于关键操作如发送邮件、删除文件不能让代理直接执行。应该在工具函数中设计确认步骤。例如发送邮件工具可以先生成邮件草稿返回给用户或记录到日志确认等待一个明确的“批准”指令后再实际发送。4.2 安全与隐私的底线思维将AI代理接入个人邮箱、文件系统甚至外部API安全是头等大事。1. 权限最小化为邮箱创建应用专用密码而非使用主密码。代理运行在沙盒环境或容器中限制其对文件系统的访问范围如只能访问特定目录。工具函数内部对参数进行严格的输入验证和清洗防止路径遍历等攻击。2. 敏感信息处理绝不将API密钥、密码等硬编码在代码中必须使用.env文件或环境变量。考虑对日志中可能出现的邮件内容、个人信息进行脱敏处理。如果代理需要处理高度敏感数据可以研究本地化模型如通过Ollama部署本地LLM来避免数据上传至第三方。3. 操作审计代理的所有决策和操作都应被完整记录到日志文件中包括时间戳、接收到的指令、调用的工具、参数和结果。这既是调试的需要也是事后审计的依据。4.3 从原型到实用系统的扩展思路当基础代理运行稳定后可以考虑以下几个扩展方向使其从一个脚本进化成一个实用的个人系统1. 持久化任务队列与调度使用schedule或APScheduler库让代理可以定时运行如每30分钟检查一次邮箱。将待处理的任务和目标存入轻量级数据库如SQLite实现任务队列避免单次运行内存状态的丢失。2. 复杂工作流编排当前代理是单目标、线性的。可以引入更强大的框架如LangChain、AutoGen的概念实现多代理协作。例如一个“调度代理”负责接收用户指令并分解一个“信息收集代理”专门调用搜索和读取工具一个“分析写作代理”负责整理和生成报告它们通过共享状态或消息队列进行通信。3. 记忆与学习能力为代理添加向量数据库如ChromaDB让它能够记住历史交互。例如每次处理完一封来自某人的邮件后可以将关键信息如该人的偏好、过往会议主题存入向量库。下次再收到此人邮件时代理可以检索相关记忆提供更个性化的处理建议。4. 人机交互界面为代理增加一个简单的交互界面可以是命令行界面CLI、Telegram机器人或一个本地Web页面。这样你就可以随时随地用自然语言给它下达新任务并实时查看执行进度和结果。构建一个像Agen这样的AI代理项目最大的收获不在于做出了一个多么强大的工具而在于亲身体验了将LLM的“智能”与具体的“行动”连接起来的完整链条。你会深刻理解提示词设计如何直接影响行为工具定义的粒度如何平衡灵活与可控以及错误处理如何决定系统的鲁棒性。这个过程充满了调试和迭代但每当看到代理成功完成一个你亲手设计的小任务时那种成就感是无可替代的。它让你真切地感受到AI不再是遥不可及的概念而是可以一步步被塑造、融入日常工作的伙伴。
基于Agen项目构建个人AI代理：从LLM原理到邮件处理实战
发布时间：2026/5/16 13:30:17
1. 项目概述从“Agen”看个人化AI代理的构建思路最近在GitHub上看到一个名为“Agen”的项目作者是Anjuan555。这个项目名本身就很值得玩味——“Agen”很容易让人联想到“Agent”代理但又少了一个“t”。这或许暗示着它并非一个功能完备的通用代理框架而更像是一个起点、一个原型或者一个高度个人化的实现。作为一名长期关注自动化工具和效率提升的从业者我本能地对这类项目产生了兴趣。它不像那些动辄宣称要颠覆行业的明星开源项目更像是一个技术爱好者为了解决自己日常工作中的特定痛点而动手搭建的一个“小工具”。这类项目往往更接地气其设计思路和实现细节对于我们理解如何将大语言模型LLM真正落地到具体、微小的场景中有着独特的参考价值。简单来说Agen项目探索的是如何利用现有的大语言模型API比如OpenAI的GPT系列构建一个能够执行特定、连贯任务的“代理”。这里的“代理”不是指那种拥有长期记忆、能自主规划复杂任务的强人工智能体而更像是一个被精心设计好流程的“自动化脚本增强版”。它接收一个目标然后按照预设或动态生成的步骤调用不同的工具可能是搜索、读写文件、调用其他API去逐步完成。这个过程的“大脑”就是大语言模型它负责理解任务、分解步骤、判断下一步该做什么。Agen的价值在于它提供了一个非常轻量级的脚手架让你可以快速试验“LLM工具”这种模式解决那些规则模糊但又有一定模式的重复性工作。这个项目适合谁呢我认为有三类人可能会从中受益。第一类是希望深入理解AI代理底层工作机制的开发者通过阅读和修改这个相对简洁的代码你能清晰地看到提示词工程、工具调用、状态管理的整个链条是如何串联起来的。第二类是有明确自动化需求但觉得现有自动化平台如Zapier, Make不够灵活或希望将AI深度集成到工作流中的技术型用户。第三类则是像我一样的“工具控”享受亲手打造并优化一个能为自己所用的智能助手的过程哪怕它最初的功能非常单一。接下来我将深入拆解这个项目的设计思路、核心实现并分享如何基于它构建一个实用的个人自动化代理。2. 核心架构与设计哲学解析2.1 轻量级与模块化为什么“小”即是美打开Agen的代码仓库第一个直观感受就是其代码结构的简洁。它没有追求大而全的框架式设计没有复杂的依赖注入、也没有抽象出多层继承体系。这种“轻量级”的选择恰恰是其核心设计哲学的体现快速验证想法降低上手门槛。在AI代理领域很多概念尚未定型过早进行过度设计反而会束缚手脚。Agen选择将核心功能——与大语言模型的交互、工具的执行、任务状态的推进——封装成几个清晰的函数或类它们之间的耦合度很低。这种模块化带来的最大好处是可插拔性。例如如果你想更换底层的LLM提供商从OpenAI切换到Claude或国内的其他模型你通常只需要修改模型调用那一小部分代码而任务逻辑、工具定义等上层结构几乎不用动。同样如果你想增加一个新的工具比如让代理能够发送邮件你只需要按照已有的模式定义一个发送邮件的函数并将其注册到工具列表中即可。这种设计鼓励实验和迭代你可以从一个最简单的“文件总结代理”开始逐步为它添加网络搜索、数据提取、内容改写等能力每一步的改动都局限在很小的范围内风险可控。注意这种轻量级设计也意味着它缺乏企业级应用所需的一些特性如分布式任务队列、完善的错误恢复机制、细粒度的权限控制等。因此它更适用于个人或小团队内的自动化场景作为生产力工具而非核心业务系统。2.2 基于LLM的任务分解与调度引擎Agen的核心“智能”来源于大语言模型对任务的分解与调度能力。这并不是一个复杂的规划算法其本质是一个精心设计的提示词Prompt循环。工作流程通常如下目标输入用户给出一个自然语言描述的目标例如“帮我总结今天项目会议记录中的行动项并发送给相关同事”。初始规划Agen会将这个目标连同可用的工具列表如read_file,search_web,send_email一起构成一个提示词发送给LLM。提示词会要求LLM将大目标分解成一系列具体的、可执行的步骤。例如步骤1读取meeting_notes.txt文件步骤2提取所有以“Action:”开头的句子步骤3整理成表格步骤4获取相关同事的邮箱列表步骤5起草邮件正文步骤6发送邮件。步骤执行与循环Agen拿到LLM输出的步骤列表后开始执行第一步。执行完成后它将当前状态已完成的步骤、结果、剩余目标再次组合成提示词询问LLM“下一步该做什么”。LLM根据当前状态决定是继续执行下一个既定步骤还是需要调整计划比如发现文件不存在需要先搜索文件然后输出下一个动作指令。终止判断循环持续进行直到LLM判断目标已达成或遇到无法克服的障碍输出“任务完成”或“任务失败”的信号。这个流程的关键在于状态提示词的设计。你需要让LLM清晰地知道它已经做了什么、得到了什么结果、还有什么没做、以及它能用什么工具。Agen的代码中通常会有一个_build_agent_prompt之类的函数这里就是核心战场。你需要反复调试这个提示词以确保LLM能做出合理的决策。例如在提示词中明确要求LLM“在决定下一步时请严格基于当前已知信息不要假设未获取的数据”可以有效减少LLM的“幻觉”避免它跳过一个未执行的搜索步骤而去直接撰写需要搜索结果的报告。2.3 工具系统的设计与扩展实践工具Tools是代理的手和脚。Agen项目中的工具本质上就是Python函数。一个典型的工具定义包含三部分函数实现、自然语言描述、参数模式。例如一个网络搜索工具可能这样定义def search_web(query: str) - str: 使用搜索引擎查询信息。 # 实际调用SerpAPI或SearXNG等搜索API的代码 results call_search_api(query) return results # 在代理中注册工具 agent.register_tool({ name: search_web, description: 当需要获取最新的、未知的或实时信息时使用此工具。输入一个搜索查询字符串。, function: search_web, parameters: {query: {type: string, description: 搜索关键词}} })描述Description是工具的灵魂。LLM并不理解你的代码它完全依靠这段自然语言描述来决定在什么情况下调用这个工具以及如何调用。因此描述必须精确、无歧义并说明使用场景和输入格式。比如“读写文件”这个描述就太模糊了更好的描述是“读取指定路径的文本文件内容。输入参数应为文件的绝对路径字符串。适用于需要获取本地文档信息的场景。”扩展工具是发挥代理威力的关键。除了文件操作、网络搜索、HTTP请求这些通用工具你可以针对特定领域创建高度定制化的工具。例如如果你是电商运营可以创建一个“查询商品库存”的工具它内部调用公司内部的库存管理系统API如果你是研究人员可以创建一个“在arXiv上搜索论文”的工具。将这些工具注册给代理后你就可以用自然语言指挥它“查一下上个月销量Top 10的商品当前库存如果任何一款库存低于安全水位就发邮件提醒采购部。”代理会自动规划先调用某个内部API获取商品列表再循环调用库存查询工具最后判断并调用邮件发送工具。实操心得在定义工具时务必让函数具有鲁棒性和明确的错误处理。因为LLM可能会生成不合法的参数比如一个不存在的文件路径。你的工具函数应该能捕获异常并返回一个结构化的错误信息如“错误文件/xxx/yyy.txt不存在”而不是直接抛出异常导致整个代理崩溃。这个错误信息会被反馈给LLM它就有可能调整计划比如改为先搜索这个文件。3. 从零构建一个个人邮件处理代理为了彻底理解Agen这类项目的精髓最好的方式就是动手构建一个。下面我将以创建一个“智能邮件处理代理”为例展示从环境搭建到核心功能实现的完整过程。这个代理的目标是定期检查邮箱根据邮件内容自动分类、总结并对特定类型的邮件如会议邀请、待办事项进行预处理。3.1 环境准备与基础框架搭建首先我们需要一个干净的Python环境。建议使用venv或conda创建独立环境。# 创建并激活虚拟环境 python -m venv agen_venv source agen_venv/bin/activate # Linux/Mac # agen_venv\Scripts\activate # Windows # 安装核心依赖 pip install openai # 或其他LLM SDK如anthropic, qianfan等 pip install python-dotenv # 用于管理API密钥接下来我们创建项目的基本结构。虽然可以直接借鉴Agen的源码但为了更透彻的理解我们尝试自己组织核心模块。my_email_agent/ ├── .env # 存储API密钥等敏感信息 ├── main.py # 主程序入口 ├── agent_core.py # 代理核心逻辑状态机、提示词构建 ├── tools.py # 所有工具函数的定义 ├── email_handler.py # 邮件相关的专用工具和逻辑 └── config.py # 配置文件在agent_core.py中我们构建最简化的代理循环骨架# agent_core.py import openai import json from typing import List, Dict, Any class SimpleAgent: def __init__(self, model: str gpt-4, tools: List[Dict] None): self.model model self.tools tools or [] self.conversation_history [] # 记录与LLM的交互历史 def run(self, initial_goal: str) - str: 运行代理直到任务完成或失败。 current_state {goal: initial_goal, completed_steps: [], findings: } for _ in range(10): # 设置最大步数防止无限循环 # 1. 构建提示词 prompt self._build_prompt(current_state) self.conversation_history.append({role: user, content: prompt}) # 2. 调用LLM获取下一步指令 response self._call_llm(self.conversation_history) self.conversation_history.append({role: assistant, content: response}) # 3. 解析LLM的响应期望是JSON格式包含action和parameters try: instruction json.loads(response) action instruction.get(action) # 例如 call_tool params instruction.get(parameters, {}) except json.JSONDecodeError: # 如果LLM没有返回JSON可能是最终答复或错误 if 任务完成 in response or final answer in response.lower(): return f任务成功结束。最终结果{response} current_state[findings] f\nLLM返回了非JSON响应{response} continue # 4. 执行动作这里以调用工具为例 if action call_tool: tool_name params.get(tool_name) tool_args params.get(arguments, {}) tool_func self._get_tool(tool_name) if tool_func: result tool_func(**tool_args) current_state[findings] f\n执行工具[{tool_name}]结果{result} current_state[completed_steps].append(f调用工具 {tool_name}) else: current_state[findings] f\n错误未找到工具 {tool_name} elif action finalize: return f任务完成。总结{params.get(summary, )} else: current_state[findings] f\n未知指令{action} # 5. 检查目标是否达成简化逻辑 # 在实际应用中这里可以加入更复杂的判断逻辑 return 任务因达到最大步数而终止。 def _build_prompt(self, state: Dict) - str: 构建当前状态下的提示词。这是核心中的核心。 tools_desc \n.join([f- {t[name]}: {t[description]} for t in self.tools]) prompt f 你是一个智能助理。你的目标是{state[goal]}。 到目前为止你已经完成了以下步骤 {chr(10).join(state[completed_steps]) if state[completed_steps] else 暂无。} 你目前掌握的信息和发现 {state[findings] if state[findings] else 暂无。} 你可以使用以下工具 {tools_desc} 请根据当前目标和已有信息决定下一步做什么。你必须以严格的JSON格式回复格式如下 {{action: call_tool, parameters: {{tool_name: 工具名, arguments: {{...}}}}}} 或者如果你认为目标已达成可以回复 {{action: finalize, parameters: {{summary: 任务总结}}}} 请只输出JSON不要有其他任何内容。 return prompt def _call_llm(self, messages: List[Dict]) - str: 调用OpenAI API示例。实际使用时请处理错误和重试。 client openai.OpenAI(api_keyos.getenv(OPENAI_API_KEY)) response client.chat.completions.create( modelself.model, messagesmessages, temperature0.1, # 低温度保证输出稳定适合结构化指令 ) return response.choices[0].message.content def _get_tool(self, name: str): 根据名称查找工具函数。 for tool in self.tools: if tool[name] name: return tool[function] return None这个骨架虽然简单但包含了代理最核心的循环构建提示 - LLM决策 - 执行动作 - 更新状态。_build_prompt函数是控制LLM行为的关键你需要不断优化它来提升代理的可靠性。3.2 邮件工具集的具体实现有了核心框架接下来实现邮件处理相关的工具。我们在tools.py和email_handler.py中编写。首先我们需要安全地读取邮箱。这里使用imaplib和email标准库。为了安全邮箱凭证存放在.env文件中。# email_handler.py import imaplib import email from email.header import decode_header import os from dotenv import load_dotenv import time load_dotenv() class EmailClient: def __init__(self): self.imap_server os.getenv(IMAP_SERVER, imap.gmail.com) self.email_address os.getenv(EMAIL_ADDRESS) self.password os.getenv(EMAIL_PASSWORD) # 对于Gmail建议使用应用专用密码 self.connection None def connect(self): 连接到IMAP服务器。 try: self.connection imaplib.IMAP4_SSL(self.imap_server) self.connection.login(self.email_address, self.password) print(邮箱连接成功) return True except Exception as e: print(f邮箱连接失败: {e}) return False def fetch_unread_emails(self, limit10): 获取未读邮件列表。 if not self.connection: if not self.connect(): return [] self.connection.select(INBOX) status, messages self.connection.search(None, UNSEEN) if status ! OK: return [] email_ids messages[0].split() emails [] for eid in email_ids[-limit:]: # 取最新的limit封 status, msg_data self.connection.fetch(eid, (RFC822)) if status OK: raw_email msg_data[0][1] msg email.message_from_bytes(raw_email) # 解析发件人、主题、正文 subject, encoding decode_header(msg[Subject])[0] if isinstance(subject, bytes): subject subject.decode(encoding if encoding else utf-8, errorsignore) from_ msg.get(From) # 提取纯文本正文 body if msg.is_multipart(): for part in msg.walk(): content_type part.get_content_type() content_disposition str(part.get(Content-Disposition)) if content_type text/plain and attachment not in content_disposition: try: body part.get_payload(decodeTrue).decode() except: pass break else: content_type msg.get_content_type() if content_type text/plain: try: body msg.get_payload(decodeTrue).decode() except: pass emails.append({ id: eid.decode(), from: from_, subject: subject, body_preview: body[:500] ... if len(body) 500 else body, # 预览 body_full: body }) return emails def mark_as_read(self, email_id): 将邮件标记为已读。 if self.connection: self.connection.store(email_id, FLAGS, \\Seen) def disconnect(self): if self.connection: self.connection.close() self.connection.logout() # 在tools.py中暴露为代理可用的工具 from email_handler import EmailClient _email_client EmailClient() def tool_fetch_unread_emails(limit: int 5) - str: 获取收件箱中未读邮件的列表。返回一个包含邮件摘要发件人、主题、正文预览的格式化字符串。 参数limit最多获取的邮件数量默认为5。 emails _email_client.fetch_unread_emails(limit) if not emails: return 当前没有未读邮件。 result f共找到 {len(emails)} 封未读邮件\n for i, eml in enumerate(emails): result f\n--- 邮件 {i1} ---\n result f发件人{eml[from]}\n result f主题{eml[subject]}\n result f内容预览{eml[body_preview]}\n # 注意这里不标记为已读由后续决策决定 return result def tool_get_email_details(email_index: int) - str: 获取指定索引从1开始的未读邮件的完整内容。需要先调用fetch_unread_emails获取列表。 参数email_index邮件在最近未读列表中的序号1, 2, 3...。 # 这里需要一个缓存机制来存储最近一次fetch的结果简化起见我们重新fetch emails _email_client.fetch_unread_emails(limit10) if 1 email_index len(emails): email_detail emails[email_index - 1] # 可以在这里标记为已读 _email_client.mark_as_read(email_detail[id]) return f邮件详情\n发件人{email_detail[from]}\n主题{email_detail[subject]}\n完整内容\n{email_detail[body_full]} else: return f错误未找到索引为 {email_index} 的未读邮件。请确认索引范围1-{len(emails)}。现在我们在main.py中将所有部分组装起来# main.py import os from dotenv import load_dotenv from agent_core import SimpleAgent import tools load_dotenv() def main(): # 1. 定义工具集 my_tools [ { name: fetch_unread_emails, description: 检查并获取收件箱中的未读邮件列表。返回邮件的发件人、主题和内容预览。可选参数limit默认5控制获取数量。, function: tools.tool_fetch_unread_emails }, { name: get_email_details, description: 获取指定序号基于fetch_unread_emails返回的列表的邮件的完整内容。参数email_index是从1开始的整数。, function: tools.tool_get_email_details }, # 未来可以添加更多工具如分析邮件内容、自动回复、分类归档等 ] # 2. 初始化代理 agent SimpleAgent(modelgpt-4, toolsmy_tools) # 3. 运行代理 goal 请检查我的未读邮件找出所有看起来像是会议邀请的邮件并为我提取出会议时间、地点和主题。 print(f开始执行任务{goal}) result agent.run(goal) print(\n *50) print(代理执行结果) print(result) if __name__ __main__: main()运行这个程序代理就会开始工作。它会先调用fetch_unread_emails工具获取邮件列表然后LLM会根据邮件预览判断哪些可能是会议邀请并决定调用get_email_details获取疑似邮件的完整内容最后再从完整内容中提取信息。整个过程完全由LLM驱动我们只需要提供目标和工具。3.3 提示词工程与代理行为调优初始版本的代理很可能表现不佳。LLM可能无法准确识别会议邀请或者提取信息格式混乱。这时就需要进行提示词工程调优。我们回到_build_prompt函数针对邮件处理场景进行强化。原始的提示词比较通用。我们可以为邮件分析任务设计一个更专业、约束更强的提示词。修改agent_core.py中的_build_prompt方法或者为不同任务类型创建不同的提示词模板。例如在邮件处理任务中我们可以在提示词中加入领域知识def _build_prompt_for_email_analysis(self, state): tools_desc ... prompt f 你是一个专业的邮件助理擅长识别和提取会议信息。 **你的终极目标**{state[goal]} **当前进展** 已完成步骤{state[completed_steps]} 现有发现{state[findings]} **你可以使用的工具** {tools_desc} **特别指令** 1. 会议邀请邮件通常包含关键词如“邀请”、“Invitation”、“Meeting”、“会议”、“Calendar”、“日程”、“时间”、“地点”、“Zoom”、“腾讯会议”等。 2. 提取信息时请严格按照以下JSON格式组织即使某些字段未找到也要包含空值 {{meetings: [ {{from: 发件人, subject: 邮件主题, time: 会议时间, location_or_link: 会议地点或链接, topic: 会议主题摘要}} ]}} 3. 你的工作流程应该是a) 获取未读邮件列表b) 逐一检查疑似会议邀请的邮件详情c) 提取信息并汇总。 4. 如果一封邮件明显不是会议邀请如促销广告、个人聊天请忽略它。 请根据当前状态决定下一步行动。你的回复必须是严格的JSON格式只能是以下两种之一 1. 调用工具{{action: call_tool, parameters: {{tool_name: ..., arguments: {{...}}}}}} 2. 最终答复{{action: finalize, parameters: {{summary: 任务总结, extracted_data: {{...}}}}}} 现在请思考并输出你的下一步JSON指令。 return prompt通过加入领域关键词、输出格式范例和工作流程引导我们可以极大地提升LLM在特定任务上的表现。这就是提示词工程的核心通过提供上下文、范例和约束将LLM的通用能力“塑造”成解决特定问题的专家能力。4. 性能优化、安全考量与扩展方向4.1 提升代理的可靠性与效率一个玩具级的代理和真正可用的代理之间隔着无数个细节的优化。1. 状态管理与上下文长度优化LLM有上下文窗口限制。我们的conversation_history会随着对话轮次增长。很快我们会遇到Token超限的问题。解决方案是选择性记忆。不要将完整的工具执行结果可能很长都塞进历史。而是进行总结。例如在current_state[“findings”]中我们存储的是工具返回结果的摘要而非全文。我们可以让LLM自己来总结或者写一个简单的摘要函数。def summarize_tool_result(result: str, max_length200) - str: 对过长的工具执行结果进行摘要。 if len(result) max_length: return result # 简单截取更好的方式是用LLM进行摘要 return result[:max_length] ...已截断2. 错误处理与重试机制网络可能波动API可能限速工具可能出错。代理必须有韧性。在_call_llm和工具函数外围添加重试逻辑和异常捕获至关重要。import tenacity from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def _call_llm_with_retry(self, messages): try: return self._call_llm(messages) except openai.RateLimitError: print(遇到速率限制等待后重试...) raise # tenacity会捕获并重试 except Exception as e: print(f调用LLM失败: {e}) return f系统错误{e} # 返回错误信息让代理知晓3. 验证与确认机制对于关键操作如发送邮件、删除文件不能让代理直接执行。应该在工具函数中设计确认步骤。例如发送邮件工具可以先生成邮件草稿返回给用户或记录到日志确认等待一个明确的“批准”指令后再实际发送。4.2 安全与隐私的底线思维将AI代理接入个人邮箱、文件系统甚至外部API安全是头等大事。1. 权限最小化为邮箱创建应用专用密码而非使用主密码。代理运行在沙盒环境或容器中限制其对文件系统的访问范围如只能访问特定目录。工具函数内部对参数进行严格的输入验证和清洗防止路径遍历等攻击。2. 敏感信息处理绝不将API密钥、密码等硬编码在代码中必须使用.env文件或环境变量。考虑对日志中可能出现的邮件内容、个人信息进行脱敏处理。如果代理需要处理高度敏感数据可以研究本地化模型如通过Ollama部署本地LLM来避免数据上传至第三方。3. 操作审计代理的所有决策和操作都应被完整记录到日志文件中包括时间戳、接收到的指令、调用的工具、参数和结果。这既是调试的需要也是事后审计的依据。4.3 从原型到实用系统的扩展思路当基础代理运行稳定后可以考虑以下几个扩展方向使其从一个脚本进化成一个实用的个人系统1. 持久化任务队列与调度使用schedule或APScheduler库让代理可以定时运行如每30分钟检查一次邮箱。将待处理的任务和目标存入轻量级数据库如SQLite实现任务队列避免单次运行内存状态的丢失。2. 复杂工作流编排当前代理是单目标、线性的。可以引入更强大的框架如LangChain、AutoGen的概念实现多代理协作。例如一个“调度代理”负责接收用户指令并分解一个“信息收集代理”专门调用搜索和读取工具一个“分析写作代理”负责整理和生成报告它们通过共享状态或消息队列进行通信。3. 记忆与学习能力为代理添加向量数据库如ChromaDB让它能够记住历史交互。例如每次处理完一封来自某人的邮件后可以将关键信息如该人的偏好、过往会议主题存入向量库。下次再收到此人邮件时代理可以检索相关记忆提供更个性化的处理建议。4. 人机交互界面为代理增加一个简单的交互界面可以是命令行界面CLI、Telegram机器人或一个本地Web页面。这样你就可以随时随地用自然语言给它下达新任务并实时查看执行进度和结果。构建一个像Agen这样的AI代理项目最大的收获不在于做出了一个多么强大的工具而在于亲身体验了将LLM的“智能”与具体的“行动”连接起来的完整链条。你会深刻理解提示词设计如何直接影响行为工具定义的粒度如何平衡灵活与可控以及错误处理如何决定系统的鲁棒性。这个过程充满了调试和迭代但每当看到代理成功完成一个你亲手设计的小任务时那种成就感是无可替代的。它让你真切地感受到AI不再是遥不可及的概念而是可以一步步被塑造、融入日常工作的伙伴。
