企业官网建设流程全解析

1. 为什么我们要自己动手写一个 Agent 框架

市面上开源的 Agent 框架已经多到挑花眼，从轻量级的脚本封装到企业级的多智能体编排平台，几乎每个月都有新项目冒出来。但真正落到实际项目里，你会发现一个很尴尬的现实：大部分框架要么太重，要么太薄。重的那些把记忆、规划、工具调用、多智能体协作全塞进一个黑盒，你想改一行调度逻辑得翻半天源码；薄的那些又只给你一个循环调用大模型的壳子，工具注册、上下文管理、错误重试全得自己补。

我在过去一年里陆续用几套主流方案做过原型，踩过的坑基本集中在三个地方：工具抽象不统一导致每接一个外部能力就要写一套适配层；上下文变量在多轮对话里到处乱飞，调试的时候根本追不到某个变量是在哪一步被改掉的；还有就是并发场景下状态互相污染，单机测试没问题，一上量就出诡异 bug。

所以这一章我打算从零搭一个自己的 Agent 框架，代号就叫HelloAgents。目标很明确：核心调度逻辑自己掌控，工具抽象统一，上下文变量显式管理，能扛住一定并发。它不追求功能大而全，而是追求每一层都看得见、改得动。适合那些已经用过现成框架、想搞清楚内部到底怎么转的开发者，也适合准备把 Agent 能力嵌进自己业务系统、需要深度定制的团队。

这一节先讲整体架构设计，把骨架搭清楚，后面几节再逐个模块填充实现。架构这件事，前期多想两天，后期少改两周，这个账我算过很多次，划得来。

2. 框架整体架构设计

2.1 从一次真实的调用链路倒推分层

设计架构最忌讳上来就画一堆框。我的习惯是先把一次完整的调用链路写出来，再倒推需要哪些层。假设用户问了一句“帮我查一下明天北京的天气，如果下雨就提醒我带伞”，一个 Agent 框架内部大概要经历这些步骤：

1. 接收用户输入，组装成消息对象
2. 把消息和历史上下文一起送给大模型
3. 模型返回一个工具调用意图（查天气）
4. 框架解析意图，找到对应工具并执行
5. 把工具结果塞回上下文，再次调用模型
6. 模型根据天气结果决定是否调用“提醒”工具
7. 汇总最终回复返回给用户

这条链路里，模型交互、工具执行、上下文管理、循环控制是四个绕不开的核心动作。再往外扩一层，还有配置加载、日志追踪、并发调度、错误处理这些横切关注点。把这些归归类，HelloAgents 的分层就出来了。

2.2 四层架构与职责边界

我把整个框架分成四层，从下往上依次是：

这么分的好处是依赖方向单一：上层依赖下层的抽象，下层不知道上层的存在。L2 定义接口，L3 提供实现，L4 负责组装和驱动。你想换一个大模型供应商，只动 L3；想改调度策略，只动 L4；想加一个日志字段，只动 L1。各层之间通过接口通信，改一处不会引发雪崩。

提示：分层不是越多越好。我见过有人把 Agent 框架分成七八层，结果一个简单的工具调用要穿过五层转发，调试的时候栈深得吓人。四层是我实践下来比较舒服的粒度，再少会耦合，再多会臃肿。

2.3 核心抽象：Agent、Tool、Context 三件套

整个框架的抽象核心就三个东西，把它们定清楚，框架就成了一半。

Agent是执行主体，它持有模型客户端、工具集合、上下文，对外暴露一个run方法。Agent 不关心工具怎么执行，也不关心上下文怎么存储，它只负责“想”和“决定下一步做什么”。

Tool是能力单元，每个工具有一个名字、一段描述、一份参数 schema、一个执行函数。描述和 schema 是给模型看的，执行函数是给框架调的。工具本身应该是无状态的纯函数，输入参数输出结果，不持有跨调用的状态。

Context是上下文容器，承载消息历史、变量表、工具调用记录。它是整个框架里最容易被写乱的地方，所以我的原则是：上下文变量必须显式声明、显式读写，禁止隐式全局状态。这一点后面会展开讲。

# 核心抽象示意，非完整实现 class Agent(ABC): @abstractmethod def run(self, input_text: str, context: Context) -> str: ... class Tool(ABC): name: str description: str parameters: dict @abstractmethod def execute(self, **kwargs) -> Any: ... class Context: messages: list variables: dict tool_calls: list

2.4 为什么选择“显式循环”而不是“隐式黑盒”

