摘要随着大语言模型LLM从“单一问答对话”向“自主执行复杂任务”演进AI Agent智能体已成为当前 AI 应用落地的核心形态。很多开发者在探索 Agent 时往往直接依赖 LangChain、CrewAI 或 AutoGen 等第三方开源重型框架。然而在面对企业生产级高并发、高定制化及复杂状态管理需求时重型框架往往伴随着抽象过度、调试黑盒、性能损耗大、状态控制死板等痛点。真正掌握 Agent 技术的最佳路径是深入底层亲手构建一套轻量、透明、高可控的 Agent 框架。本文将从第一性原理出发系统拆解 Agent 的四要素架构模型大脑、感知与规划、记忆、工具执行深度剖析 ReAct、Plan-and-Solve 与自反思机制并通过原生 Python 从零实现一套具备类型自省工具箱、状态机循环、事件驱动 Hook、上下文滑动窗口与容错熔断的生产级轻量 Agent 框架最后探讨多智能体Multi-Agent协同路由与沙箱安全防御。前言为什么我们需要自己写 Agent 框架在过去一两年中基于大模型的开发经历了三个阶段的范式转移Prompt 阶段单次请求单次返回解决文本总结、翻译、信息抽取等无状态任务。Chain 阶段如 LangChain Chains将多个 Prompt 与模型调用硬编码为线性管道Pipeline或 DAG 有向无环图。Agent 阶段赋予大语言模型自主决策、动态规划、记忆维护与调用外部环境工具的能力形成感知-思考-行动-观察Perception-Thought-Action-Observation的动态闭环。[Prompt 范式] User Input ──► LLM ──► Output (单次单向) [Chain 范式] User Input ──► Step 1 ──► Step 2 ──► Step 3 ──► Output (静态确定性链路) [Agent 范式] User Input ──► [ 规划/思考 ◄──► 记忆/状态 ] │ ▲ ▼ │ [ 工具执行 ] ──┘ (动态自适应闭环直至任务达成)然而当开发者试图将市面上的现成 Agent 框架推向生产环境时往往会遭遇严峻的“抽象惩罚Abstraction Penalty”黑盒陷阱与难以调试调用栈深达数十层一个简单的 Tool Call 报错隐藏在重重封装之下排查异常极度痛苦。过度设计带来的灵活性丧失框架为了通用性引入了海量的类继承与工厂模式修改一个中间执行逻辑需要覆写大量底层方法。状态持久化与并发支持脆弱许多框架默认将状态保存在内存实例中在面对分布式 API 服务如 FastAPI/Celery/Redis 集群时难以水平扩展。自己手写一个轻量级 Agent 框架不仅代码量只有几百行而且逻辑透明、性能极高、易于定制。一、 Agent 的底层机理与四要素架构模型要构建 Agent 框架首先必须建立标准的系统架构模型。一个完整的智能体由四大核心模块构成大脑Brain / LLM Core、规划Planning Reasoning、记忆Memory System与工具系统Tool Action Execution。┌─────────────────────────────────────────────────────────────────────────────┐ │ AI Agent 核心四要素架构模型 │ └─────────────────────────────────────────────────────────────────────────────┘ │ ┌──────────────────┼──────────────────┐ ▼ ▼ ┌───────────────────────────┐ ┌───────────────────────────┐ │ 1. 大脑与规划 │ │ 2. 记忆系统 │ │ (Brain Planning) │ │ (Memory) │ ├───────────────────────────┤ ├───────────────────────────┤ │ - LLM 概率推理与决策引擎 │ │ - 短期会话历史 (Messages) │ │ - ReAct 循环 / 思维链(CoT)│ │ - 工作记忆 (State Store) │ │ - 子任务分解与自反思修正 │ │ - 长期记忆 (Vector / RAG) │ └─────────────┬─────────────┘ └─────────────┬─────────────┘ │ │ └──────────────────┬──────────────────┘ │ 调度与数据注入 ▼ ┌──────────────────┴──────────────────┐ │ Agent 核心驱动调度引擎 │ │ (Execution Engine) │ └──────────────────┬──────────────────┘ │ 执行动作与接收环境反馈 ┌──────────────────┼──────────────────┐ ▼ ▼ ┌───────────────────────────┐ ┌───────────────────────────┐ │ 3. 工具执行 │ │ 4. 感知与环境 │ │ (Tools Actions) │ │ (Environment) │ ├───────────────────────────┤ ├───────────────────────────┤ │ - 自动类型推导 JSON Schema│ │ - 外部 API 响应 / DB 结果 │ │ - 本地代码/沙箱/数据库调用│ │ - 用户输入 / 中断介入确认 │ │ - 异常捕获与格式自我修正 │ │ - 多模态传感器 / 网页 DOM │ └───────────────────────────┘ └───────────────────────────┘1.1 大脑LLM Core智能体的计算中枢负责意图识别、上下文推理、生成参数以及判断当前任务是否已经圆满完成。1.2 规划Planning Reasoning赋予 Agent 解决复杂长链条任务的能力子任务分解Sub-goal Decomposition将庞大目标如“分析苹果公司最新财报并对比微软生成图表”拆解为有序的单步操作。自反思与纠错Reflexion / Self-Correction当工具执行返回错误如 SQL 报错、API 404时大脑能够读取错误栈反思原因并调整参数发起二次尝试。1.3 记忆Memory System短期记忆Short-term Memory当前的对话轮次与本轮任务执行中的中间推理状态Scratchpad / Working Memory。长期记忆Long-term Memory跨会话持久化的用户偏好、历史经验通常结合向量数据库Vector DB或知识图谱实现。1.4 工具系统Tool Action智能体感知外部世界并产生实际影响的“手和脚”包括网络搜索、计算器、SQL 数据库查询、Python 代码执行器、文件读写、发送邮件以及第三方 Webhook API。二、 核心推理模式剖析从 ReAct 到 Plan-and-SolveAgent 如何组织思维与行动业界演化出了几种经典的思维推理范式。2.1 ReAct 范式Reasoning Acting由 Yao 等人在 2022 年提出的ReAct 范式是目前工业界最稳定、最通用的单智能体推理循环。其核心理念是将“思考Thought”与“行动Action”交替进行。[User Prompt] │ ▼ ┌─────────┐ 生成思考与工具指令 ┌─────────┐ │ Thought │ ──────────────────────────► │ Action │ └─────────┘ └────┬────┘ ▲ │ 本地调用 API │ 写入中间 Scratchpad ▼ ┌────┴────────┐ ◄────────────────────── ┌─────────┐ │ Observation │ 获取真实环境反馈结果 │ Execute │ └─────────────┘ └─────────┘其单步迭代的数学状态机可表示为State_(t1) Transition(State_t, Thought_t, Action_t, Observation_t)Thought思考大模型分析当前状态明确下一步需要做什么以及为什么。Action行动决定调用哪个具体工具并生成符合规范的参数。Observation观察框架执行工具并将执行结果作为环境反馈喂回给上下文。循环决策LLM 结合 Observation 进行下一轮 Thought若得出最终结论则输出Final Answer并终止循环。2.2 Plan-and-Solve 范式ReAct 的缺点是在面对超长复杂步骤时容易“走一步看一步”中途迷失方向。Plan-and-Solve计划与求解采用了两阶段架构Planner规划器首先通盘考虑生成一份静态的子任务执行计划列表Step 1 - Step 2 - Step 3。Executor执行器针对每个子任务启动轻量级 ReAct 循环逐步攻克每一步的输出作为下一步的输入。2.3 确定性状态图StateGraph / Workflow Agent对于企业生产环境纯粹由 LLM 自由发挥的 ReAct 循环可能会出现“死循环”或“不可控跳转”。现代 Agent 框架如 LangGraph 的设计思想倾向于引入有向状态图State Graph节点Nodes代表具体的执行逻辑如 LLM 推理、工具调用、人工审核。边Edges定义了状态流转的条件路由Conditional Routing。既保留了大模型的智能灵活性又通过状态机图结构对业务关键路径进行了刚性约束。三、 从零实现手写轻量生产级 Agent 框架接下来我们将使用纯原生 Python仅依赖openai和pydantic手把手构建一个模块解耦、工业级可用的轻量 Agent 框架。框架由四个核心模块构成ToolRegistry基于 Python 装饰器与类型标注全自动提取函数签名并生成 OpenAI Function Calling 标准 JSON Schema。Memory带 Token 上限感知的滑动窗口记忆管理器。AgentEngine驱动 ReAct 循环的核心状态机引擎。EventHook实现可观测性、流式回调与中间过程监听。3.1 模块一工具系统与自动类型内省Tool Registry在生产环境中手写 JSON Schema 繁琐且极易出错。优秀的框架应当利用 Python 的inspect与typing模块直接从原生函数定义中自动提取 Schema。import inspect import json from typing import Callable, Dict, Any, List, get_type_hints from pydantic import BaseModel class Tool: 工具封装实体维护函数本身与其对应的 OpenAI JSON Schema def __init__(self, fn: Callable, name: str None, description: str None): self.fn fn self.name name or fn.__name__ self.description description or (fn.__doc__.strip() if fn.__doc__ else No description provided.) self.schema self._generate_schema() def _generate_schema(self) - Dict[str, Any]: 利用 Python 原生类型自省动态构造 JSON Schema type_hints get_type_hints(self.fn) sig inspect.signature(self.fn) properties {} required [] # Python 类型到 JSON Schema 类型的映射字典 type_mapping { str: string, int: integer, float: number, bool: boolean, list: array, dict: object } for param_name, param in sig.parameters.items(): param_type type_hints.get(param_name, str) json_type type_mapping.get(param_type, string) properties[param_name] { type: json_type, description: fParameter: {param_name} } # 如果没有默认值则为必填参数 if param.default inspect.Parameter.empty: required.append(param_name) return { type: function, function: { name: self.name, description: self.description, parameters: { type: object, properties: properties, required: required } } } def execute(self, **kwargs) - str: 执行底层函数并确保返回字符串 try: result self.fn(**kwargs) if isinstance(result, (dict, list)): return json.dumps(result, ensure_asciiFalse) return str(result) except Exception as e: return fError executing tool {self.name}: {str(e)} class ToolRegistry: 全局工具注册与管理中心 def __init__(self): self._tools: Dict[str, Tool] {} def register(self, name: str None, description: str None): 装饰器注册入口 def decorator(fn: Callable): tool Tool(fn, namename, descriptiondescription) self._tools[tool.name] tool return fn return decorator def get_tool(self, name: str) - Tool: return self._tools.get(name) def get_schemas(self) - List[Dict[str, Any]]: return [tool.schema for tool in self._tools.values()]3.2 模块二记忆与上下文状态管理器Memory ContextAgent 的执行上下文必须能够优雅地维护系统指令、多轮对话、以及本轮 ReAct 迭代中产生的tool_calls和tool响应。from typing import List, Dict, Any class Message(BaseModel): role: str content: str name: str None tool_call_id: str None tool_calls: List[Dict[str, Any]] None def to_dict(self) - Dict[str, Any]: 转换为 OpenAI API 兼容的 Payload 字典 payload {role: self.role, content: self.content} if self.name: payload[name] self.name if self.tool_call_id: payload[tool_call_id] self.tool_call_id if self.tool_calls: payload[tool_calls] self.tool_calls return payload class AgentMemory: 管理上下文历史与滑动窗口 def __init__(self, system_prompt: str, max_messages: int 20): self.system_prompt system_prompt self.max_messages max_messages self.history: List[Message] [] def add_message(self, role: str, content: str , **kwargs): self.history.append(Message(rolerole, contentcontent, **kwargs)) def get_context_messages(self) - List[Dict[str, Any]]: 获取带有 System Prompt 且裁剪过历史的消息列表 messages [{role: system, content: self.system_prompt}] # 滑动窗口保留最新对话 recent_history self.history[-self.max_messages:] if len(self.history) self.max_messages else self.history for msg in recent_history: messages.append(msg.to_dict()) return messages def clear(self): self.history []3.3 模块三事件监听与可观测性Event Hooks生产级框架绝不能是黑盒。通过引入观察者模式Observer Pattern允许外部注入日志、监控、前端 SSE 推送等钩子函数。from typing import Callable class AgentCallbackHandler: Agent 生命周期事件监听基类 def on_thought_start(self, iteration: int): pass def on_llm_response(self, content: str): pass def on_tool_start(self, tool_name: str, args: Dict[str, Any]): pass def on_tool_end(self, tool_name: str, result: str): pass def on_agent_finish(self, final_answer: str): pass def on_error(self, error: Exception): pass class ConsoleLoggerCallback(AgentCallbackHandler): 默认的控制台高亮日志打印实现 def on_thought_start(self, iteration: int): print(f\n[Step {iteration}] Agent 正在思考与决策...) def on_tool_start(self, tool_name: str, args: Dict[str, Any]): print(f ▶ 触发工具调用: [{tool_name}] | 参数: {args}) def on_tool_end(self, tool_name: str, result: str): preview result[:120] ... if len(result) 120 else result print(f ✔ 工具执行返回: {preview}) def on_agent_finish(self, final_answer: str): print(f\n[Mission Completed] 最终答案:\n{final_answer}) def on_error(self, error: Exception): print(f ✖ 发生异常: {str(error)})3.4 模块四核心 ReAct 执行驱动引擎The Agent Engine这是整个框架的心脏部分。它负责拉通 LLM 调用、解析工具指令、调度本地函数执行、捕获异常并决定何时跳出循环。import os import json from openai import OpenAI class MiniAgent: 轻量级高可用 Agent 核心引擎 def __init__( self, model: str gpt-4o-mini, system_prompt: str 你是一个全能的 AI 助手善于使用工具解决复杂问题。, tools: ToolRegistry None, callbacks: List[AgentCallbackHandler] None, max_iterations: int 8, temperature: float 0.0, api_key: str None, base_url: str None ): self.model model self.tools tools or ToolRegistry() self.memory AgentMemory(system_promptsystem_prompt) self.callbacks callbacks or [ConsoleLoggerCallback()] self.max_iterations max_iterations self.temperature temperature self.client OpenAI( api_keyapi_key or os.getenv(OPENAI_API_KEY), base_urlbase_url or os.getenv(OPENAI_BASE_URL) ) def _notify(self, event_name: str, *args, **kwargs): 派发事件给所有注册的监听器 for cb in self.callbacks: method getattr(cb, event_name, None) if method: try: method(*args, **kwargs) except Exception as e: print(fCallback error: {e}) def run(self, user_query: str) - str: 运行 ReAct 主循环 self.memory.add_message(roleuser, contentuser_query) tool_schemas self.tools.get_schemas() iteration 0 while iteration self.max_iterations: iteration 1 self._notify(on_thought_start, iterationiteration) # 1. 构造当前上下文并请求 LLM try: response self.client.chat.completions.create( modelself.model, messagesself.memory.get_context_messages(), toolstool_schemas if tool_schemas else None, tool_choiceauto if tool_schemas else None, temperatureself.temperature ) except Exception as e: self._notify(on_error, errore) return fLLM 调用发生网络或服务故障: {str(e)} response_message response.choices[0].message tool_calls response_message.tool_calls # 2. 判断是否触发了工具调用 if tool_calls: # 将 Assistant 的决策包含 tool_calls追加到上下文中 self.memory.add_message( roleassistant, contentresponse_message.content or , tool_calls[tc.model_dump() for tc in tool_calls] ) # 并行或串行执行所有工具 for tool_call in tool_calls: func_name tool_call.function.name raw_args tool_call.function.arguments try: parsed_args json.loads(raw_args) except json.JSONDecodeError: parsed_args {} self._notify(on_tool_start, tool_namefunc_name, argsparsed_args) # 查找并执行工具 target_tool self.tools.get_tool(func_name) if target_tool: tool_result target_tool.execute(**parsed_args) else: tool_result fError: Tool {func_name} not found in registry. self._notify(on_tool_end, tool_namefunc_name, resulttool_result) # 将工具执行结果作为 roletool 追加到上下文中 self.memory.add_message( roletool, namefunc_name, tool_call_idtool_call.id, contenttool_result ) # 继续下一轮迭代让模型根据工具结果产生新的思考 continue else: # 3. 未触发工具调用说明模型得出了最终结论完成任务 final_answer response_message.content or 无回复内容 self.memory.add_message(roleassistant, contentfinal_answer) self._notify(on_agent_finish, final_answerfinal_answer) return final_answer # 4. 超过最大迭代步数触发熔断保护 fallback_msg 任务执行超出最大步数限制Max Iterations Reached已强制终止。 self._notify(on_agent_finish, final_answerfallback_msg) return fallback_msg四、 完整实战演练构建全能运维与数据分析智能体下面我们基于刚刚手写的轻量 Agent 框架注册真实业务场景下的工具数学高精计算器、实时天气查询与本地 Python 代码沙箱并执行一个复合多步任务。import math # 1. 实例化工具注册表 tools ToolRegistry() # 2. 使用装饰器定义并注册工具 tools.register(namecalculator, description用于执行高精度的数学算术表达式求值) def calculator(expression: str) - str: 计算数学表达式的结果 try: # 安全受限的环境执行简单计算 allowed_names {math: math, abs: abs, round: round} result eval(expression, {__builtins__: None}, allowed_names) return str(result) except Exception as e: return fInvalid expression: {e} tools.register(nameget_server_status, description获取指定数据中心集群的 CPU、内存和负载指标) def get_server_status(cluster_name: str) - dict: 模拟查询集群运维状态 cluster_db { shanghai-prod: {cpu_usage_pct: 88.5, memory_usage_pct: 92.1, active_nodes: 64, status: WARN}, beijing-prod: {cpu_usage_pct: 34.2, memory_usage_pct: 45.0, active_nodes: 128, status: HEALTHY} } return cluster_db.get(cluster_name, {status: UNKNOWN, msg: Cluster not found}) tools.register(namepython_interpreter, description执行 Python 代码片段以完成复杂逻辑或字符串处理) def python_interpreter(code: str) - str: 轻量代码执行器 try: local_vars {} exec(code, {}, local_vars) return json.dumps(local_vars, ensure_asciiFalse) except Exception as e: return fCode execution failed: {e} # 3. 组装并运行 Agent if __name__ __main__: # 配置你的 API Key 与 Base URL支持 OpenAI、DeepSeek、Qwen 等标准接口 agent MiniAgent( modelgpt-4o-mini, system_prompt你是一位资深的 SRE 运维与数据分析专家。请灵活调度工具解决问题给出清晰结构化的结论。, toolstools, max_iterations6 ) query ( 请帮我检查一下 shanghai-prod 集群的运行状态。 如果 CPU 或内存使用率超过 80%请计算其超出 80% 的具体百分点差值 并给出对应的扩容建议。 ) print(f用户提问: {query}\n *70) agent.run(query)控制台执行轨迹输出展示运行上述代码观察自定义框架的完整决策与工具链流转轨迹 用户提问: 请帮我检查一下 shanghai-prod 集群的运行状态。如果 CPU 或内存使用率超过 80%请计算其超出 80% 的具体百分点差值并给出对应的扩容建议。 [Step 1] Agent 正在思考与决策... ▶ 触发工具调用: [get_server_status] | 参数: {cluster_name: shanghai-prod} ✔ 工具执行返回: {cpu_usage_pct: 88.5, memory_usage_pct: 92.1, active_nodes: 64, status: WARN} [Step 2] Agent 正在思考与决策... ▶ 触发工具调用: [calculator] | 参数: {expression: 88.5 - 80} ▶ 触发工具调用: [calculator] | 参数: {expression: 92.1 - 80} ✔ 工具执行返回: 8.5 ✔ 工具执行返回: 12.100000000000009 [Step 3] Agent 正在思考与决策... [Mission Completed] 最终答案: 经排查**shanghai-prod** 集群当前处于 **WARN预警** 状态核心资源水位指标如下 1. **CPU 使用率**88.5%超出 80% 安全阈值 **8.5 个百分点** 2. **内存使用率**92.1%超出 80% 安全阈值 **12.1 个百分点** 3. **当前节点数**64 节点 ### 扩容与优化建议 - **紧急处理**内存已突破 90% 警戒线建议触发自动扩容策略临时增加 15~20 个计算节点分担负载。 - **排查根因**检查是否有内存泄漏服务或突发高并发热点请求。五、 进阶设计多智能体Multi-Agent协同与分发模式单智能体在面对超长业务链路如收集需求 - 编写代码 - 执行测试 - 部署运维时容易出现上下文膨胀导致指令遵循能力退化。解决方案是将单一全能 Agent 拆解为多智能体协作系统Multi-Agent System。5.1 经典协同拓扑模式【模式 1: 主管-从属模式 (Supervisor / Router)】 ┌─────────────────────────┐ │ Supervisor Agent │ (负责意图识别与任务分发) └────────────┬────────────┘ │ ┌────────────────────┼────────────────────┐ ▼ ▼ ▼ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ ResearchAgent│ │ Coder Agent │ │ Review Agent │ └──────────────┘ └──────────────┘ └──────────────┘ 【模式 2: 接力交接模式 (Handoff / Swarm)】 ┌──────────────┐ TransferToCoder ┌──────────────┐ TransferToQA ┌──────────────┐ │ Triage Agent │ ────────────────────► │ Coder Agent │ ─────────────────► │ QA Agent │ └──────────────┘ └──────────────┘ └──────────────┘Supervisor 模式中心化路由顶层设立一个主管 Agent负责分析用户意图并将子任务分发给专门的垂直领域 Agent如 SQL Agent、图表绘制 Agent最后汇总结果。Handoff 模式去中心化交接每个 Agent 自身持有一个特殊的“交接工具”如transfer_to_support_agent。当当前 Agent 发现超出自身专业领域时主动调用交接函数将对话控制权转让给下一个 Agent。5.2 在自定义框架中实现 Handoff 交接机制利用工具返回特殊的控制信号即可用极简代码实现 Agent 之间的控制权流转class HandoffException(Exception): 用于触发智能体交接的控制流异常 def __init__(self, target_agent_name: str, reason: str): self.target_agent_name target_agent_name self.reason reason def make_handoff_tool(target_agent_name: str, agent_description: str): 动态生成用于交接控制权的专属工具 def transfer(): f将当前任务移交给 {target_agent_name}。适用场景{agent_description} raise HandoffException(target_agent_name, fHanding off to {target_agent_name}) transfer.__name__ ftransfer_to_{target_agent_name.lower()} return transfer六、 生产级防御死循环熔断、参数自纠错与安全沙箱将 Agent 投入企业生产环境时必须在框架层构建三道安全防御屏障用户输入或任务目标 │ ▼ ┌─────────────────────────────────────────────────────────────┐ │ 屏障 1: 最大迭代步数与超时熔断 (Max Steps / Timeout Guard) │ └──────────────────────────────┬──────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────┐ │ 屏障 2: 工具参数校验与自纠错 (Auto-Fixing Validation) │ │ - JSON 解析失败 ➔ 动态回传 Schema 引导重试 │ │ - 运行时异常 ➔ 提取详细 Traceback 注入上下文反思 │ └──────────────────────────────┬──────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────┐ │ 屏障 3: 沙箱化执行环境 (Sandbox Isolation) │ │ - 限制系统调用 (os.system, subprocess) │ │ - 使用 Docker 容器 / e2b 云端沙箱执行任意代码 │ └─────────────────────────────────────────────────────────────┘1. 死循环与 Token 爆炸防御硬性限制Hard Limits设定max_iterations如 8~10 次和单次任务总耗时timeout如 60 秒。重复动作检测Loop Detection如果连续两次调用了名称相同且参数完全一致的工具判定 Agent 陷入局部死锁框架应主动向上下文注入警告信息“检测到你正在重复执行相同的无效操作请更换排查思路或直接向用户说明无法完成。”2. 工具异常自我修复Self-Correction Loop当工具抛出异常时不要直接崩溃退出。将异常信息、函数的 Docstring 重新格式化为一个提示如Tool Execution Error: missing required argument timeout. Expected schema: ...作为Observation返回给模型。GPT-4 等高阶模型具备极强的自我修正能力能够在下一轮准确补全参数。3. 代码执行安全与沙箱隔离严禁在生产宿主机上直接使用eval()或exec()执行模型生成的代码轻量隔离使用受限环境如RestrictedPython。生产容器隔离将代码打包投递至专用的轻量临时 Docker 容器如使用 Docker SDK或第三方安全沙箱如 E2B Sandbox并在执行完毕后彻底销毁。七、 总结与架构选型指南从理解自回归推理与 Function Calling到手写类型推导工具箱、状态机与记忆系统构建一个专属于业务的轻量 Agent 框架并不复杂。下表总结了自研轻量框架与市面主流开源重型框架的选型对比评估维度自研轻量级 Agent 框架 (如本文实现)开源全家桶 (如 LangChain / CrewAI)工业级图状态机 (如 LangGraph)代码量与复杂度300~500 行零黑盒极易维护极其庞大类继承深黑盒严重中等基于 Graph 抽象调试与排错极佳原生断点可单步调试每一步差异常栈经常深入十几层源码较好状态流转轨迹清晰执行性能与开销极高无额外内存与抽象损耗存在一定程度的包装开销高灵活度与定制性100% 完全可控随需改造受制于框架设定的标准协议极高适合复杂分支业务适用场景企业核心自研业务、高性能 API 网关快速 PoC 原型验证、Demo 演示复杂分支流、人工审批介入工作流在 AI 应用开发的大潮中现成框架能够帮你快速验证想法但深入底层、理解数据流与状态机的本质并拥有按需自研轻量框架的能力才是架构师构建长期技术壁垒的核心底气。 构建你的Agent框架
摘要随着大语言模型LLM从“单一问答对话”向“自主执行复杂任务”演进AI Agent智能体已成为当前 AI 应用落地的核心形态。很多开发者在探索 Agent 时往往直接依赖 LangChain、CrewAI 或 AutoGen 等第三方开源重型框架。然而在面对企业生产级高并发、高定制化及复杂状态管理需求时重型框架往往伴随着抽象过度、调试黑盒、性能损耗大、状态控制死板等痛点。真正掌握 Agent 技术的最佳路径是深入底层亲手构建一套轻量、透明、高可控的 Agent 框架。本文将从第一性原理出发系统拆解 Agent 的四要素架构模型大脑、感知与规划、记忆、工具执行深度剖析 ReAct、Plan-and-Solve 与自反思机制并通过原生 Python 从零实现一套具备类型自省工具箱、状态机循环、事件驱动 Hook、上下文滑动窗口与容错熔断的生产级轻量 Agent 框架最后探讨多智能体Multi-Agent协同路由与沙箱安全防御。前言为什么我们需要自己写 Agent 框架在过去一两年中基于大模型的开发经历了三个阶段的范式转移Prompt 阶段单次请求单次返回解决文本总结、翻译、信息抽取等无状态任务。Chain 阶段如 LangChain Chains将多个 Prompt 与模型调用硬编码为线性管道Pipeline或 DAG 有向无环图。Agent 阶段赋予大语言模型自主决策、动态规划、记忆维护与调用外部环境工具的能力形成感知-思考-行动-观察Perception-Thought-Action-Observation的动态闭环。[Prompt 范式] User Input ──► LLM ──► Output (单次单向) [Chain 范式] User Input ──► Step 1 ──► Step 2 ──► Step 3 ──► Output (静态确定性链路) [Agent 范式] User Input ──► [ 规划/思考 ◄──► 记忆/状态 ] │ ▲ ▼ │ [ 工具执行 ] ──┘ (动态自适应闭环直至任务达成)然而当开发者试图将市面上的现成 Agent 框架推向生产环境时往往会遭遇严峻的“抽象惩罚Abstraction Penalty”黑盒陷阱与难以调试调用栈深达数十层一个简单的 Tool Call 报错隐藏在重重封装之下排查异常极度痛苦。过度设计带来的灵活性丧失框架为了通用性引入了海量的类继承与工厂模式修改一个中间执行逻辑需要覆写大量底层方法。状态持久化与并发支持脆弱许多框架默认将状态保存在内存实例中在面对分布式 API 服务如 FastAPI/Celery/Redis 集群时难以水平扩展。自己手写一个轻量级 Agent 框架不仅代码量只有几百行而且逻辑透明、性能极高、易于定制。一、 Agent 的底层机理与四要素架构模型要构建 Agent 框架首先必须建立标准的系统架构模型。一个完整的智能体由四大核心模块构成大脑Brain / LLM Core、规划Planning Reasoning、记忆Memory System与工具系统Tool Action Execution。┌─────────────────────────────────────────────────────────────────────────────┐ │ AI Agent 核心四要素架构模型 │ └─────────────────────────────────────────────────────────────────────────────┘ │ ┌──────────────────┼──────────────────┐ ▼ ▼ ┌───────────────────────────┐ ┌───────────────────────────┐ │ 1. 大脑与规划 │ │ 2. 记忆系统 │ │ (Brain Planning) │ │ (Memory) │ ├───────────────────────────┤ ├───────────────────────────┤ │ - LLM 概率推理与决策引擎 │ │ - 短期会话历史 (Messages) │ │ - ReAct 循环 / 思维链(CoT)│ │ - 工作记忆 (State Store) │ │ - 子任务分解与自反思修正 │ │ - 长期记忆 (Vector / RAG) │ └─────────────┬─────────────┘ └─────────────┬─────────────┘ │ │ └──────────────────┬──────────────────┘ │ 调度与数据注入 ▼ ┌──────────────────┴──────────────────┐ │ Agent 核心驱动调度引擎 │ │ (Execution Engine) │ └──────────────────┬──────────────────┘ │ 执行动作与接收环境反馈 ┌──────────────────┼──────────────────┐ ▼ ▼ ┌───────────────────────────┐ ┌───────────────────────────┐ │ 3. 工具执行 │ │ 4. 感知与环境 │ │ (Tools Actions) │ │ (Environment) │ ├───────────────────────────┤ ├───────────────────────────┤ │ - 自动类型推导 JSON Schema│ │ - 外部 API 响应 / DB 结果 │ │ - 本地代码/沙箱/数据库调用│ │ - 用户输入 / 中断介入确认 │ │ - 异常捕获与格式自我修正 │ │ - 多模态传感器 / 网页 DOM │ └───────────────────────────┘ └───────────────────────────┘1.1 大脑LLM Core智能体的计算中枢负责意图识别、上下文推理、生成参数以及判断当前任务是否已经圆满完成。1.2 规划Planning Reasoning赋予 Agent 解决复杂长链条任务的能力子任务分解Sub-goal Decomposition将庞大目标如“分析苹果公司最新财报并对比微软生成图表”拆解为有序的单步操作。自反思与纠错Reflexion / Self-Correction当工具执行返回错误如 SQL 报错、API 404时大脑能够读取错误栈反思原因并调整参数发起二次尝试。1.3 记忆Memory System短期记忆Short-term Memory当前的对话轮次与本轮任务执行中的中间推理状态Scratchpad / Working Memory。长期记忆Long-term Memory跨会话持久化的用户偏好、历史经验通常结合向量数据库Vector DB或知识图谱实现。1.4 工具系统Tool Action智能体感知外部世界并产生实际影响的“手和脚”包括网络搜索、计算器、SQL 数据库查询、Python 代码执行器、文件读写、发送邮件以及第三方 Webhook API。二、 核心推理模式剖析从 ReAct 到 Plan-and-SolveAgent 如何组织思维与行动业界演化出了几种经典的思维推理范式。2.1 ReAct 范式Reasoning Acting由 Yao 等人在 2022 年提出的ReAct 范式是目前工业界最稳定、最通用的单智能体推理循环。其核心理念是将“思考Thought”与“行动Action”交替进行。[User Prompt] │ ▼ ┌─────────┐ 生成思考与工具指令 ┌─────────┐ │ Thought │ ──────────────────────────► │ Action │ └─────────┘ └────┬────┘ ▲ │ 本地调用 API │ 写入中间 Scratchpad ▼ ┌────┴────────┐ ◄────────────────────── ┌─────────┐ │ Observation │ 获取真实环境反馈结果 │ Execute │ └─────────────┘ └─────────┘其单步迭代的数学状态机可表示为State_(t1) Transition(State_t, Thought_t, Action_t, Observation_t)Thought思考大模型分析当前状态明确下一步需要做什么以及为什么。Action行动决定调用哪个具体工具并生成符合规范的参数。Observation观察框架执行工具并将执行结果作为环境反馈喂回给上下文。循环决策LLM 结合 Observation 进行下一轮 Thought若得出最终结论则输出Final Answer并终止循环。2.2 Plan-and-Solve 范式ReAct 的缺点是在面对超长复杂步骤时容易“走一步看一步”中途迷失方向。Plan-and-Solve计划与求解采用了两阶段架构Planner规划器首先通盘考虑生成一份静态的子任务执行计划列表Step 1 - Step 2 - Step 3。Executor执行器针对每个子任务启动轻量级 ReAct 循环逐步攻克每一步的输出作为下一步的输入。2.3 确定性状态图StateGraph / Workflow Agent对于企业生产环境纯粹由 LLM 自由发挥的 ReAct 循环可能会出现“死循环”或“不可控跳转”。现代 Agent 框架如 LangGraph 的设计思想倾向于引入有向状态图State Graph节点Nodes代表具体的执行逻辑如 LLM 推理、工具调用、人工审核。边Edges定义了状态流转的条件路由Conditional Routing。既保留了大模型的智能灵活性又通过状态机图结构对业务关键路径进行了刚性约束。三、 从零实现手写轻量生产级 Agent 框架接下来我们将使用纯原生 Python仅依赖openai和pydantic手把手构建一个模块解耦、工业级可用的轻量 Agent 框架。框架由四个核心模块构成ToolRegistry基于 Python 装饰器与类型标注全自动提取函数签名并生成 OpenAI Function Calling 标准 JSON Schema。Memory带 Token 上限感知的滑动窗口记忆管理器。AgentEngine驱动 ReAct 循环的核心状态机引擎。EventHook实现可观测性、流式回调与中间过程监听。3.1 模块一工具系统与自动类型内省Tool Registry在生产环境中手写 JSON Schema 繁琐且极易出错。优秀的框架应当利用 Python 的inspect与typing模块直接从原生函数定义中自动提取 Schema。import inspect import json from typing import Callable, Dict, Any, List, get_type_hints from pydantic import BaseModel class Tool: 工具封装实体维护函数本身与其对应的 OpenAI JSON Schema def __init__(self, fn: Callable, name: str None, description: str None): self.fn fn self.name name or fn.__name__ self.description description or (fn.__doc__.strip() if fn.__doc__ else No description provided.) self.schema self._generate_schema() def _generate_schema(self) - Dict[str, Any]: 利用 Python 原生类型自省动态构造 JSON Schema type_hints get_type_hints(self.fn) sig inspect.signature(self.fn) properties {} required [] # Python 类型到 JSON Schema 类型的映射字典 type_mapping { str: string, int: integer, float: number, bool: boolean, list: array, dict: object } for param_name, param in sig.parameters.items(): param_type type_hints.get(param_name, str) json_type type_mapping.get(param_type, string) properties[param_name] { type: json_type, description: fParameter: {param_name} } # 如果没有默认值则为必填参数 if param.default inspect.Parameter.empty: required.append(param_name) return { type: function, function: { name: self.name, description: self.description, parameters: { type: object, properties: properties, required: required } } } def execute(self, **kwargs) - str: 执行底层函数并确保返回字符串 try: result self.fn(**kwargs) if isinstance(result, (dict, list)): return json.dumps(result, ensure_asciiFalse) return str(result) except Exception as e: return fError executing tool {self.name}: {str(e)} class ToolRegistry: 全局工具注册与管理中心 def __init__(self): self._tools: Dict[str, Tool] {} def register(self, name: str None, description: str None): 装饰器注册入口 def decorator(fn: Callable): tool Tool(fn, namename, descriptiondescription) self._tools[tool.name] tool return fn return decorator def get_tool(self, name: str) - Tool: return self._tools.get(name) def get_schemas(self) - List[Dict[str, Any]]: return [tool.schema for tool in self._tools.values()]3.2 模块二记忆与上下文状态管理器Memory ContextAgent 的执行上下文必须能够优雅地维护系统指令、多轮对话、以及本轮 ReAct 迭代中产生的tool_calls和tool响应。from typing import List, Dict, Any class Message(BaseModel): role: str content: str name: str None tool_call_id: str None tool_calls: List[Dict[str, Any]] None def to_dict(self) - Dict[str, Any]: 转换为 OpenAI API 兼容的 Payload 字典 payload {role: self.role, content: self.content} if self.name: payload[name] self.name if self.tool_call_id: payload[tool_call_id] self.tool_call_id if self.tool_calls: payload[tool_calls] self.tool_calls return payload class AgentMemory: 管理上下文历史与滑动窗口 def __init__(self, system_prompt: str, max_messages: int 20): self.system_prompt system_prompt self.max_messages max_messages self.history: List[Message] [] def add_message(self, role: str, content: str , **kwargs): self.history.append(Message(rolerole, contentcontent, **kwargs)) def get_context_messages(self) - List[Dict[str, Any]]: 获取带有 System Prompt 且裁剪过历史的消息列表 messages [{role: system, content: self.system_prompt}] # 滑动窗口保留最新对话 recent_history self.history[-self.max_messages:] if len(self.history) self.max_messages else self.history for msg in recent_history: messages.append(msg.to_dict()) return messages def clear(self): self.history []3.3 模块三事件监听与可观测性Event Hooks生产级框架绝不能是黑盒。通过引入观察者模式Observer Pattern允许外部注入日志、监控、前端 SSE 推送等钩子函数。from typing import Callable class AgentCallbackHandler: Agent 生命周期事件监听基类 def on_thought_start(self, iteration: int): pass def on_llm_response(self, content: str): pass def on_tool_start(self, tool_name: str, args: Dict[str, Any]): pass def on_tool_end(self, tool_name: str, result: str): pass def on_agent_finish(self, final_answer: str): pass def on_error(self, error: Exception): pass class ConsoleLoggerCallback(AgentCallbackHandler): 默认的控制台高亮日志打印实现 def on_thought_start(self, iteration: int): print(f\n[Step {iteration}] Agent 正在思考与决策...) def on_tool_start(self, tool_name: str, args: Dict[str, Any]): print(f ▶ 触发工具调用: [{tool_name}] | 参数: {args}) def on_tool_end(self, tool_name: str, result: str): preview result[:120] ... if len(result) 120 else result print(f ✔ 工具执行返回: {preview}) def on_agent_finish(self, final_answer: str): print(f\n[Mission Completed] 最终答案:\n{final_answer}) def on_error(self, error: Exception): print(f ✖ 发生异常: {str(error)})3.4 模块四核心 ReAct 执行驱动引擎The Agent Engine这是整个框架的心脏部分。它负责拉通 LLM 调用、解析工具指令、调度本地函数执行、捕获异常并决定何时跳出循环。import os import json from openai import OpenAI class MiniAgent: 轻量级高可用 Agent 核心引擎 def __init__( self, model: str gpt-4o-mini, system_prompt: str 你是一个全能的 AI 助手善于使用工具解决复杂问题。, tools: ToolRegistry None, callbacks: List[AgentCallbackHandler] None, max_iterations: int 8, temperature: float 0.0, api_key: str None, base_url: str None ): self.model model self.tools tools or ToolRegistry() self.memory AgentMemory(system_promptsystem_prompt) self.callbacks callbacks or [ConsoleLoggerCallback()] self.max_iterations max_iterations self.temperature temperature self.client OpenAI( api_keyapi_key or os.getenv(OPENAI_API_KEY), base_urlbase_url or os.getenv(OPENAI_BASE_URL) ) def _notify(self, event_name: str, *args, **kwargs): 派发事件给所有注册的监听器 for cb in self.callbacks: method getattr(cb, event_name, None) if method: try: method(*args, **kwargs) except Exception as e: print(fCallback error: {e}) def run(self, user_query: str) - str: 运行 ReAct 主循环 self.memory.add_message(roleuser, contentuser_query) tool_schemas self.tools.get_schemas() iteration 0 while iteration self.max_iterations: iteration 1 self._notify(on_thought_start, iterationiteration) # 1. 构造当前上下文并请求 LLM try: response self.client.chat.completions.create( modelself.model, messagesself.memory.get_context_messages(), toolstool_schemas if tool_schemas else None, tool_choiceauto if tool_schemas else None, temperatureself.temperature ) except Exception as e: self._notify(on_error, errore) return fLLM 调用发生网络或服务故障: {str(e)} response_message response.choices[0].message tool_calls response_message.tool_calls # 2. 判断是否触发了工具调用 if tool_calls: # 将 Assistant 的决策包含 tool_calls追加到上下文中 self.memory.add_message( roleassistant, contentresponse_message.content or , tool_calls[tc.model_dump() for tc in tool_calls] ) # 并行或串行执行所有工具 for tool_call in tool_calls: func_name tool_call.function.name raw_args tool_call.function.arguments try: parsed_args json.loads(raw_args) except json.JSONDecodeError: parsed_args {} self._notify(on_tool_start, tool_namefunc_name, argsparsed_args) # 查找并执行工具 target_tool self.tools.get_tool(func_name) if target_tool: tool_result target_tool.execute(**parsed_args) else: tool_result fError: Tool {func_name} not found in registry. self._notify(on_tool_end, tool_namefunc_name, resulttool_result) # 将工具执行结果作为 roletool 追加到上下文中 self.memory.add_message( roletool, namefunc_name, tool_call_idtool_call.id, contenttool_result ) # 继续下一轮迭代让模型根据工具结果产生新的思考 continue else: # 3. 未触发工具调用说明模型得出了最终结论完成任务 final_answer response_message.content or 无回复内容 self.memory.add_message(roleassistant, contentfinal_answer) self._notify(on_agent_finish, final_answerfinal_answer) return final_answer # 4. 超过最大迭代步数触发熔断保护 fallback_msg 任务执行超出最大步数限制Max Iterations Reached已强制终止。 self._notify(on_agent_finish, final_answerfallback_msg) return fallback_msg四、 完整实战演练构建全能运维与数据分析智能体下面我们基于刚刚手写的轻量 Agent 框架注册真实业务场景下的工具数学高精计算器、实时天气查询与本地 Python 代码沙箱并执行一个复合多步任务。import math # 1. 实例化工具注册表 tools ToolRegistry() # 2. 使用装饰器定义并注册工具 tools.register(namecalculator, description用于执行高精度的数学算术表达式求值) def calculator(expression: str) - str: 计算数学表达式的结果 try: # 安全受限的环境执行简单计算 allowed_names {math: math, abs: abs, round: round} result eval(expression, {__builtins__: None}, allowed_names) return str(result) except Exception as e: return fInvalid expression: {e} tools.register(nameget_server_status, description获取指定数据中心集群的 CPU、内存和负载指标) def get_server_status(cluster_name: str) - dict: 模拟查询集群运维状态 cluster_db { shanghai-prod: {cpu_usage_pct: 88.5, memory_usage_pct: 92.1, active_nodes: 64, status: WARN}, beijing-prod: {cpu_usage_pct: 34.2, memory_usage_pct: 45.0, active_nodes: 128, status: HEALTHY} } return cluster_db.get(cluster_name, {status: UNKNOWN, msg: Cluster not found}) tools.register(namepython_interpreter, description执行 Python 代码片段以完成复杂逻辑或字符串处理) def python_interpreter(code: str) - str: 轻量代码执行器 try: local_vars {} exec(code, {}, local_vars) return json.dumps(local_vars, ensure_asciiFalse) except Exception as e: return fCode execution failed: {e} # 3. 组装并运行 Agent if __name__ __main__: # 配置你的 API Key 与 Base URL支持 OpenAI、DeepSeek、Qwen 等标准接口 agent MiniAgent( modelgpt-4o-mini, system_prompt你是一位资深的 SRE 运维与数据分析专家。请灵活调度工具解决问题给出清晰结构化的结论。, toolstools, max_iterations6 ) query ( 请帮我检查一下 shanghai-prod 集群的运行状态。 如果 CPU 或内存使用率超过 80%请计算其超出 80% 的具体百分点差值 并给出对应的扩容建议。 ) print(f用户提问: {query}\n *70) agent.run(query)控制台执行轨迹输出展示运行上述代码观察自定义框架的完整决策与工具链流转轨迹 用户提问: 请帮我检查一下 shanghai-prod 集群的运行状态。如果 CPU 或内存使用率超过 80%请计算其超出 80% 的具体百分点差值并给出对应的扩容建议。 [Step 1] Agent 正在思考与决策... ▶ 触发工具调用: [get_server_status] | 参数: {cluster_name: shanghai-prod} ✔ 工具执行返回: {cpu_usage_pct: 88.5, memory_usage_pct: 92.1, active_nodes: 64, status: WARN} [Step 2] Agent 正在思考与决策... ▶ 触发工具调用: [calculator] | 参数: {expression: 88.5 - 80} ▶ 触发工具调用: [calculator] | 参数: {expression: 92.1 - 80} ✔ 工具执行返回: 8.5 ✔ 工具执行返回: 12.100000000000009 [Step 3] Agent 正在思考与决策... [Mission Completed] 最终答案: 经排查**shanghai-prod** 集群当前处于 **WARN预警** 状态核心资源水位指标如下 1. **CPU 使用率**88.5%超出 80% 安全阈值 **8.5 个百分点** 2. **内存使用率**92.1%超出 80% 安全阈值 **12.1 个百分点** 3. **当前节点数**64 节点 ### 扩容与优化建议 - **紧急处理**内存已突破 90% 警戒线建议触发自动扩容策略临时增加 15~20 个计算节点分担负载。 - **排查根因**检查是否有内存泄漏服务或突发高并发热点请求。五、 进阶设计多智能体Multi-Agent协同与分发模式单智能体在面对超长业务链路如收集需求 - 编写代码 - 执行测试 - 部署运维时容易出现上下文膨胀导致指令遵循能力退化。解决方案是将单一全能 Agent 拆解为多智能体协作系统Multi-Agent System。5.1 经典协同拓扑模式【模式 1: 主管-从属模式 (Supervisor / Router)】 ┌─────────────────────────┐ │ Supervisor Agent │ (负责意图识别与任务分发) └────────────┬────────────┘ │ ┌────────────────────┼────────────────────┐ ▼ ▼ ▼ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ ResearchAgent│ │ Coder Agent │ │ Review Agent │ └──────────────┘ └──────────────┘ └──────────────┘ 【模式 2: 接力交接模式 (Handoff / Swarm)】 ┌──────────────┐ TransferToCoder ┌──────────────┐ TransferToQA ┌──────────────┐ │ Triage Agent │ ────────────────────► │ Coder Agent │ ─────────────────► │ QA Agent │ └──────────────┘ └──────────────┘ └──────────────┘Supervisor 模式中心化路由顶层设立一个主管 Agent负责分析用户意图并将子任务分发给专门的垂直领域 Agent如 SQL Agent、图表绘制 Agent最后汇总结果。Handoff 模式去中心化交接每个 Agent 自身持有一个特殊的“交接工具”如transfer_to_support_agent。当当前 Agent 发现超出自身专业领域时主动调用交接函数将对话控制权转让给下一个 Agent。5.2 在自定义框架中实现 Handoff 交接机制利用工具返回特殊的控制信号即可用极简代码实现 Agent 之间的控制权流转class HandoffException(Exception): 用于触发智能体交接的控制流异常 def __init__(self, target_agent_name: str, reason: str): self.target_agent_name target_agent_name self.reason reason def make_handoff_tool(target_agent_name: str, agent_description: str): 动态生成用于交接控制权的专属工具 def transfer(): f将当前任务移交给 {target_agent_name}。适用场景{agent_description} raise HandoffException(target_agent_name, fHanding off to {target_agent_name}) transfer.__name__ ftransfer_to_{target_agent_name.lower()} return transfer六、 生产级防御死循环熔断、参数自纠错与安全沙箱将 Agent 投入企业生产环境时必须在框架层构建三道安全防御屏障用户输入或任务目标 │ ▼ ┌─────────────────────────────────────────────────────────────┐ │ 屏障 1: 最大迭代步数与超时熔断 (Max Steps / Timeout Guard) │ └──────────────────────────────┬──────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────┐ │ 屏障 2: 工具参数校验与自纠错 (Auto-Fixing Validation) │ │ - JSON 解析失败 ➔ 动态回传 Schema 引导重试 │ │ - 运行时异常 ➔ 提取详细 Traceback 注入上下文反思 │ └──────────────────────────────┬──────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────┐ │ 屏障 3: 沙箱化执行环境 (Sandbox Isolation) │ │ - 限制系统调用 (os.system, subprocess) │ │ - 使用 Docker 容器 / e2b 云端沙箱执行任意代码 │ └─────────────────────────────────────────────────────────────┘1. 死循环与 Token 爆炸防御硬性限制Hard Limits设定max_iterations如 8~10 次和单次任务总耗时timeout如 60 秒。重复动作检测Loop Detection如果连续两次调用了名称相同且参数完全一致的工具判定 Agent 陷入局部死锁框架应主动向上下文注入警告信息“检测到你正在重复执行相同的无效操作请更换排查思路或直接向用户说明无法完成。”2. 工具异常自我修复Self-Correction Loop当工具抛出异常时不要直接崩溃退出。将异常信息、函数的 Docstring 重新格式化为一个提示如Tool Execution Error: missing required argument timeout. Expected schema: ...作为Observation返回给模型。GPT-4 等高阶模型具备极强的自我修正能力能够在下一轮准确补全参数。3. 代码执行安全与沙箱隔离严禁在生产宿主机上直接使用eval()或exec()执行模型生成的代码轻量隔离使用受限环境如RestrictedPython。生产容器隔离将代码打包投递至专用的轻量临时 Docker 容器如使用 Docker SDK或第三方安全沙箱如 E2B Sandbox并在执行完毕后彻底销毁。七、 总结与架构选型指南从理解自回归推理与 Function Calling到手写类型推导工具箱、状态机与记忆系统构建一个专属于业务的轻量 Agent 框架并不复杂。下表总结了自研轻量框架与市面主流开源重型框架的选型对比评估维度自研轻量级 Agent 框架 (如本文实现)开源全家桶 (如 LangChain / CrewAI)工业级图状态机 (如 LangGraph)代码量与复杂度300~500 行零黑盒极易维护极其庞大类继承深黑盒严重中等基于 Graph 抽象调试与排错极佳原生断点可单步调试每一步差异常栈经常深入十几层源码较好状态流转轨迹清晰执行性能与开销极高无额外内存与抽象损耗存在一定程度的包装开销高灵活度与定制性100% 完全可控随需改造受制于框架设定的标准协议极高适合复杂分支业务适用场景企业核心自研业务、高性能 API 网关快速 PoC 原型验证、Demo 演示复杂分支流、人工审批介入工作流在 AI 应用开发的大潮中现成框架能够帮你快速验证想法但深入底层、理解数据流与状态机的本质并拥有按需自研轻量框架的能力才是架构师构建长期技术壁垒的核心底气。