1. 为什么我要从零手搓一个 Agent市面上讲 Agent 的文章十篇里有八篇是拿 LangChain 或者某个框架跑一个 ReAct 循环然后告诉你“这就是 Agent”。我一开始也是这么入门的跑通了 demo 挺开心但真到了要上生产、要处理长对话、要接知识库、要控制成本的时候问题就全冒出来了。框架帮你屏蔽了细节但代价是你不知道底下发生了什么一旦出问题就只能靠猜。所以我决定从零手搓一个 Agent不依赖任何重型框架把 LLM 调用、工具注册、记忆管理、RAG 检索、Rerank 重排这些环节全部自己实现一遍。这个过程踩了不少坑也让我真正理解了 Agent 到底是什么——它不是一个神奇的智能体而是一个由 LLM 驱动的循环控制系统核心就是“让模型决定下一步做什么然后执行再把结果喂回去”。这篇文章适合两类人一是刚接触 Agent 开发、想搞清楚底层原理的开发者二是已经用过框架、但遇到瓶颈想深入理解工程化细节的人。我会从最基础的 LLM 调用讲起一路讲到 RAG 检索、向量库选型、Rerank 策略、记忆管理和工程化部署。代码以 Python 为主但思路是通用的换成任何语言都一样。先明确一个概念Agent 和普通的 LLM 调用有什么区别普通调用是“你问一句模型答一句”一问一答就结束了。Agent 是“你给一个目标模型自己决定要调用哪些工具、分几步完成、每步的结果怎么处理”它是一个多轮循环直到任务完成或者达到终止条件。这个区别听起来简单但工程上的复杂度差了一个数量级。2. Agent 的核心架构拆解2.1 最小可用 Agent 的四个组成部分一个能跑起来的 Agent最少需要四个东西LLM 推理引擎负责理解用户意图、决定下一步动作、生成最终回答。这是大脑。工具注册表告诉模型有哪些工具可以用、每个工具的参数是什么、怎么调用。这是手脚。执行循环控制“推理-行动-观察”的循环处理终止条件、错误重试、超时。这是骨架。记忆系统保存对话历史、中间结果、长期知识。这是记忆。很多人一上来就搞很复杂的架构其实没必要。我建议先用最少的代码跑通一个能查天气、能算数学的 Agent然后再逐步加东西。下面是我手搓的第一版核心循环大概五十行import json class MinimalAgent: def __init__(self, llm_client, tools, max_steps10): self.llm llm_client self.tools {t[name]: t for t in tools} self.max_steps max_steps self.history [] def run(self, user_input): self.history.append({role: user, content: user_input}) for step in range(self.max_steps): response self.llm.chat( messagesself.history, toolslist(self.tools.values()) ) if response.tool_calls: for call in response.tool_calls: result self._execute_tool(call) self.history.append({ role: tool, tool_call_id: call.id, content: str(result) }) else: return response.content return 达到最大步数限制任务未完成 def _execute_tool(self, call): tool self.tools.get(call.name) if not tool: return f错误工具 {call.name} 不存在 try: return tool[func](**json.loads(call.arguments)) except Exception as e: return f工具执行失败{e}这段代码虽然简单但已经包含了 Agent 的核心逻辑循环调用 LLM如果模型返回工具调用就执行否则就返回最终答案。max_steps是防止死循环的保险丝这个非常重要后面会细说。2.2 为什么我选择不用重型框架LangChain、AutoGPT 这些框架确实方便但我手搓的原因有三个第一调试困难。框架封装了太多层出了错你很难定位是 prompt 的问题、工具定义的问题、还是框架本身的 bug。我自己写的代码每一行都知道在干什么出问题直接打断点。第二成本控制。框架往往会在你看不到的地方多发请求比如自动做摘要、自动做检索、自动重试。这些在 demo 阶段无所谓但上了生产token 成本会失控。自己写可以精确控制每一次 LLM 调用。第三定制化需求。真实业务里Agent 的行为往往需要根据具体场景调整比如特定的终止条件、特定的工具调用顺序、特定的错误处理逻辑。框架的抽象层反而会成为阻碍。当然我不是说框架不能用。如果你只是做个原型验证用框架完全没问题。但如果你想深入理解 Agent 的工作原理或者要上生产环境手搓一遍是值得的。2.3 Agent 执行循环中的关键决策点在写循环的时候有几个关键决策点需要想清楚终止条件怎么定最常见的是“模型不再调用工具时终止”但这不够。还需要加上最大步数限制、超时限制、以及特定错误的重试上限。我见过太多 Agent 陷入死循环反复调用同一个工具烧了一堆 token 什么都没干成。工具调用失败怎么办直接把错误信息喂回给模型让它自己决定是重试、换工具、还是放弃。这比你自己写重试逻辑要灵活得多。但要注意错误信息要简洁明了不要把整个堆栈都塞进去否则会浪费大量 token。并行工具调用怎么处理现代 LLM 支持一次返回多个工具调用比如同时查天气和查汇率。这时候可以并行执行但要注意工具之间是否有依赖关系。我的做法是如果工具之间没有依赖就并行执行如果有依赖就串行。3. 工具系统的设计与实现3.1 工具定义的最佳实践工具定义是 Agent 开发中最容易被忽视、但影响最大的环节。模型能不能正确调用工具很大程度上取决于你的工具定义写得好不好。一个好的工具定义包含四个要素名称简短、明确、动词开头。比如get_weather比weather_tool好search_documents比doc_search好。描述说清楚这个工具做什么、什么时候用、什么时候不用。描述要写给模型看不是写给人看。参数每个参数的类型、含义、是否必填、取值范围。参数描述要具体不要写“输入一个字符串”这种废话。返回值说明返回什么格式的数据方便模型理解结果。我举个例子对比一下好坏# 差的定义 { name: search, description: 搜索, parameters: { query: {type: string} } } # 好的定义 { name: search_knowledge_base, description: 在内部知识库中搜索相关文档。当用户询问产品功能、操作步骤、常见问题时使用此工具。不要用于搜索实时信息或外部网页。, parameters: { query: { type: string, description: 搜索关键词应该是用户问题的核心概念不要包含礼貌用语 }, top_k: { type: integer, description: 返回的文档数量默认5范围1-20, default: 5 } } }差别很明显。好的定义会告诉模型什么时候用、什么时候不用、参数怎么填这能大幅降低误调用的概率。3.2 工具注册与动态发现当工具数量多了之后不可能把所有工具都塞进 prompt 里那样会占用大量 token而且模型容易混淆。我的做法是分组注册 动态发现把工具按领域分组比如“数据库操作”、“文件处理”、“网络请求”、“知识库检索”。然后在系统 prompt 里只列出组名和每组工具的简要说明。当模型需要某个领域的工具时再通过一个list_tools工具动态获取该组的详细定义。这样做的好处是prompt 长度可控模型不会被无关工具干扰而且新增工具时不需要改 prompt。TOOL_GROUPS { knowledge: { description: 知识库检索相关工具, tools: [search_knowledge_base, get_document, list_documents] }, data: { description: 数据处理相关工具, tools: [query_database, export_csv, run_sql] } } def list_tools(group_name): group TOOL_GROUPS.get(group_name) if not group: return f工具组 {group_name} 不存在 return [TOOL_REGISTRY[t] for t in group[tools]]3.3 工具执行的安全边界工具执行是 Agent 最容易出安全问题的地方。模型可能会调用你不希望它调用的工具或者传入危险的参数。我总结了几个必须做的防护白名单机制只允许模型调用注册过的工具任何未注册的调用直接拒绝。参数校验对每个参数做类型检查、范围检查、格式检查。比如 SQL 查询工具要检查是否包含DROP、DELETE等危险操作。沙箱执行对于执行代码、访问文件系统的工具一定要在沙箱里跑限制资源使用和访问范围。超时控制每个工具调用都要有超时防止某个工具卡死导致整个 Agent 挂起。审计日志记录每一次工具调用的输入输出方便事后排查问题。注意千万不要让模型直接执行生成的 SQL 或 shell 命令除非你有完善的沙箱和权限控制。我见过有人让 Agent 直接连生产数据库结果模型生成了一个DELETE FROM users幸好有权限限制才没出事。4. RAG 检索让 Agent 拥有外部知识4.1 RAG 的核心流程与常见误区RAGRetrieval-Augmented Generation是 Agent 最重要的能力之一。它的核心思路很简单先把相关文档检索出来再把文档和问题一起喂给 LLM让 LLM 基于文档回答。这样既能利用 LLM 的推理能力又能保证答案基于真实资料减少幻觉。但很多人对 RAG 的理解停留在“向量检索 拼 prompt”这个层面实际做起来会发现效果很差。问题出在几个地方误区一以为向量检索就够了。向量检索擅长语义相似但对精确匹配、关键词匹配、结构化查询效果不好。比如用户问“2024年Q3的营收是多少”向量检索可能返回一堆讲营收的文档但就是找不到那个具体的数字。误区二以为 chunk 越小越好。chunk 太小会丢失上下文chunk 太大会引入噪声。我一般用 500-800 字符作为 chunk 大小重叠 100-200 字符。但这不是固定的要根据文档类型调整。误区三以为检索一次就够了。复杂问题往往需要多轮检索或者需要先检索再重排再检索。这就是所谓的 Agentic RAG——让 Agent 自己决定什么时候检索、检索什么、检索几次。4.2 向量库选型到底该用什么数据库这是被问得最多的问题之一。我的建议是根据数据量和部署环境选不要盲目追新。向量库适用场景优点缺点FAISS本地开发、小规模数据轻量、快、无需部署不支持持久化、不支持分布式Chroma原型验证、个人项目简单、易用、支持持久化性能一般、不适合大规模Milvus生产环境、大规模数据高性能、支持分布式、功能全部署复杂、资源占用高Qdrant生产环境、中等规模性能好、API 友好、支持过滤生态相对小pgvector已有 PostgreSQL 的场景无需额外部署、支持 SQL性能不如专用向量库Elasticsearch已有 ES 的场景支持混合检索、生态成熟向量功能相对弱我个人的选择逻辑是个人项目用 Chroma 或 FAISS中小规模生产用 Qdrant 或 pgvector大规模生产用 Milvus。如果你已经有 PostgreSQLpgvector 是最省事的方案不用额外维护一个数据库。4.3 向量检索的完整实现下面是我手搓的向量检索模块包含文档切分、向量化、存储、检索四个环节import hashlib from typing import List, Dict class VectorStore: def __init__(self, embedding_client, collection_namedefault): self.embedding embedding_client self.collection collection_name self.documents [] # 实际项目用向量库替代 def add_documents(self, docs: List[Dict], chunk_size600, overlap150): chunks [] for doc in docs: text doc[content] for i in range(0, len(text), chunk_size - overlap): chunk text[i:i chunk_size] if len(chunk) 50: continue chunks.append({ id: hashlib.md5(chunk.encode()).hexdigest(), content: chunk, source: doc.get(source, unknown), metadata: doc.get(metadata, {}) }) # 批量向量化 texts [c[content] for c in chunks] vectors self.embedding.embed_batch(texts) for chunk, vec in zip(chunks, vectors): chunk[vector] vec self.documents.extend(chunks) return len(chunks) def search(self, query: str, top_k5, filtersNone): query_vec self.embedding.embed(query) scored [] for doc in self.documents: if filters and not self._match_filters(doc, filters): continue score self._cosine_similarity(query_vec, doc[vector]) scored.append((score, doc)) scored.sort(keylambda x: x[0], reverseTrue) return [doc for _, doc in scored[:top_k]] def _cosine_similarity(self, a, b): dot sum(x * y for x, y in zip(a, b)) norm_a sum(x * x for x in a) ** 0.5 norm_b sum(x * x for x in b) ** 0.5 return dot / (norm_a * norm_b 1e-8) def _match_filters(self, doc, filters): for key, value in filters.items(): if doc.get(metadata, {}).get(key) ! value: return False return True这段代码是简化版实际项目里应该用真正的向量库。但核心逻辑是一样的切分、向量化、存储、检索。4.4 Rerank检索效果提升的关键一步向量检索出来的结果往往前几名是相关的但排序不一定最优。Rerank 的作用就是对初步检索的结果做精细排序把最相关的文档排到前面。Rerank 模型和 Embedding 模型不一样。Embedding 模型是双塔结构query 和 document 分别编码速度快但精度有限。Rerank 模型是交叉编码器把 query 和 document 拼在一起编码精度高但速度慢。所以典型流程是先用向量检索召回 top 50再用 Rerank 精排取 top 5。class Reranker: def __init__(self, model_client): self.model model_client def rerank(self, query: str, documents: List[Dict], top_k5): pairs [(query, doc[content]) for doc in documents] scores self.model.score_batch(pairs) for doc, score in zip(documents, scores): doc[rerank_score] score documents.sort(keylambda x: x[rerank_score], reverseTrue) return documents[:top_k]Rerank 的效果提升非常明显。我实测下来在同一个知识库上加了 Rerank 之后答案准确率能从 60% 左右提升到 80% 以上。这个投入产出比非常高强烈建议加上。提示Rerank 模型可以选择开源的 BGE-Reranker 系列也可以调用商业 API。如果数据量不大用开源模型本地部署就够了。如果追求效果商业 API 通常更好但成本要考虑。5. 记忆系统让 Agent 记住上下文5.1 短期记忆与长期记忆的区分Agent 的记忆分两种短期记忆和长期记忆。短期记忆就是当前对话的历史包括用户说了什么、Agent 调用了什么工具、工具返回了什么。这部分直接放在 prompt 里但要注意长度控制。对话太长会超出上下文窗口也会增加成本。长期记忆是跨对话的知识比如用户的偏好、历史交互中的重要信息、领域知识。这部分需要存储和检索不能直接放在 prompt 里。很多人把这两者混在一起结果要么是 prompt 太长要么是 Agent 记不住东西。我的做法是短期记忆用滑动窗口 摘要长期记忆用向量库检索。5.2 滑动窗口与摘要压缩短期记忆的管理策略滑动窗口只保留最近 N 轮对话超出的丢弃。简单但会丢失早期信息。摘要压缩把早期对话用 LLM 压缩成摘要保留关键信息。效果好但增加一次 LLM 调用。混合策略最近几轮保留原文更早的压缩成摘要。这是我常用的方案。class MemoryManager: def __init__(self, llm_client, max_recent10, summary_threshold20): self.llm llm_client self.recent [] self.summary self.max_recent max_recent self.summary_threshold summary_threshold def add(self, message): self.recent.append(message) if len(self.recent) self.summary_threshold: self._compress() def _compress(self): to_compress self.recent[:-self.max_recent] self.recent self.recent[-self.max_recent:] text \n.join([f{m[role]}: {m[content]} for m in to_compress]) prompt f请把以下对话压缩成简洁的摘要保留关键信息和结论\n{text} new_summary self.llm.chat([{role: user, content: prompt}]) self.summary f{self.summary}\n{new_summary}.strip() def get_context(self): context [] if self.summary: context.append({role: system, content: f历史对话摘要{self.summary}}) context.extend(self.recent) return context5.3 长期记忆的存储与检索长期记忆的核心是什么时候存、存什么、怎么取。存的时机用户明确表达了偏好、Agent 完成了重要任务、出现了值得记住的结论。不要什么都存否则检索时会引入噪声。存的内容结构化的信息比原始对话更好用。比如“用户偏好用 Python”比“用户说我平时都用 Python 写代码”更好检索。取的方式用当前问题去检索相关记忆把 top 3-5 条注入 prompt。注意要控制数量太多会干扰模型。class LongTermMemory: def __init__(self, vector_store): self.store vector_store def remember(self, content, categorygeneral, importance1.0): self.store.add_documents([{ content: content, metadata: {category: category, importance: importance} }]) def recall(self, query, top_k3): results self.store.search(query, top_ktop_k) return [r[content] for r in results]注意长期记忆要定期清理和更新。过时的信息会误导 Agent矛盾的记忆会让 Agent 困惑。我一般会加一个时间戳检索时优先返回较新的记忆。6. 工程化从 demo 到生产6.1 错误处理与重试策略Agent 执行过程中会遇到各种错误LLM 调用超时、工具执行失败、返回格式不对、达到步数限制。这些都要有处理策略。我的原则是能重试的重试不能重试的降级降级不了的明确报错。LLM 调用失败指数退避重试最多 3 次。工具执行失败把错误信息喂回给模型让它决定下一步。返回格式不对尝试解析失败则要求模型重新生成。达到步数限制返回当前进度让用户决定是否继续。import time def retry_with_backoff(func, max_retries3, base_delay1): for attempt in range(max_retries): try: return func() except Exception as e: if attempt max_retries - 1: raise delay base_delay * (2 ** attempt) time.sleep(delay)6.2 成本控制与性能优化Agent 的成本主要来自 LLM 调用。控制成本的手段缓存相同的问题和上下文直接返回缓存结果。模型分级简单任务用小模型复杂任务用大模型。prompt 精简去掉不必要的说明和示例控制 token 数量。并行调用没有依赖的工具调用并行执行。提前终止模型给出最终答案后立即停止不要继续循环。性能优化方面主要是减少 LLM 调用次数和工具执行时间。我一般会监控每次对话的 token 消耗和响应时间找出瓶颈。6.3 可观测性与日志生产环境的 Agent 必须有完善的日志和监控。我记录的东西包括每次对话的完整输入输出每次 LLM 调用的 prompt、响应、token 数、耗时每次工具调用的参数、结果、耗时错误和异常用户反馈这些日志不仅用于排查问题也是优化 prompt 和工具定义的依据。我经常通过分析日志发现模型在某些场景下反复出错然后针对性地调整 prompt。6.4 部署与扩展Agent 的部署方式取决于你的场景。如果是内部工具一个简单的 FastAPI 服务就够了。如果是对外服务需要考虑并发、限流、鉴权、监控。扩展方面主要是水平扩展和垂直扩展。水平扩展就是多实例部署用负载均衡分发请求。垂直扩展就是提升单实例的性能比如用更快的模型、优化检索速度。提示Agent 服务通常是无状态的会话状态存在外部存储里。这样扩展起来很方便加机器就行。但要注意会话的粘性问题同一个会话的请求最好路由到同一个实例或者把状态完全外置。7. 常见问题与排查技巧7.1 Agent 陷入死循环怎么办这是最常见的问题。表现是 Agent 反复调用同一个工具或者反复说同样的话。原因通常是工具返回的结果没有让模型满意模型不断重试终止条件不明确模型不知道该什么时候停prompt 里有矛盾模型无所适从解决方法加最大步数限制、在 prompt 里明确终止条件、检查工具返回结果是否清晰。我一般还会加一个“重复检测”如果连续两次调用同一个工具且参数相同就强制终止。7.2 检索结果不相关怎么办RAG 效果差的原因很多排查顺序是检查 chunk 切分是否合理有没有把完整语义切碎检查 embedding 模型是否适合你的领域检查检索参数top_k 是否太小加 Rerank考虑混合检索向量 关键词我遇到最多的问题是 chunk 切分不合理。比如把一段完整的操作步骤切成了两半检索时只返回一半模型就理解错了。7.3 工具调用参数错误怎么办模型经常传错参数比如该传数字传了字符串、该传列表传了单个值。解决方法在参数描述里写清楚类型和格式在工具执行前做参数校验和转换把错误信息喂回给模型让它重新生成我一般会在工具定义里加examples字段给模型几个正确的调用示例效果很好。7.4 响应太慢怎么办响应慢的原因可能是 LLM 调用慢、工具执行慢、检索慢。排查方法加耗时日志定位瓶颈LLM 调用可以换更快的模型或加缓存工具执行可以并行化检索可以加索引或减少 top_k我实测下来最大的瓶颈通常是 LLM 调用。如果对延迟敏感可以考虑用流式输出让用户先看到部分结果。7.5 常见问题速查表问题可能原因解决方法死循环终止条件不明确加最大步数、重复检测检索不相关chunk 切分不合理调整 chunk 大小和重叠参数错误工具定义不清晰补充参数描述和示例响应慢LLM 调用慢换模型、加缓存、流式输出成本高token 消耗大精简 prompt、模型分级幻觉严重检索结果没用好加 Rerank、强制引用来源8. 我踩过的坑与实操心得第一个坑是过早引入框架。我一开始用 LangChain跑通了 demo 很兴奋但后来发现很多行为不符合预期调试起来非常痛苦。后来我决定手搓一遍虽然花了两周但对 Agent 的理解深了很多。我的建议是先用框架快速验证想法但一定要手搓一遍核心逻辑否则你永远不知道框架在背后做了什么。第二个坑是忽视 prompt 工程。我一开始觉得 prompt 随便写写就行结果模型各种不听话。后来我认真研究了 prompt 的结构把系统提示、工具说明、输出格式、示例都分开写清楚效果好了很多。特别是工具定义一定要写给模型看不是写给人看。第三个坑是不做成本监控。我有个项目上线第一周token 成本超预算三倍。原因是 Agent 在某些场景下会反复调用工具每次调用都是一次 LLM 请求。后来我加了缓存和最大步数限制成本降下来了。所以一定要监控 token 消耗设置预算告警。第四个坑是忽视错误处理。demo 阶段一切顺利上线后各种错误。LLM 超时、工具返回格式不对、网络抖动这些都要处理。我的经验是任何可能失败的地方都要有 fallback不能让一个错误导致整个 Agent 崩溃。第五个坑是检索质量不够重视。我一开始觉得向量检索就够了结果答案准确率只有 60% 左右。后来加了 Rerank、调整了 chunk 策略、加了混合检索准确率提升到 85% 以上。RAG 的效果检索质量占七成LLM 能力占三成。最后分享一个小技巧给 Agent 加一个“思考”步骤。在调用工具之前让模型先输出它的推理过程比如“我需要先查天气然后根据天气推荐活动”。这个思考过程不仅能让模型更准确也方便你调试。我一般会在 prompt 里加一句“在调用工具前先简要说明你的计划”。这个项目后续还可以扩展的方向很多比如多 Agent 协作、工具自动生成、基于反馈的自我优化。但核心逻辑是不变的LLM 驱动循环工具扩展能力记忆保持上下文检索引入知识。把这四件事做好Agent 就能解决大部分实际问题。