這次我們來看一個偏“硬核”的方向不用任何現(xiàn)成 Agent 框架從零搭建一套屬于自己的智能體工具鏈。所謂 Agent本質(zhì)上就是讓大模型不只停留在“對話”而是具備工具調(diào)用、任務(wù)拆解、記憶管理和自主執(zhí)行的能力。市面上的 LangChain、AutoGPT、MetaGPT 等框架很多但如果只是調(diào)用框架 API很多底層機制其實是黑盒。自己動手搭一遍之后再回頭看這些框架的設(shè)計思路會通透很多。這篇文章會從工具鏈的整體架構(gòu)講起然后直接進入代碼層面實現(xiàn)一個最小可運行的 Agent 核心LLM 接口接入、工具注冊與調(diào)用、短期/長期記憶、執(zhí)行循環(huán)、批量任務(wù)調(diào)度。還會給出 API 接入示例、資源占用觀察方法、常見問題排查清單和工程化最佳實踐。適合以下讀者想理解 Agent 工作原理的開發(fā)者準備做 Agent 相關(guān)面試或項目沉淀的工程師需要把 Agent 接入業(yè)務(wù)系統(tǒng)做自動化測試、內(nèi)容生成、批量任務(wù)處理的技術(shù)人員。不需要有框架使用經(jīng)驗但要會 Python 基礎(chǔ)并且至少有一個可調(diào)用的 LLM API 密鑰。1. 核心能力速覽能力項說明項目類型智能體Agent開發(fā)工具鏈從零手寫核心機制核心能力工具注冊與調(diào)用、任務(wù)拆解、多輪執(zhí)行循環(huán)、記憶管理依賴模型支持 OpenAI 兼容接口的 LLM如 DeepSeek、GPT 系列等開發(fā)語言Python 3.10啟動方式命令行啟動可封裝為 API 服務(wù)是否支持 CPU取決于所選 LLM 是云端 API 還是本地模型顯存需求若調(diào)用云端 API 則本機基本無顯存壓力若本地部署 7B/14B 模型需按模型參數(shù)量預(yù)留顯存是否支持批量任務(wù)支持可通過并發(fā)隊列批量執(zhí)行是否支持 API支持可將 Agent 封裝為 HTTP 服務(wù)適合場景自動化測試、內(nèi)容生成、信息檢索、數(shù)據(jù)分析、內(nèi)部工具串聯(lián)2. 適用場景與使用邊界自己搭建 Agent 工具鏈意味著你不受某個框架綁定可以在核心循環(huán)上做任意定制。比較典型的使用場景包括自動化測試Agent 根據(jù)需求描述自動生成測試用例、調(diào)用測試工具、分析失敗日志。內(nèi)容生成管線Agent 調(diào)用搜索工具、資料庫、寫作模型完成從資料收集到成稿的全流程。數(shù)據(jù)報表分析Agent 連接數(shù)據(jù)庫工具、代碼執(zhí)行工具按用戶提問生成分析結(jié)果。內(nèi)部工具串聯(lián)把公司內(nèi)部 API 注冊成工具讓 Agent 按權(quán)限調(diào)用。但 Agent 不是萬能的。它不適合以下場景對實時性要求極高的系統(tǒng)模型推理耗時疊加工具調(diào)用耗時可能達到數(shù)秒甚至數(shù)十秒。完全離線、不允許任何外部 API 調(diào)用的內(nèi)網(wǎng)環(huán)境如果用云端大模型數(shù)據(jù)流轉(zhuǎn)到第三方需要嚴格評估。高風(fēng)險決策場景醫(yī)療診斷、金融交易等Agent 的幻覺風(fēng)險不可接受。合規(guī)邊界必須提前確認所有工具調(diào)用要遵循最小權(quán)限原則Agent 只能訪問它完成任務(wù)所必需的資源。涉及用戶隱私數(shù)據(jù)、業(yè)務(wù)數(shù)據(jù)的場景必須先獲得合法授權(quán)且數(shù)據(jù)處理鏈路要符合相關(guān)法規(guī)。不能把 Agent 用于繞過安全限制、攻擊系統(tǒng)、竊取賬號或批量爬取他人數(shù)據(jù)。如果 Agent 生成的內(nèi)容對外發(fā)布或商用必須做人工復(fù)核。3. 環(huán)境準備與前置條件3.1 開發(fā)環(huán)境檢查清單在開始寫代碼之前先確認以下環(huán)境是否就緒項目建議要求說明操作系統(tǒng)Windows 10/11、macOS、Linux核心代碼跨平臺Python3.10 或更高建議使用 3.11性能和兼容性均衡依賴管理venv 或 conda建議每個項目隔離虛擬環(huán)境LLM APIOpenAI 兼容接口的 API KeyDeepSeek、GPT 等均可網(wǎng)絡(luò)可正常訪問 API 服務(wù)如果部署本地模型則不受此限制磁盤空間至少 5GB 空閑含依賴和日志空間端口避免 8000、8080、7860 被占用如做 API 服務(wù)需提前確認3.2 創(chuàng)建項目目錄mkdir agent-toolchain cd agent-toolchain python -m venv venv source venv/bin/activate # Windows 下則執(zhí)行 venv\Scripts\activate3.3 安裝基礎(chǔ)依賴pip install openai python-dotenv pydantic requests rich說明openai官方 Python SDK 可以調(diào)用 OpenAI 兼容接口DeepSeek 等國產(chǎn)模型也支持同一協(xié)議。python-dotenv用于管理環(huán)境變量避免把 API Key 寫死在代碼里。pydantic做數(shù)據(jù)校驗和工具參數(shù) schema 校驗。rich用于終端日志輸出美化方便觀察 Agent 執(zhí)行過程。4. Agent 工具鏈的總體架構(gòu)手寫 Agent 工具鏈最少需要六個模塊用戶輸入 → 規(guī)劃器 → 工具調(diào)度 → 執(zhí)行器 → 反饋處理器 → 輸出結(jié)果 ↑______ 記憶模塊 ←______|4.1 六個核心模塊規(guī)劃器Planner接收用戶目標由 LLM 決定下一步行動?？赡懿鸾鉃槎嗖饺蝿?wù)也可能直接選擇工具。工具注冊表Tool RegistryAgent 可調(diào)用的所有工具的元信息集合。每個工具包含名稱、描述、參數(shù) schema、執(zhí)行函數(shù)。LLM 根據(jù)描述決定調(diào)哪個工具。執(zhí)行器Executor實際執(zhí)行工具函數(shù)的代碼邏輯。注意工具函數(shù)在本地運行LLM 只負責(zé)給出工具名和參數(shù)不直接執(zhí)行。反饋處理器Feedback Handler把工具執(zhí)行結(jié)果返回給 LLM讓模型判斷下一步動作。記憶模塊Memory Module短期記憶保留當(dāng)前任務(wù)上下文長期記憶按需存儲歷史事實和偏好。輸出解析器Output Parser把 LLM 的響應(yīng)解析為結(jié)構(gòu)化指令。這是最容易出錯的一環(huán)也是最需要自定義的地方。4.2 一次完整執(zhí)行流程第 1 輪用戶輸入查詢北京今天的天氣 第 2 輪規(guī)劃器決定調(diào)用 weather.get_current_weather 工具參數(shù) city北京 第 3 輪執(zhí)行器本地執(zhí)行天氣查詢函數(shù)返回北京今天晴22 度 第 4 輪反饋處理器將天氣結(jié)果返回給 LLM 第 5 輪LLM 根據(jù)結(jié)果生成最終回答這套循環(huán)就是 Agent 的“ReAct 模式”Reasoning推理 Acting行動。整個 Agent 引擎的核心就是把這個循環(huán)跑穩(wěn)。5. 從零實現(xiàn) Agent 核心代碼5.1 定義工具基類工具是整個 Agent 的能力來源。設(shè)計成裝飾器注冊模式最靈活# tools.py import inspect from typing import Any, Callable, Dict, List class Tool: def __init__(self, name: str, description: str, func: Callable, parameters: Dict[str, Any]): self.name name self.description description self.func func self.parameters parameters def run(self, **kwargs): return self.func(**kwargs) def to_openai_schema(self) - Dict[str, Any]: return { type: function, function: { name: self.name, description: self.description, parameters: self.parameters, } } _TOOL_REGISTRY: Dict[str, Tool] {} def register_tool(name: str, description: str, parameters: Dict[str, Any]): def decorator(func): _TOOL_REGISTRY[name] Tool( namename, descriptiondescription, funcfunc, parametersparameters, ) print(f[Tool Registry] 已注冊工具: {name}) return func return decorator def get_tool(name: str) - Tool: if name not in _TOOL_REGISTRY: raise KeyError(f工具不存在: {name}) return _TOOL_REGISTRY[name] def list_tools() - List[Tool]: return list(_TOOL_REGISTRY.values())這里的關(guān)鍵設(shè)計點每個工具都有標準化的to_openai_schema()方法可以直接轉(zhuǎn)換成 OpenAI 函數(shù)調(diào)用協(xié)議需要的 JSON Schema。5.2 定義示例工具下面注冊兩個最常用工具加法和查詢時間。# main.py import datetime from tools import register_tool, list_tools register_tool( namecalculator.add, description計算兩數(shù)相加, parameters{ type: object, properties: { x: {type: number, description: 第一個數(shù)字}, y: {type: number, description: 第二個數(shù)字} }, required: [x, y] } ) def add(a: float, b: float) - float: return a b register_tool( namesystem.current_time, description獲取當(dāng)前系統(tǒng)時間, parameters{ type: object, properties: {} } ) def current_time() - str: return datetime.datetime.now().strftime(%Y-%m-%d %H:%M:%S) if __name__ __main__: tools list_tools() print(f當(dāng)前注冊了 {len(tools)} 個工具)運行一次確認工具注冊正常這是后續(xù)所有邏輯的地基。5.3 實現(xiàn) LLM 客戶端封裝封裝一個ChatClient使用 openai SDK 調(diào)用 OpenAI 兼容接口。# llm_client.py import os from openai import OpenAI from dotenv import load_dotenv load_dotenv() class ChatClient: def __init__(self, model: str deepseek-chat): self.client OpenAI( api_keyos.getenv(LLM_API_KEY), base_urlos.getenv(LLM_BASE_URL, https://api.deepseek.com/v1), ) self.model model def chat(self, messages: list, tools: list None) - dict: kwargs {model: self.model, messages: messages} if tools: kwargs[tools] tools kwargs[tool_choice] auto response self.client.chat.completions.create(**kwargs) message response.choices[0].message return { content: message.content, tool_calls: message.tool_calls, }這里有幾個細節(jié)base_url配置成環(huán)境變量方便切換不同模型服務(wù)商。tool_choiceauto讓模型自主決定調(diào)用哪個工具。返回的tool_calls就是模型給出的“工具調(diào)用指令”。5.4 實現(xiàn) Agent 執(zhí)行循環(huán)這是整套工具鏈最核心的部分。執(zhí)行循環(huán)要做的事發(fā)送消息給 LLM。判斷是否有工具調(diào)用。如果有執(zhí)行工具將結(jié)果追加到消息記錄返回 LLM。如果沒有工具調(diào)用將 content 作為最終回答返回。# agent.py import json from llm_client import ChatClient from tools import get_tool, list_tools class Agent: def __init__(self, system_prompt: str, max_iters: int 10): self.client ChatClient() self.system_prompt system_prompt self.max_iters max_iters self.messages [ {role: system, content: self.system_prompt} ] def run(self, user_input: str) - str: self.messages.append({role: user, content: user_input}) tool_schemas [tool.to_openai_schema() for tool in list_tools()] for step in range(self.max_iters): print(f\n Step {step 1} ) result self.client.chat(self.messages, toolstool_schemas) # 情況一模型沒有發(fā)起工具調(diào)用直接返回結(jié)果 if not result[tool_calls]: answer result[content] or 模型未返回內(nèi)容 self.messages.append({role: assistant, content: answer}) return answer # 情況二模型發(fā)起工具調(diào)用 assistant_msg {role: assistant, content: None} tool_calls [] for tc in result[tool_calls]: tool_calls.append({ id: tc.id, type: function, function: { name: tc.function.name, arguments: tc.function.arguments, } }) print(f[Agent] 調(diào)用工具: {tc.function.name} 參數(shù): {tc.function.arguments}) assistant_msg[tool_calls] tool_calls self.messages.append(assistant_msg) # 逐個執(zhí)行工具并把結(jié)果寫回消息記錄 for tc in tool_calls: func_name tc[function][name] args_str tc[function][arguments] try: args json.loads(args_str) tool_func get_tool(func_name) output tool_func.run(**args) output_str json.dumps(output, ensure_asciiFalse) except Exception as e: output_str f工具執(zhí)行出錯: {str(e)} print(f[Tool] {func_name} 返回: {output_str[:100]}) self.messages.append({ role: tool, tool_call_id: tc[id], content: output_str, }) raise TimeoutError(f超過最大執(zhí)行輪數(shù) {self.max_iters} 次任務(wù)中止)這個循環(huán)已經(jīng)能跑通最基礎(chǔ)的 Agent 能力。注意幾個容易踩的坑工具調(diào)用后必須追加role: tool的消息并且tool_call_id必須對上。執(zhí)行工具前要做 try-except防止單個工具報錯直接讓整個 Agent 崩潰。max_iters必須限制否則模型可能陷入無限循環(huán)。5.5 接入記憶模塊記憶由兩種短期記憶就是上面的self.messages上下文窗口長期記憶需要獨立存儲。簡單實現(xiàn)一個基于 JSON 文件的長期記憶# memory.py import json import os from datetime import datetime class MemoryStore: def __init__(self, memory_file: str memory.json): self.memory_file memory_file self._ensure_file() def _ensure_file(self): if not os.path.exists(self.memory_file): with open(self.memory_file, w, encodingutf-8) as f: json.dump({}, f, ensure_asciiFalse, indent2) def save(self, key: str, value: str): data self._load() data[key] { value: value, updated_at: datetime.now().isoformat() } self._write(data) def load(self, key: str) - str or None: data self._load() item data.get(key) return item[value] if item else None def _load(self): with open(self.memory_file, r, encodingutf-8) as f: return json.load(f) def _write(self, data): with open(self.memory_file, w, encodingutf-8) as f: json.dump(data, f, ensure_asciiFalse, indent2)在 Agent 類中把記憶模塊組合進去每次對話開始時檢查是否有歷史的偏好或事實可以注入 system prompt。這里不做太復(fù)雜的向量檢索先保證“能存能取”。6. Agent 接口 API 與批量任務(wù)6.1 將 Agent 封裝為 FastAPI 服務(wù)工具鏈搭好之后如果要接業(yè)務(wù)系統(tǒng)需要把 Agent 暴露成 HTTP API。用 FastAPI 是最快的方式pip install fastapi uvicorn# api_service.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from agent import Agent app FastAPI(titleAgent Toolchain API, version0.1.0) class AgentRequest(BaseModel): input: str session_id: str default class AgentResponse(BaseModel): output: str session_prompts {} app.post(/agent/run, response_modelAgentResponse) def run_agent(req: AgentRequest): try: if req.session_id not in session_prompts: session_prompts[req.session_id] ( 你是智能助手任務(wù)復(fù)雜時請拆解為多個步驟使用工具完成。 ) agent Agent(system_promptsession_prompts[req.session_id]) output agent.run(req.input) return AgentResponse(outputoutput) except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.get(/agent/tools) def list_agent_tools(): return {tools: [t.to_openai_schema() for t in list_tools()]} if __name__ __main__: import uvicorn uvicorn.run(app, host127.0.0.1, port8000)啟動命令uvicorn api_service:app --host 127.0.0.1 --port 8000啟動后可以直接訪問http://127.0.0.1:8000/docs查看 Swagger 接口文檔。接口可以跑通后面就可以把 Agent 能力接到自己的業(yè)務(wù)工具里了。6.2 curl 調(diào)用示例curl -X POST http://127.0.0.1:8000/agent/run \ -H Content-Type: application/json \ -d {input: 請計算 123 456然后告訴我結(jié)果, session_id: test-001}預(yù)期返回{ output: 123 456 的結(jié)果是 579。 }6.3 Python 批量任務(wù)調(diào)度Agent 服務(wù)化之后批量任務(wù)就是并發(fā)調(diào)用的問題。寫一個簡單的批量調(diào)度器# batch_runner.py import json import requests from concurrent.futures import ThreadPoolExecutor, as_completed API_URL http://127.0.0.1:8000/agent/run def run_one(input_text: str, session_id: str) - dict: payload {input: input_text, session_id: session_id} try: resp requests.post(API_URL, jsonpayload, timeout120) resp.raise_for_status() return {ok: True, input: input_text, data: resp.json()} except Exception as e: return {ok: False, input: input_text, error: str(e)} def batch_run(tasks: list[dict], max_workers: int 4): results [] with ThreadPoolExecutor(max_workersmax_workers) as executor: futures [ executor.submit(run_one, task[input], task.get(session_id, batch)) for task in tasks ] for future in as_completed(futures): results.append(future.result()) return results if __name__ __main__: tasks [ {input: 計算 1 1, session_id: batch-1}, {input: 計算 10 20, session_id: batch-2}, {input: 計算 100 200, session_id: batch-3}, ] results batch_run(tasks, max_workers3) for r in results: print(json.dumps(r, ensure_asciiFalse, indent2))批量任務(wù)的幾個要點max_workers不建議開太大因為 LLM API 通常有并發(fā)限制且大批量并發(fā)會導(dǎo)致延時不穩(wěn)。每個任務(wù)要有獨立 session_id避免上下文污染。單個任務(wù)必須設(shè)置超時時間否則某個任務(wù)卡住會占用線程池。批量任務(wù)建議先跑 3 到 5 條測試確認接口穩(wěn)定后再全量跑。7. 資源占用與性能觀察7.1 調(diào)用云端 API 的資源特點如果你用的是 DeepSeek、GPT 等云端 API本機資源占用非常小主要是 Python 進程內(nèi)存和網(wǎng)絡(luò)帶寬。一個并發(fā)數(shù)為 4 的批量調(diào)度進程內(nèi)存通常在 200MB 到 500MB 之間。這種方式的好處是顯存無壓力劣勢是每次調(diào)用都要付費、有網(wǎng)絡(luò)延遲并且存在數(shù)據(jù)出域的合規(guī)風(fēng)險。7.2 本地部署 LLM 的資源特點如果要完全本地部署需要一個支持 OpenAI 兼容接口的推理服務(wù)資源占用取決于模型參數(shù)量。一個 7B 參數(shù)的量化模型通常需要 6GB 到 10GB 顯存14B 模型需要 15GB 以上70B 模型則需要多張顯卡或者大內(nèi)存配合 CPU 推理。實際顯存占用必須以你選擇的模型和推理框架為準不同量化級別差異很大。7.3 如何觀察資源占用開發(fā)調(diào)試時可以用以下命令觀察資源# Linux/macOS top -p pid # Python 腳本內(nèi)查看內(nèi)存 import psutil process psutil.Process() print(process.memory_info().rss / 1024 / 1024, MB)7.4 降低資源占用的策略Agent 循環(huán)中消息會不斷累積建議對歷史消息做截斷只保留最近 N 輪。工具返回內(nèi)容如果很大先做截斷再塞回上下文。并發(fā)批量任務(wù)時控制max_workers避免同時創(chuàng)建大量 HTTP 連接。本地部署推理時啟用 KV Cache 量化、Flash Attention 等特性但這需要推理框架支持。8. 常見問題與排查方法問題現(xiàn)象可能原因排查方式解決方案調(diào)用 API 返回 401API Key 錯誤或未加載打印環(huán)境變量確認.env文件存在重新配置LLM_API_KEYAPI 返回 429觸發(fā)限流查看 API 返回錯誤碼降低并發(fā)數(shù)加入重試退避機制Agent 不調(diào)用任何工具工具 schema 不合法或模型不認為需要工具打印 tool_schemas 檢查格式修正 JSON Schema調(diào)低要求顯式在 prompt 中提示模型使用工具工具執(zhí)行報錯參數(shù)類型不匹配打印 tool_calls 原始 arguments在工具函數(shù)內(nèi)部做類型轉(zhuǎn)換和默認值兜底tool_call_id不匹配assistant 消息和 tool 消息未正確對應(yīng)檢查消息記錄順序嚴格按“assistant 聲明 tool_calls”后逐個追加 tool 響應(yīng)Agent 陷入無限循環(huán)上下文太長或工具返回內(nèi)容讓模型誤解查看日志判斷重復(fù)行為設(shè)置 max_iters 上限讓工具返回內(nèi)容更清晰批量任務(wù)部分失敗網(wǎng)絡(luò)抖動或 API 限流查看失敗任務(wù)的錯誤字段加入超時、重試和失敗記錄內(nèi)存持續(xù)增長工具執(zhí)行中對象未釋放或消息列表過長監(jiān)控 RSS 內(nèi)存變化限制單次任務(wù)的消息數(shù)及時清理大對象端口被占用8000 被其他服務(wù)占用lsof -i :8000或netstat -ano換端口啟動8.1 一個典型的 API 重試封裝對于 429 和網(wǎng)絡(luò)錯誤推薦在調(diào)用層加指數(shù)退避重試import time import random def request_with_retry(func, max_retries: int 3, base_delay: float 2.0): for attempt in range(max_retries): try: return func() except Exception as e: if attempt max_retries - 1: raise e delay base_delay * (2 ** attempt) random.uniform(0, 1) print(f請求失敗{delay:.1f}s 后重試第 {attempt 1} 次: {e}) time.sleep(delay)9. 最佳實踐與使用建議9.1 先小后大先慢后快第一次跑通 Agent 時先把 max_iters 設(shè)為 3工具只注冊一個。等核心循環(huán)穩(wěn)定了再逐步加工具、加并發(fā)、加記憶。9.2 保持最小可運行配置建議把“單工具調(diào)用”的用例固化為一個自動化測試。每次改動核心循環(huán)后先跑這個用例確認沒有回歸。9.3 工具函數(shù)做防御式編程Agent 傳入的參數(shù)來自 LLM 生成非常不可控。工具函數(shù)內(nèi)部不要信任輸入盡量加類型校驗、長度限制、空值兜底。比如日期工具要處理非法日期字符串文件工具要限制路徑不能越界。9.4 日志和可觀測性Agent 的調(diào)試難度比普通程序高因為中間多了一層模型推理不確定性。日志至少包含每次 LLM 調(diào)用的耗時和 token 數(shù)。工具調(diào)用名稱、參數(shù)、返回內(nèi)容摘要。消息上下文長度變化。錯誤堆棧和重試記錄。建議以結(jié)構(gòu)化 JSON 落盤方便事后排查。9.5 安全邊界服務(wù)化后不要把 Agent 暴露到公網(wǎng)。如果必須暴露一定要加鑒權(quán)。工具的權(quán)限限制在最小范圍內(nèi)。比如 Agent 能調(diào)用數(shù)據(jù)庫查詢工具就只能用只讀賬號而不是管理員賬號。凡是涉及程序執(zhí)行、文件讀寫、命令行的工具必須做非常嚴格的輸入校驗和沙箱限制。涉及人臉、聲音、版權(quán)素材、第三方數(shù)據(jù)的內(nèi)容需要確認授權(quán)后方可使用和分發(fā)。9.6 成本控制云端 API 模式下Agent 每輪任務(wù)可能調(diào)用多次 LLM。一個多步任務(wù)可能消耗幾千到幾萬 token。批量任務(wù)上線前建議先用一條樣本估算 token 消耗。給每個會話設(shè)置 token 預(yù)算。對過長的工具返回內(nèi)容做截斷。保留每次調(diào)用的 token 使用記錄按天匯總。10. 總結(jié)與下一步這套從零搭建的 Agent 工具鏈核心價值在于學(xué)會了“執(zhí)行循環(huán)”的底層機制。真正把message - tool_calls - tool result - message這條閉環(huán)跑通之后你再去使用 LangChain、LangGraph 這類框架會清楚每個抽象層在做的事。先別急著加復(fù)雜功能。把最常見的問題驗證完單工具調(diào)用、多工具連續(xù)調(diào)用、算數(shù)計算、時間查詢跑通這四條就說明核心引擎是可用的。最容易踩的坑集中在工具參數(shù)解析和消息序列管理上這兩個地方多花時間打磨。后面值得繼續(xù)擴展的方向加入基于向量數(shù)據(jù)庫的長期記憶。加入多 Agent 協(xié)作機制比如規(guī)劃 Agent 和執(zhí)行 Agent 分離。加入人工確認環(huán)節(jié)當(dāng)工具調(diào)用風(fēng)險較高時先暫停等用戶確認再執(zhí)行。加入評測集自動回歸測試 Agent 在固定問題集上的穩(wěn)定性。Agent 開發(fā)的核心不在于堆疊多少模型而在于把工具鏈的每個環(huán)節(jié)設(shè)計得可控、可觀測、可恢復(fù)。從最小實現(xiàn)開始迭代是能找到瓶頸最快的方式。建議收藏備用動手寫第一版循環(huán)的時候照著這份架構(gòu)對照著做會比自己摸索快很多。