☰
523节全手写AI Agent框架:从零实现Python与TypeScript核心逻辑
2026/10/7 12:07:18 网站建设 项目流程

1. 这个项目到底在讲什么

第一次看到“523节全手写实现”这个数字,我的反应是:这要么是个噱头,要么是个狠人。523节,按一天看3节的节奏,得连续看半年。但仔细拆开来看,这个项目的核心价值不在于“课多”,而在于全手写实现这四个字。

市面上讲AI的课程和开源项目多如牛毛,但绝大多数停留在“调包”层面——import一个transformers,加载预训练权重,跑个demo,结束。真正从零手写一个Agent框架、手写注意力机制、手写向量检索的,少之又少。这个项目选择了一条更难走的路:每一行核心代码都自己写,不依赖高层封装。

它解决的是什么问题?说白了,就是从“会用AI”到“理解AI怎么跑起来”之间的鸿沟。你可以用LangChain三行代码搭一个Agent,但当Agent行为不符合预期时,你根本不知道问题出在提示词拼接、工具调用解析还是循环控制逻辑上。手写一遍,这些环节全部暴露在你面前。

适合谁来参考?有三类人收益最大:一是想从传统开发转AI方向的工程师,Python和TypeScript都有涉及,正好覆盖后端和前端两个入口;二是已经会用AI工具但想深入理解底层机制的开发者;三是需要在自己的项目里嵌入Agent能力、但不想被第三方框架绑死的团队。

关键词里出现了Agent、Python、TypeScript、开源,这几个词基本勾勒出了项目的技术轮廓:用Python做核心AI逻辑,用TypeScript做前端交互和类型安全,整体以开源项目的形式组织。下面我按自己的理解,把这个项目的设计思路、核心细节、实操要点和踩坑经验完整拆一遍。

2. 整体架构与设计思路拆解

2.1 为什么选择“全手写”而不是“调包”

这是整个项目最核心的设计决策,值得先说清楚。

调包的好处显而易见:快。用LangChain或者LlamaIndex,一个基础Agent半小时就能跑起来。但问题在于,这些框架的抽象层太厚。当你需要自定义一个工具调用协议、修改记忆检索策略、或者优化多轮对话的上下文管理时,你会发现框架的抽象反而成了障碍——你要么接受它的限制,要么去读它的源码然后hack。

全手写的好处是完全可控。每一个环节的逻辑都是透明的,你可以精确地知道:用户输入经过了几次转换、每次转换消耗了多少token、工具调用的解析用了什么正则、失败重试的策略是什么。这对于需要精细调优的场景来说,是刚需。

注意:全手写不等于“不用任何库”。HTTP请求可以用requests,向量计算可以用numpy,前端可以用React。手写的是核心逻辑层,不是基础设施层。这个边界要划清楚,否则就变成了重复造轮子。

2.2 Python和TypeScript的分工逻辑

项目同时涉及Python和TypeScript,这不是为了炫技,而是有明确的职责划分。

Python负责的部分:Agent的核心循环、工具注册与调用、记忆管理、向量检索、模型API的封装。选择Python的原因很直接——AI生态的默认语言,numpy、sentence-transformers这些库都是Python优先,而且模型API的官方SDK通常也是Python版本最完整。

TypeScript负责的部分:前端交互界面、类型定义、流式输出的处理、以及部分需要跑在浏览器端的轻量逻辑。TypeScript的类型系统在这里发挥了关键作用——Agent的工具定义、消息格式、状态流转,全部用interface和type约束住,编译期就能发现大部分集成错误。

两边通过HTTP API或者WebSocket通信。Python端暴露RESTful接口,TypeScript端消费这些接口并渲染UI。这种前后端分离的架构,也让项目更容易被不同技术背景的人单独使用——你只想学Agent逻辑,就看Python部分;你只想学前端集成,就看TypeScript部分。

2.3 523节的内容组织方式

523节不是随便凑的数。按我的理解,它大致分为几个模块:

  • 基础篇(约80节):Python环境搭建、TypeScript类型系统入门、HTTP协议基础、异步编程模型。这部分是给基础薄弱的人补课的。
  • 核心篇(约200节):手写Agent循环、工具调用协议设计、提示词模板引擎、对话记忆管理、流式输出处理。这是整个项目最硬核的部分。
  • 进阶篇(约150节):向量数据库手写实现、RAG检索增强生成、多Agent协作、并发控制与限流。
  • 实战篇(约90节):完整项目案例,包括代码助手、文档问答、自动化工作流等。