很多框架把 Agent 的执行循环藏得很深，你调用一个agent.chat()就完事了，中间发生了什么完全看不到。这在 demo 阶段很爽，但一旦出问题就抓瞎。HelloAgents 选择把执行循环暴露出来，用一个Executor类显式驱动：

class Executor: def __init__(self, agent, tools, max_steps=10): self.agent = agent self.tools = {t.name: t for t in tools} self.max_steps = max_steps def run(self, input_text, context): context.add_message("user", input_text) for step in range(self.max_steps): response = self.agent.think(context) if response.is_final: return response.content for call in response.tool_calls: tool = self.tools.get(call.name) result = tool.execute(**call.arguments) context.add_tool_result(call.id, result) raise MaxStepsExceeded()

这样每一步都可见、可打断、可记录。max_steps是防止死循环的硬保险，我一般设 10 到 15，具体看任务复杂度。超过就抛异常，而不是无限转下去烧 token。

2.5 工具抽象的设计取舍

工具抽象这块我纠结过很久，最后定下来的方案是基于 schema 的声明式注册。每个工具用装饰器注册，自动从函数签名和类型注解生成参数 schema：

@tool(name="get_weather", description="查询指定城市的天气") def get_weather(city: str, date: str = "today") -> dict: ...

这样做的好处是模型能直接拿到结构化的参数定义，减少它瞎编参数的概率。坏处是对复杂嵌套参数支持不够优雅，遇到那种需要传数组或对象的工具，还是得手写 schema。我的做法是简单工具用装饰器，复杂工具允许手动指定 schema，两条路都留着。

工具执行必须加超时和异常隔离。一个工具卡住不能拖垮整个 Agent，一个工具抛异常不能中断整个循环。我在 Executor 里对每次工具调用都包了 try-except 和超时控制，失败的结果作为一条特殊的 tool result 塞回上下文，让模型自己决定是重试还是换方案。

3. 核心模块的细节拆解与实操要点

3.1 上下文管理：变量显式化是底线

上下文管理是 Agent 框架里最容易埋雷的地方。我见过太多项目用一个全局 dict 存所有状态，结果多轮对话之间互相污染，并发一上来直接串数据。HelloAgents 的做法是把上下文分成三个独立的区：

• messages：对话消息列表，只增不改，保证可追溯
• variables：显式声明的变量表，读写都要走方法
• tool_calls：工具调用记录，用于审计和调试

变量表这块我强制要求先声明后使用。Context 初始化时传入一个变量名列表，之后只能读写这些变量，写未声明的变量直接报错。这个约束看起来麻烦，但它能在开发阶段就拦住大部分“变量名拼错”“作用域搞混”的低级 bug。

class Context: def __init__(self, allowed_vars: list[str]): self.messages = [] self.variables = {v: None for v in allowed_vars} self.tool_calls = [] def set_var(self, name, value): if name not in self.variables: raise UndeclaredVariable(name) self.variables[name] = value def get_var(self, name): if name not in self.variables: raise UndeclaredVariable(name) return self.variables[name]

注意：上下文不要存大对象。我踩过一次坑，把整个数据库查询结果集塞进 variables，结果每轮对话都带着几百 KB 的数据发给模型，token 费用直接起飞。大结果要么截断，要么存到外部存储只留一个引用 ID。

3.2 消息组装与 token 预算控制

消息组装看着简单，其实有讲究。模型能接受的上下文长度有限，历史消息不能无限堆。我的策略是滑动窗口加摘要：保留最近 N 轮完整消息，更早的消息压缩成一段摘要。

具体实现上，我维护一个 token 计数器，每加一条消息就估算一下总长度。超过预算的 80% 就触发压缩，把最老的一批消息交给模型生成摘要，替换掉原始消息。这样既保留了长期记忆，又控制了成本。

def add_message(self, role, content): self.messages.append({"role": role, "content": content}) if self.estimate_tokens() > self.budget * 0.8: self.compress_oldest()

token 估算不用太精确，按字符数除以 3 到 4 粗估就够用，目的是触发压缩而不是精确计费。真要精确计费，用对应模型的 tokenizer，但那个开销也不小，没必要每轮都算。

3.3 工具路由与参数校验

模型返回的工具调用意图，参数不一定合法。可能缺字段，可能类型不对，可能编了一个不存在的工具名。框架必须在执行前做校验，不能直接把模型的输出喂给工具函数。

我的校验分三步：先查工具是否存在，再用 schema 校验参数类型和必填项，最后做一次业务层面的合法性检查（比如日期格式、城市名是否在支持列表里）。任何一步失败，都把错误信息作为 tool result 返回给模型，让它自己修正。