这种组织方式的好处是渐进式。你不会一上来就被Agent的复杂逻辑劝退,而是从最基础的环境配置开始,一步步走到能跑通一个完整项目。

3. 核心细节解析与实操要点

3.1 Agent循环的手写实现

Agent的本质是一个循环:接收输入→思考→选择工具→执行工具→观察结果→继续思考→输出最终答案。这个循环看起来简单,但手写起来有几个关键点。

第一是终止条件的设计。你不能让Agent无限循环下去。常见的做法是设置最大迭代次数(比如10次),或者检测到模型输出中包含特定的终止标记时退出。我建议两者结合:迭代次数作为硬限制,终止标记作为软退出。

第二是工具调用的解析。模型输出的工具调用请求通常是JSON格式,但模型有时候会输出不规范的JSON——比如多了一个逗号、少了引号、或者嵌套了markdown代码块标记。你需要一个健壮的解析器,先尝试直接json.loads,失败后尝试提取代码块内容再解析,再失败就返回错误让模型重新生成。

第三是错误处理。工具执行失败是常态——API超时、参数错误、权限不足。关键是要把错误信息作为观察结果返回给模型,让模型自己决定是重试、换工具还是放弃。这比直接抛异常给用户要优雅得多。

def agent_loop(user_input, max_iterations=10): messages = [{"role": "user", "content": user_input}] for i in range(max_iterations): response = call_llm(messages) if response.is_final_answer: return response.content tool_call = parse_tool_call(response.content) if tool_call is None: messages.append({"role": "assistant", "content": response.content}) messages.append({"role": "user", "content": "请使用工具或给出最终答案"}) continue try: result = execute_tool(tool_call.name, tool_call.args) except Exception as e: result = f"工具执行失败: {str(e)}" messages.append({"role": "assistant", "content": response.content}) messages.append({"role": "user", "content": f"观察结果: {result}"}) return "达到最大迭代次数,未能完成任务"

这段代码看起来简单,但每一行背后都有坑。比如messages的拼接顺序,必须是user→assistant→user交替,否则某些模型API会报错。再比如工具执行失败时,返回的错误信息要足够具体,让模型能判断是参数问题还是网络问题。

3.2 TypeScript类型定义的关键作用

在TypeScript端,类型定义不只是为了编辑器提示,它直接影响了运行时的正确性。

Agent的工具定义、消息格式、状态机,全部用类型约束住。比如一个工具的定义:

interface ToolDefinition { name: string; description: string; parameters: { type: 'object'; properties: Record<string, { type: string; description: string; enum?: string[]; }>; required: string[]; }; } interface AgentMessage { role: 'user' | 'assistant' | 'system'; content: string; toolCalls?: ToolCall[]; toolCallId?: string; }

这些类型定义在编译期就能发现大部分集成错误。比如你忘记给某个工具参数加description,TypeScript会报错;比如你在处理消息时把role写成了'bot'而不是'assistant',编译不过。

实操心得:TypeScript的.d.ts声明文件在这里特别有用。当你从Python端拿到JSON数据时,可以先定义一个对应的interface,然后用类型断言或者运行时校验(比如zod)来确保数据结构正确。这比any走天下要安全得多。

3.3 向量检索的手写实现

RAG是Agent项目里绕不开的话题。很多人直接用现成的向量数据库,但手写一遍能让你理解背后的原理。

核心步骤就三步:嵌入、存储、检索。

嵌入就是把文本转成向量。可以用OpenAI的embedding API,也可以用本地的sentence-transformers模型。本地模型的优势是免费、离线可用,劣势是效果可能不如大厂API。

存储就是把这些向量存起来。最简单的做法是用numpy数组存,配合一个id到文本的映射表。数据量大了再考虑用FAISS或者Annoy这些专门的库。

检索就是计算查询向量和所有存储向量的相似度,返回top-k。相似度通常用余弦相似度,计算方式是点积除以模长乘积。

import numpy as np class SimpleVectorStore: def __init__(self): self.vectors = [] self.texts = [] def add(self, text, vector): self.texts.append(text) self.vectors.append(vector) def search(self, query_vector, top_k=5): if not self.vectors: return [] matrix = np.array(self.vectors) query = np.array(query_vector) similarities = np.dot(matrix, query) / ( np.linalg.norm(matrix, axis=1) * np.linalg.norm(query) ) indices = np.argsort(similarities)[::-1][:top_k] return [(self.texts[i], similarities[i]) for i in indices]

这个实现很粗糙,但足够让你理解向量检索的本质。实际项目中,你需要考虑分块策略(chunk size和overlap)、嵌入模型的选择、以及检索后的重排序。

3.4 流式输出的前后端配合

流式输出是提升用户体验的关键。用户不需要等整个回答生成完,而是可以看到文字一个个蹦出来。

Python端用SSE(Server-Sent Events)或者WebSocket推送数据。SSE更简单,单向推送就够了。关键是要处理好分块——模型API返回的流是token级别的,你需要按句子或者按固定长度聚合后再推送给前端,避免前端频繁重渲染。

TypeScript端用EventSource或者WebSocket接收数据,然后增量更新UI。React里可以用useState配合useEffect,每次收到新数据就append到现有内容后面。

注意:流式输出时,错误处理会更复杂。因为连接可能在中途断开,你需要在前端做重连逻辑,在后端做超时清理。我踩过的坑是:没有设置心跳机制,导致长时间没有数据传输时连接被中间层断开。

4. 实操过程与核心环节实现

4.1 环境搭建:从零到跑通第一个Agent

先说Python端的环境。我推荐用conda或者venv创建独立环境,避免污染系统Python。

python -m venv agent-env source agent-env/bin/activate # Windows用 agent-env\Scripts\activate pip install numpy requests openai python-dotenv

如果你要用本地的嵌入模型,还需要装sentence-transformers和torch。这两个包比较大,建议配置国内镜像源加速。

TypeScript端用Node.js 18以上版本,包管理用pnpm或者npm都行。

npm init -y npm install typescript ts-node @types/node npx tsc --init

tsconfig.json里关键配置:target设为ES2020以上,module设为commonjs或者esnext,strict设为true。strict一定要开,虽然初期会报很多类型错误,但长期来看能省下大量调试时间。

4.2 模型API的封装与重试策略

调用模型API是Agent项目里最频繁的操作,封装得好不好直接影响开发效率。

我的做法是写一个统一的LLMClient类,把API调用、重试、超时、日志全部封装进去。重试策略用指数退避:第一次失败等1秒,第二次等2秒,第三次等4秒,最多重试3次。

import time import requests class LLMClient: def __init__(self, api_key, base_url, model, max_retries=3): self.api_key = api_key self.base_url = base_url self.model = model self.max_retries = max_retries def chat(self, messages, temperature=0.7, stream=False): for attempt in range(self.max_retries): try: response = requests.post( f"{self.base_url}/chat/completions", headers={"Authorization": f"Bearer {self.api_key}"}, json={ "model": self.model, "messages": messages, "temperature": temperature, "stream": stream }, timeout=60 ) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: if attempt == self.max_retries - 1: raise wait_time = 2 ** attempt print(f"请求失败,{wait_time}秒后重试: {e}") time.sleep(wait_time)

这里有个细节:timeout要设得合理。太短了容易误杀正常请求,太长了用户等得着急。60秒是个比较平衡的值,流式请求可以设更长。

4.3 工具注册与调用的完整流程

工具是Agent的手和脚。没有工具,Agent只能聊天;有了工具,Agent才能查天气、搜网页、执行代码。

工具注册的核心是一个字典,key是工具名,value是工具的定义和执行函数。

class ToolRegistry: def __init__(self): self.tools = {} def register(self, name, description, parameters, func): self.tools[name] = { "definition": { "name": name, "description": description, "parameters": parameters }, "func": func } def get_definitions(self): return [t["definition"] for t in self.tools.values()] def execute(self, name, args): if name not in self.tools: raise ValueError(f"未知工具: {name}") return self.tools[name]["func"](**args)

注册一个天气查询工具:

def get_weather(city: str) -> str: # 实际项目中调用天气API return f"{city}今天晴,气温25度" registry.register( name="get_weather", description="查询指定城市的天气", parameters={ "type": "object", "properties": { "city": {"type": "string", "description": "城市名称"} }, "required": ["city"] }, func=get_weather )

工具描述的质量直接影响Agent的调用准确率。description要写清楚工具能做什么、什么时候用、参数是什么格式。我见过太多因为description写得太模糊导致Agent乱调工具的情况。

4.4 对话记忆的管理策略

Agent需要记住之前的对话,否则每轮都是失忆状态。但你不能把所有历史都塞进上下文,token有限且贵。

常见的策略有三种:

全量保留:简单粗暴,把所有历史消息都传给模型。适合对话轮次少的场景,超过10轮就会爆token。

滑动窗口:只保留最近N轮对话。实现简单,但会丢失早期的重要信息。