def dispatch(self, call): tool = self.tools.get(call.name) if not tool: return ToolResult(error=f"未知工具: {call.name}") try: validate(call.arguments, tool.parameters) except ValidationError as e: return ToolResult(error=f"参数错误: {e}") try: return ToolResult(data=tool.execute(**call.arguments)) except Exception as e: return ToolResult(error=f"执行失败: {e}")

这套流程实测下来，模型自我修正的成功率挺高。第一次参数错了，把错误信息回传，它第二次基本能改对。比在框架层面硬猜要靠谱得多。

3.4 并发与状态隔离

Agent 扛并发是个绕不开的话题。单机跑一个 Agent 没问题，同时跑一百个就开始出各种诡异问题。核心矛盾在于共享状态。模型客户端可以共享，工具实例可以共享（前提是无状态），但 Context 绝对不能共享。

我的做法是每个请求一个独立的 Context 实例，Executor 也是每次请求新建。共享的只有那些只读的配置和线程安全的客户端。工具如果必须有状态（比如带缓存的工具），要么加锁，要么每个请求一个实例，看性能取舍。

# 每个请求独立上下文，共享只读资源 def handle_request(user_input, shared_resources): context = Context(allowed_vars=["user_id", "session_id"]) executor = Executor( agent=shared_resources.agent, tools=shared_resources.tools, max_steps=10 ) return executor.run(user_input, context)

提示：并发压测一定要做。我见过一个框架单请求测试完美，并发 50 就开始返回错乱的结果，查了半天发现是工具实例里缓存了上一次的输入。工具无状态这条原则，在并发场景下是保命的。

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

4.1 从零搭建项目骨架

先把目录结构定下来，后面填代码就顺了。我的习惯是按层分目录，不按功能分：

helloagents/ ├── core/ # L2 核心抽象 │ ├── agent.py │ ├── tool.py │ ├── context.py │ └── message.py ├── infra/ # L1 基础设施 │ ├── config.py │ ├── logger.py │ └── errors.py ├── impl/ # L3 能力实现 │ ├── agents/ │ ├── tools/ │ └── memory/ ├── executor/ # L4 编排调度 │ ├── executor.py │ └── router.py └── tests/

这个结构的好处是依赖关系一目了然。core不依赖任何其他目录，infra只被上层依赖，impl依赖core和infra，executor依赖所有下层。你想找某个功能的实现，按层找就行，不用在功能目录里翻来翻去。

4.2 定义核心接口

先写core里的抽象。这部分要克制，接口越少越好，每个接口只暴露必要的方法。

# core/message.py from dataclasses import dataclass, field from typing import Literal @dataclass class Message: role: Literal["system", "user", "assistant", "tool"] content: str = "" tool_calls: list = field(default_factory=list) tool_call_id: str = "" # core/tool.py class Tool(ABC): name: str description: str parameters: dict @abstractmethod def execute(self, **kwargs): ... # core/agent.py class Agent(ABC): @abstractmethod def think(self, context: Context) -> Message: ...

think方法只负责调用模型并返回一条消息，不负责执行工具。执行工具是 Executor 的事。这个职责划分很关键，它让 Agent 保持纯粹，也让 Executor 可以灵活替换调度策略。

4.3 实现一个最小可用的 Executor

Executor 是整个框架的心脏，先把最小版本跑通，再逐步加功能。

# executor/executor.py class Executor: def __init__(self, agent, tools, max_steps=10, timeout=30): self.agent = agent self.tools = {t.name: t for t in tools} self.max_steps = max_steps self.timeout = timeout def run(self, input_text, context): context.add_message("user", input_text) for step in range(self.max_steps): msg = self.agent.think(context) context.add_message(msg) if not msg.tool_calls: return msg.content for call in msg.tool_calls: result = self._execute_tool(call) context.add_tool_result(call.id, result) raise MaxStepsExceeded(f"超过 {self.max_steps} 步仍未完成") def _execute_tool(self, call): tool = self.tools.get(call.name) if not tool: return f"错误：工具 {call.name} 不存在" try: return tool.execute(**call.arguments) except Exception as e: return f"错误：{type(e).__name__}: {e}"

这个版本没有超时控制、没有并发、没有重试，但核心循环是完整的。先让它跑起来，再逐步加固。我强烈建议不要一上来就写完整版，那样调试起来会很痛苦。

4.4 注册第一个工具并跑通链路

写一个最简单的工具验证整条链路：