摘要压缩:当历史超过一定长度时,用模型把早期对话压缩成摘要。效果好但增加了一次模型调用。

我的做法是滑动窗口+关键信息提取。保留最近5轮完整对话,同时从更早的对话中提取关键实体(人名、地名、数字)作为附加上下文。

class ConversationMemory: def __init__(self, max_recent=5): self.messages = [] self.max_recent = max_recent self.key_entities = {} def add(self, role, content): self.messages.append({"role": role, "content": content}) self._extract_entities(content) def get_context(self): recent = self.messages[-self.max_recent * 2:] if self.key_entities: entity_str = "已知信息: " + ", ".join( f"{k}={v}" for k, v in self.key_entities.items() ) return [{"role": "system", "content": entity_str}] + recent return recent

实体提取可以用简单的正则,也可以用模型。正则快但覆盖不全,模型准但慢。我一般用正则处理数字和日期,用模型处理人名和专有名词。

5. 常见问题与排查技巧实录

5.1 Agent不调用工具怎么办

这是最常见的问题。你定义了一堆工具,但Agent就是不用,直接自己编答案。

原因通常有三个:一是工具描述不够清晰,Agent不知道什么时候该用;二是系统提示词没有强调“优先使用工具”;三是模型本身的能力限制。

解决办法:在系统提示词里明确写“当需要实时信息或计算时,必须使用提供的工具,不要自己编造答案”。同时检查工具描述,确保每个工具都有明确的触发场景说明。

5.2 工具调用参数解析失败

模型输出的JSON格式不对,导致解析失败。这种情况在国产模型上尤其常见。

排查步骤:先打印模型原始输出,看看它到底生成了什么。如果是markdown代码块包裹的JSON,就提取代码块内容再解析。如果是JSON本身格式错误,可以尝试用json5或者demjson这些宽松的解析库。

实操心得:在提示词里明确要求“工具调用请输出纯JSON,不要包含markdown标记”,能减少大部分解析问题。

5.3 流式输出中文乱码

这个问题通常出在编码上。Python端要确保response的encoding是utf-8,TypeScript端要确保TextDecoder指定了utf-8。

const decoder = new TextDecoder('utf-8'); const text = decoder.decode(chunk, { stream: true });

注意stream: true这个参数,不加的话多字节字符会被截断。

5.4 并发请求下的状态污染

如果你的Agent服务要同时处理多个用户的请求,全局变量会导致状态污染。比如你把对话历史存在一个全局list里,两个用户的消息就会混在一起。

解决办法很简单:每个请求创建一个独立的Agent实例,所有状态都封装在实例内部。Python里用类,TypeScript里用闭包或者类,都能达到这个效果。

5.5 常见问题速查表

问题现象可能原因排查方法解决方案
Agent不调用工具工具描述模糊检查工具description补充触发场景说明
JSON解析失败模型输出格式不规范打印原始输出提取代码块+宽松解析
流式输出乱码编码不一致检查两端编码设置统一utf-8+stream解码
响应速度慢上下文过长统计token数量滑动窗口+摘要压缩
工具执行超时外部API不稳定查看超时日志设置合理timeout+重试
多用户状态混乱全局变量污染检查状态存储位置每请求独立实例

6. 从523节里真正该学到的东西

523节的内容量确实大,但如果你时间有限,我建议优先看这几块:Agent循环的实现(理解核心机制)、工具注册与调用(最实用的部分)、流式输出处理(提升体验的关键)、以及向量检索的手写实现(RAG的基础)。

其他的比如前端UI美化、部署配置这些,可以等核心逻辑跑通了再回头看。毕竟这个项目的价值在于让你理解AI应用是怎么跑起来的,而不是教你调CSS。

我在实际跑这个项目的过程中,最大的收获不是某个具体的代码技巧,而是对Agent行为有了可预测性。以前用框架的时候,Agent出问题我只能猜;现在每一行逻辑都是我写的,出问题我能直接定位到是哪一步的哪个判断出了偏差。这种掌控感,是调包永远给不了的。

另外说一个实际体会:手写一遍之后,再看那些框架的源码,会发现它们的设计思路变得非常清晰。因为你已经踩过了所有的坑,知道每个抽象是为了解决什么问题。这时候再去用框架,就是带着理解去用,而不是盲目地调API。

最后分享一个小技巧:如果你在学这个项目的过程中卡住了,不要死磕。先把那一节的代码完整抄一遍跑通,再回头理解每一行的含义。有时候跑通比理解更重要,跑通了之后理解会自然发生。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询