# impl/tools/calculator.py from helloagents.core.tool import Tool class CalculatorTool(Tool): name = "calculator" description = "执行简单的四则运算，输入表达式字符串" parameters = { "type": "object", "properties": { "expression": {"type": "string", "description": "如 1+2*3"} }, "required": ["expression"] } def execute(self, expression: str): allowed = set("0123456789+-*/(). ") if not set(expression) <= allowed: raise ValueError("表达式包含非法字符") return eval(expression, {"__builtins__": {}}, {})

注意这里用eval之前做了字符白名单校验，并且清空了 builtins。生产环境更稳妥的做法是用ast解析后只允许特定节点，但作为示例这个够用了。

跑通链路：

agent = MyAgent(model_client) tools = [CalculatorTool()] executor = Executor(agent, tools) context = Context(allowed_vars=["user_id"]) result = executor.run("帮我算一下 (12+8)*3 等于多少", context) print(result)

如果模型正确调用了 calculator 工具并返回了 60，说明整条链路通了。这一步跑通，后面加什么功能都是在这个骨架上长肉。

4.5 加上日志与追踪

链路跑通后第一件事是加日志。没有日志的 Agent 框架就是个黑盒，出了问题只能靠猜。我在 L1 的 logger 里定义了几个关键事件：模型调用、工具调用、上下文变更、异常。

# infra/logger.py import logging, json logger = logging.getLogger("helloagents") def log_event(event_type, **kwargs): logger.info(json.dumps({"event": event_type, **kwargs}, ensure_ascii=False))

在 Executor 的关键节点埋点：

log_event("model_call", step=step, tokens=context.estimate_tokens()) log_event("tool_call", name=call.name, args=call.arguments) log_event("tool_result", name=call.name, result=str(result)[:200])

日志用 JSON 格式，方便后续接入日志系统做检索和告警。result截断到 200 字符，避免大结果把日志撑爆。这些细节看着小，真出问题的时候能救命。

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

5.1 模型不调用工具怎么办

这是新手最常遇到的问题。模型明明有能力调工具，但它就是直接回答，不走工具。原因通常有三个：工具描述写得太模糊、系统提示词没强调要用工具、或者模型本身能力不够。

排查顺序：先看工具描述，是不是写得太笼统。description="查询天气"就不如description="查询指定城市指定日期的天气情况，当用户询问天气时必须使用此工具"。描述里明确写出“什么时候该用”，模型调用率会明显提升。再看系统提示词，有没有明确告诉模型“你有以下工具可用，需要时请调用”。最后才考虑换模型。

5.2 工具参数总是传错

模型传错参数，八成是 schema 写得不够清楚。每个参数的description要写明白格式和示例。比如日期参数，写"格式为 YYYY-MM-DD，如 2024-01-15"比只写"日期"强太多。

还有一个技巧是给参数加枚举约束。如果某个参数只能是几个固定值之一，在 schema 里用enum列出来，模型基本不会传错。这比在代码里做校验再回传错误要高效。

5.3 死循环与步数爆炸

Agent 陷入死循环的典型表现是反复调用同一个工具，或者两个工具来回调用。max_steps是最后的保险，但更好的做法是提前检测。我在 Executor 里加了一个简单的重复检测：如果连续三次工具调用完全一样（工具名和参数都相同），直接中断并返回错误。

def _is_repeating(self, history, call): recent = history[-3:] return all(c.name == call.name and c.arguments == call.arguments for c in recent)

这个检测能拦住大部分无意义的循环。真遇到模型钻牛角尖，中断比让它继续烧钱要明智。

5.4 并发下的状态污染

前面提过，这里给一个具体的排查方法。如果并发测试出现结果错乱，先检查三件事：Context 是不是每个请求独立的、工具实例有没有持有可变状态、模型客户端是不是线程安全的。

我常用的排查手段是在 Context 里加一个唯一的request_id，每条日志都带上。出问题时按request_id过滤日志，就能看到某个请求的完整链路，很快能定位到是哪一步串了数据。

5.5 上下文膨胀导致成本失控

跑一段时间后发现 token 费用飙升，基本是上下文没控制好。检查两个地方：一是历史消息有没有做压缩，二是工具返回的大结果有没有截断。

我的经验值是单次请求的上下文控制在模型窗口的 50% 以内比较安全，留出空间给模型输出和工具结果。超过就压缩，压缩策略用摘要而不是直接丢弃，保留关键信息。

注意：压缩本身也要花 token。如果对话轮次不多，压缩的成本可能比省下来的还高。我一般设置一个阈值，比如历史消息超过 20 条才触发压缩，低于这个数就让它堆着。

5.6 工具执行超时拖垮整个请求

外部工具（比如调 API）响应慢是常态。没有超时控制的话，一个慢工具能把整个 Agent 请求卡死。我在 Executor 里对每次工具调用都加了超时：

import signal def _execute_with_timeout(self, tool, args, timeout): def handler(signum, frame): raise TimeoutError() signal.signal(signal.SIGALRM, handler) signal.alarm(timeout) try: return tool.execute(**args) finally: signal.alarm(0)

signal方案在 Unix 下可用，Windows 下得用线程池加future.result(timeout=...)。超时后返回一个错误结果给模型，让它决定是重试还是换方案。这比直接抛异常中断整个请求要友好。

6. 架构演进与扩展方向

6.1 从单 Agent 到多 Agent 协作

单 Agent 跑通后，很自然会想扩展到多 Agent。我的建议是不要急着上多 Agent。多 Agent 带来的复杂度是数量级的：通信协议、任务分配、结果汇总、冲突解决，每一个都是坑。先把单 Agent 的工具调用和上下文管理做扎实，确实遇到单 Agent 搞不定的场景（比如需要不同角色从不同角度分析同一问题），再考虑多 Agent。

真要扩展的话，HelloAgents 的架构是支持的。Executor 可以升级成 Orchestrator，管理多个 Agent 实例，每个 Agent 有自己的 Context，通过消息传递协作。核心抽象不用动，只是编排层变复杂。

6.2 记忆系统的接入点

长期记忆是另一个常见的扩展需求。我的设计里，记忆作为 Context 的一个可选组件接入，不侵入核心逻辑。Context 在组装消息时，如果配置了记忆模块，就先从记忆里检索相关片段，拼接到系统提示词里。

class Context: def __init__(self, allowed_vars, memory=None): self.memory = memory ... def build_prompt(self): base = self.system_prompt if self.memory: relevant = self.memory.retrieve(self.current_input) base += "\n\n相关记忆：\n" + relevant return base

记忆检索的时机和数量要控制。检索太多会稀释当前对话的焦点，检索太少又没效果。我一般限制在 3 到 5 条，按相关度排序。

6.3 可观测性的持续建设

框架上线后，可观测性决定了你排查问题的速度。除了日志，我还建议加两个东西：指标采集和链路追踪。指标包括每步耗时、token 消耗、工具调用成功率、异常率。链路追踪把一次请求的所有步骤串起来，用 trace_id 关联。

这些不用一开始就做全，但架构上要留好口子。L1 的 logger 和 metrics 模块就是干这个的，上层只管埋点，具体往哪送由配置决定。这样从开发环境到生产环境，可观测性方案可以平滑切换。

6.4 安全边界的考虑

Agent 能调工具就意味着能产生副作用。删数据、发消息、改配置，这些操作一旦被模型误触发，后果可能很严重。我的做法是给工具分级：只读工具直接执行，写操作工具需要二次确认，高危工具（比如删除）默认禁用，需要显式开启。

class Tool: risk_level: Literal["read", "write", "dangerous"] = "read"

Executor 在执行前检查风险等级，write级别记录审计日志，dangerous级别直接拒绝并提示需要人工确认。这个机制在早期就加上，比事后补救要省心得多。

7. 我在实际搭建中的几点体会

搭这个框架的过程中，最大的体会是克制。功能很容易越加越多，但每加一个功能，调试成本就上升一截。我砍掉过好几个“看起来很酷”的设计，比如自动工具发现、动态提示词优化，因为它们带来的复杂度超过了实际收益。

另一个体会是先跑通再优化。最小可用的 Executor 只有几十行代码，但它能跑通完整链路。有了这个基础，加日志、加超时、加并发控制都是增量改进，每一步都能验证。反过来，如果一上来就设计一个“完美”的架构，很可能写到一半发现某个假设不成立，推倒重来的代价太大。

最后一点是关于工具抽象的。工具是 Agent 和外部世界的接口，这个接口设计得好不好，直接决定了框架的扩展性。我的经验是工具要小而专，一个工具只做一件事。get_weather和send_notification分开，不要合成一个handle_weather_request。小工具组合起来灵活，大工具改起来牵一发动全身。

这套架构目前在我自己的几个项目里跑着，单机并发几十路没问题，工具调用链路清晰，出问题能快速定位。后面几节会逐个模块展开实现细节，包括模型客户端的封装、工具注册机制、上下文压缩算法、并发调度策略。如果你也在搭自己的 Agent 框架，希望这个架构设计能给你省点试错的时间。