企业官网建设流程全解析

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

市面上关于 Agent 的文章已经多到看不过来了，但真正落到代码层面，你会发现大部分内容要么停留在概念层面反复讲 ReAct、CoT、Tool Use 这些名词，要么直接甩出一个庞大框架的文档链接让你自己去啃。我最初也是抱着“找个开源框架用用算了”的心态，结果翻了好几个仓库之后发现，要么抽象层太厚导致调试困难，要么依赖太重跑起来一堆环境问题。后来我干脆决定：自己参照一个轻量级 Agent 的实现思路，从零搭一个最小可用的版本。

这个想法听起来有点“重复造轮子”，但实际做下来收获远超预期。当你亲手处理消息拼接、工具调用解析、循环终止条件这些细节时，对 Agent 运行机制的理解会发生质变。这篇文章就是把我整个实现过程拆开来讲——不是教你调某个框架的 API，而是从架构设计到代码落地，完整走一遍一个小 Agent 的诞生过程。适合有一定编程基础、想真正搞懂 Agent 内部运转逻辑的开发者，也适合那些用现成框架遇到瓶颈、想自己定制核心逻辑的人。

我参照的核心思路来自一个叫 pi-agent 的轻量实现，它的设计哲学很简单：用最少的抽象层，把 Agent 的核心循环暴露给开发者。下面我会先讲整体设计思路，再逐层拆解关键模块，最后给出可直接运行的代码和踩坑记录。

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

2.1 一个 Agent 到底需要哪些东西

先把问题简化。抛开那些花哨的多模态、长期记忆、多 Agent 协作不谈，一个能干活的最小 Agent 本质上就是三件事的循环：接收输入、决定下一步动作、执行动作并把结果喂回去。这个循环听起来简单，但每个环节都有设计决策要做。

我最终确定的架构包含四个核心模块：

• 消息管理器：负责维护对话历史，包括系统提示、用户输入、助手回复、工具调用结果。它决定了每次请求发给模型时携带哪些上下文。
• 模型接口层：封装对语言模型的调用，处理请求格式化和响应解析。这一层要屏蔽不同模型提供方的差异。
• 工具注册与调度器：管理可用工具的定义、参数校验和实际执行。工具是 Agent 与外部世界交互的唯一通道。
• 主循环控制器：串联以上所有模块，控制“思考-行动-观察”的循环节奏，决定何时终止。

为什么这样分？因为我在实际调试中发现，Agent 出问题的时候，故障点几乎总是落在这四个模块之一。分层清晰意味着出问题时你能快速定位是上下文拼错了、模型返回解析失败了、工具参数不对、还是循环条件写错了。如果所有逻辑揉在一个大函数里，调试就是灾难。

2.2 为什么选择“薄抽象”而不是“厚框架”

这是整个项目最关键的决策。我试过用一些重型框架，它们提供了链式调用、自动重试、内置记忆管理等一大堆功能。好处是上手快，坏处是当你想改一个细节行为时，往往要翻好几层源码才能找到对应的配置项。

pi-agent 的思路正好相反：核心循环就几十行代码，所有可定制的地方都以最直白的方式暴露出来。我采纳了这个思路，具体体现在几个方面：

• 消息历史就是一个普通的列表，不做任何自动截断或摘要，截断策略由使用者自己决定。
• 工具定义就是一个字典，键是工具名，值是包含描述、参数 schema 和执行函数的对象。
• 循环终止条件就是一个简单的判断：模型不再请求调用工具，或者达到最大迭代次数。

这种设计的代价是你需要自己处理一些边界情况，但收益是你对 Agent 的每一个行为都有完全的掌控力。对于想深入理解 Agent 的人来说，这个代价完全值得。

2.3 消息格式的设计考量

消息格式是整个系统的骨架，设计不好后面全是坑。我采用的是类似 OpenAI Chat Completions 的消息结构，每条消息有role和content两个核心字段。role有四种取值：system、user、assistant、tool。

这里有个容易踩的坑：当模型决定调用工具时，它的回复里content可能为空，但会带一个tool_calls字段。而工具执行结果需要以role: tool的消息追加回去，并且要带上对应的tool_call_id来匹配。很多初学者在这里出错，导致模型收到工具结果后无法正确关联到之前的调用请求。

注意：不同模型提供方对工具调用结果的消息格式要求略有差异。有的要求工具结果消息必须紧跟在助手消息之后，有的允许中间插入其他消息。实现时最好严格按照目标模型的文档来构造。

我选择用一个统一的内部格式，然后在模型接口层做转换。这样切换模型时只需要改接口层的适配代码，主循环逻辑完全不用动。

3. 核心模块的详细实现与关键细节

3.1 消息管理器：上下文窗口的取舍艺术

消息管理器看起来只是维护一个列表，但实际写起来要考虑的事情不少。最核心的问题是：上下文窗口有限，历史消息无限增长，怎么取舍？

我的做法是分三层处理：

第一层是系统提示，永远保留在消息列表最前面。系统提示里定义了 Agent 的角色、能力边界、输出格式要求等。这部分内容通常不长，但对 Agent 行为影响巨大。

第二层是最近的对话轮次，按时间倒序保留。我设置了一个可配置的max_recent_turns参数，默认保留最近 10 轮完整对话。一轮指的是用户消息加上助手回复（可能包含多次工具调用）。

第三层是更早的历史，如果总 token 数超过阈值，就从最老的消息开始丢弃。这里有个细节：丢弃时要保证不破坏工具调用和工具结果之间的配对关系，否则模型会报错。

def trim_messages(messages, max_tokens, token_counter): system_msgs = [m for m in messages if m["role"] == "system"] other_msgs = [m for m in messages if m["role"] != "system"] # 从后往前保留，直到超出预算 kept = [] total = sum(token_counter(m) for m in system_msgs) for msg in reversed(other_msgs): msg_tokens = token_counter(msg) if total + msg_tokens > max_tokens: break kept.insert(0, msg) total += msg_tokens # 确保不切断工具调用配对 kept = fix_tool_pairing(kept) return system_msgs + kept

fix_tool_pairing这个函数的作用是：如果保留列表的第一个消息是role: tool，说明它的配对助手消息被截掉了，需要把这条工具结果也去掉。反过来，如果最后一条助手消息包含tool_calls但没有对应的工具结果，也要处理。

3.2 模型接口层：请求构造与响应解析

这一层的职责是把内部消息格式转换成目标模型 API 要求的格式，然后把响应解析回内部格式。我以 OpenAI 兼容接口为例来说明。

请求构造时需要注意几个点：

• 工具定义需要转换成模型要求的 JSON Schema 格式。每个工具包含name、description、parameters三个字段。
• description的写法直接影响模型能否正确选择工具。我踩过的坑是描述写得太简略，模型经常选错工具或者参数填错。后来我把每个工具的描述都写成“什么场景下用这个工具 + 每个参数的含义和格式”，准确率明显提升。
• temperature参数对 Agent 场景建议设低一些，0 到 0.3 之间比较合适。太高会导致模型输出不稳定，工具调用格式容易出错。

响应解析主要处理两种情况：普通文本回复和工具调用请求。工具调用请求里，模型会返回工具名和参数 JSON 字符串。这里必须做 JSON 解析的异常处理，因为模型偶尔会返回格式不正确的 JSON。

def parse_tool_calls(response): tool_calls = [] for tc in response.choices[0].message.tool_calls or []: try: args = json.loads(tc.function.arguments) except json.JSONDecodeError: # 尝试修复常见格式问题 args = try_fix_json(tc.function.arguments) tool_calls.append({ "id": tc.id, "name": tc.function.name, "arguments": args }) return tool_calls

try_fix_json是我后来加的一个兜底函数，处理模型返回单引号、尾逗号、缺少引号等常见问题。虽然不优雅，但实际运行中确实能救回不少失败请求。

3.3 工具注册与调度：让 Agent 真正能干活

工具是 Agent 的手和脚。没有工具，Agent 就只是一个聊天机器人。我的工具注册设计如下：

class Tool: def __init__(self, name, description, parameters, func): self.name = name self.description = description self.parameters = parameters # JSON Schema self.func = func def execute(self, **kwargs): return self.func(**kwargs) registry = {} def register_tool(tool): registry[tool.name] = tool

每个工具本质上就是一个带元数据的函数。parameters用 JSON Schema 描述，既用于生成给模型的工具定义，也用于执行前的参数校验。

我内置了几个基础工具来验证整个链路：

• 计算器：接收一个数学表达式字符串，返回计算结果。用来测试模型能否正确提取参数。
• 当前时间：无参数，返回当前时间字符串。用来测试无参数工具的调用。
• 文本统计：接收一段文本，返回字符数、词数、行数。用来测试多参数工具。

工具执行时的关键细节：

第一，参数校验不能省。模型可能传入多余参数、缺少必要参数、或者类型不对。我在execute前加了一层校验，不通过就返回错误信息给模型，让它重新尝试。

第二，工具执行要加超时和异常捕获。工具函数可能因为各种原因失败（网络问题、文件不存在、除零错误等），这些异常必须被捕获并作为工具结果返回给模型，而不是让整个程序崩溃。

第三，工具结果要控制长度。如果工具返回大量文本，直接塞进消息历史会迅速消耗上下文窗口。我的做法是对超过一定长度的结果做截断，并附上提示说明被截断了。

实操心得：工具描述的质量直接决定 Agent 的可用性。我建议每个工具的描述都包含三部分：功能一句话概括、适用场景说明、每个参数的详细解释。花十分钟写好描述，能省掉后面几个小时的调试。

3.4 主循环控制器：Agent 的心跳

主循环是整个 Agent 的引擎。它的逻辑用伪代码表示就是：

def run_agent(user_input, max_iterations=10): messages.append({"role": "user", "content": user_input}) for i in range(max_iterations): response = call_model(messages) assistant_msg = parse_response(response) messages.append(assistant_msg) tool_calls = assistant_msg.get("tool_calls", []) if not tool_calls: return assistant_msg["content"] for tc in tool_calls: result = execute_tool(tc["name"], tc["arguments"]) messages.append({ "role": "tool", "tool_call_id": tc["id"], "content": str(result) }) return "达到最大迭代次数，任务未完成"

看起来简单，但有几个关键决策点：

最大迭代次数设多少？我默认设 10。太少了复杂任务做不完，太多了万一模型陷入死循环会浪费大量 token。实际使用中可以根据任务复杂度调整，简单问答 3 到 5 次足够，复杂的数据处理任务可能需要 15 到 20 次。

一次响应中有多个工具调用怎么处理？我的做法是全部执行，然后把所有结果都追加到消息历史中。模型下一轮会看到所有工具的结果，再决定下一步。这里要注意执行顺序——如果工具之间有依赖关系，模型通常会在不同轮次分别调用，同一轮内的多个工具调用一般是无依赖的。

循环终止后如果任务没完成怎么办？我选择返回一个明确的提示，而不是静默失败。这样使用者知道是迭代次数不够，可以调整参数重试。

4. 完整实操过程与代码实现

4.1 环境准备与依赖安装

整个项目只需要一个 HTTP 客户端库来调用模型接口，其他都是标准库。我用的是requests，你也可以用httpx或任何你习惯的库。

pip install requests

项目结构如下：

mini-agent/ ├── agent.py # 主循环控制器 ├── message.py # 消息管理器 ├── model.py # 模型接口层 ├── tools.py # 工具注册与调度 └── main.py # 入口

我刻意保持文件数量少，每个文件职责单一。这样你读代码时不需要在多个文件之间跳来跳去。

4.2 模型接口层的完整实现

先定义配置：

# model.py import json import requests class ModelClient: def __init__(self, api_base, api_key, model_name): self.api_base = api_base.rstrip("/") self.api_key = api_key self.model_name = model_name def chat(self, messages, tools=None, temperature=0.1): payload = { "model": self.model_name, "messages": messages, "temperature": temperature, } if tools: payload["tools"] = tools payload["tool_choice"] = "auto" resp = requests.post( f"{self.api_base}/chat/completions", headers={ "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" }, json=payload, timeout=60 ) resp.raise_for_status() return resp.json()

这里temperature默认设 0.1，原因前面说过，Agent 场景需要稳定输出。tool_choice设为auto让模型自己决定是否调用工具。如果你希望强制模型在特定场景必须调用工具，可以设为required，但一般不建议，因为有些问题确实不需要工具。

响应解析部分：

def parse_response(self, raw_response): choice = raw_response["choices"][0] message = choice["message"] result = {"role": "assistant", "content": message.get("content") or ""} if message.get("tool_calls"): result["tool_calls"] = [] for tc in message["tool_calls"]: try: args = json.loads(tc["function"]["arguments"]) except (json.JSONDecodeError, KeyError): args = {} result["tool_calls"].append({ "id": tc["id"], "name": tc["function"]["name"], "arguments": args }) return result

4.3 工具系统的完整实现

工具基类和注册机制：

# tools.py import json import time import math class Tool: def __init__(self, name, description, parameters, func): self.name = name self.description = description self.parameters = parameters self.func = func def to_schema(self): return { "type": "function", "function": { "name": self.name, "description": self.description, "parameters": self.parameters } } def execute(self, arguments): # 参数校验 required = self.parameters.get("required", []) for param in required: if param not in arguments: return f"错误：缺少必要参数 {param}" try: return self.func(**arguments) except Exception as e: return f"工具执行出错：{str(e)}" # 全局注册表 _registry = {} def register(tool): _registry[tool.name] = tool def get_tool(name): return _registry.get(name) def all_schemas(): return [t.to_schema() for t in _registry.values()]

内置工具的实现：

def calc(expression: str) -> str: """安全的数学计算""" allowed = set("0123456789+-*/.() ") if not all(c in allowed for c in expression): return "错误：表达式包含不允许的字符" try: result = eval(expression, {"__builtins__": {}}, {}) return str(result) except Exception as e: return f"计算错误：{str(e)}" register(Tool( name="calculator", description="执行数学计算。当用户需要进行算术运算时使用此工具。参数 expression 是一个数学表达式字符串，例如 '2 + 3 * 4'。", parameters={ "type": "object", "properties": { "expression": { "type": "string", "description": "要计算的数学表达式，支持加减乘除和括号" } }, "required": ["expression"] }, func=calc )) def get_time() -> str: return time.strftime("%Y-%m-%d %H:%M:%S") register(Tool( name="get_current_time", description="获取当前日期和时间。当用户询问现在几点、今天日期等问题时使用。无需参数。", parameters={"type": "object", "properties": {}, "required": []}, func=get_time )) def text_stats(text: str) -> str: chars = len(text) words = len(text.split()) lines = text.count("\n") + 1 return f"字符数：{chars}，词数：{words}，行数：{lines}" register(Tool( name="text_stats", description="统计文本的字符数、词数和行数。当用户需要分析文本长度时使用。参数 text 是要统计的文本内容。", parameters={ "type": "object", "properties": { "text": { "type": "string", "description": "需要统计的文本内容" } }, "required": ["text"] }, func=text_stats ))

注意calc函数里我用了eval但做了字符白名单过滤，并且把__builtins__设为空字典。这是最小化的安全措施，生产环境建议用更严格的表达式解析器。

4.4 主循环的完整实现

# agent.py from model import ModelClient from tools import get_tool, all_schemas class Agent: def __init__(self, client, system_prompt, max_iterations=10): self.client = client self.system_prompt = system_prompt self.max_iterations = max_iterations self.messages = [] def reset(self): self.messages = [{"role": "system", "content": self.system_prompt}] def run(self, user_input): if not self.messages: self.reset() self.messages.append({"role": "user", "content": user_input}) for i in range(self.max_iterations): raw = self.client.chat( messages=self.messages, tools=all_schemas() ) assistant_msg = self.client.parse_response(raw) self.messages.append(assistant_msg) tool_calls = assistant_msg.get("tool_calls", []) if not tool_calls: return assistant_msg["content"] for tc in tool_calls: tool = get_tool(tc["name"]) if tool is None: result = f"错误：未知工具 {tc['name']}" else: result = tool.execute(tc["arguments"]) self.messages.append({ "role": "tool", "tool_call_id": tc["id"], "content": str(result) }) return "抱歉，任务处理步骤过多，未能完成。请尝试简化你的请求。"

入口文件：

# main.py from model import ModelClient from agent import Agent client = ModelClient( api_base="你的接口地址", api_key="你的密钥", model_name="你的模型名称" ) agent = Agent( client=client, system_prompt="你是一个乐于助人的助手。你可以使用提供的工具来帮助回答问题。当需要计算时使用计算器工具，当需要知道时间时使用时间工具。回答要简洁准确。", max_iterations=10 ) while True: user_input = input("你：") if user_input.lower() in ("exit", "quit"): break reply = agent.run(user_input) print(f"助手：{reply}")

4.5 运行效果与关键日志分析

跑起来之后，我测试了几个典型场景。第一个是简单计算：“帮我算一下 123 乘以 456 等于多少”。Agent 的行为是：模型返回工具调用请求，参数为{"expression": "123 * 456"}，工具执行返回56088，模型收到结果后生成最终回复“123 乘以 456 等于 56088”。

第二个是复合任务：“现在几点了？另外帮我统计一下‘hello world’有几个字符”。这一轮模型返回了两个工具调用，分别调用时间工具和文本统计工具。两个工具都执行完后，模型综合两个结果给出了完整回复。

第三个是边界测试：“帮我查一下明天的天气”。由于我没有注册天气工具，模型没有调用任何工具，直接回复说它没有查询天气的能力。这说明系统提示和工具描述共同约束了模型的行为边界。

注意：如果你发现模型该调用工具的时候不调用，首先检查工具描述是否清晰，其次检查系统提示里是否明确引导了工具使用。这两个地方调整之后，工具调用率通常会有明显改善。

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

5.1 工具调用失败排查表

5.2 我踩过的三个典型坑

第一个坑：工具结果消息格式不对。我最初把工具结果作为role: user的消息追加，结果模型完全无法理解这是工具返回的内容，表现得很混乱。后来改成role: tool并带上tool_call_id才正常。这个细节在文档里往往一笔带过，但实际影响巨大。

第二个坑：系统提示写得太长。我一开始在系统提示里塞了大量规则和示例，结果模型反而变得畏手畏脚，该调用工具的时候不调用。后来精简到只保留核心角色定义和工具使用原则，效果反而更好。系统提示不是越长越好，关键是信息密度。

第三个坑：没有处理模型返回空内容的情况。有时候模型既没有返回文本内容，也没有返回工具调用，就是一个空回复。这种情况如果不处理，程序会卡住或者返回空字符串。我的处理方式是：如果连续两次收到空回复，就返回一个提示信息并终止循环。

5.3 提升 Agent 稳定性的几个实用技巧

技巧一：给工具调用加日志。每次工具调用都记录工具名、参数、执行结果和耗时。出问题的时候，日志是最快的排查手段。我用一个简单的列表存这些记录，调试时打印出来一目了然。

技巧二：参数校验要宽松但有底线。模型可能传入字符串类型的数字而不是数字类型，这种可以自动转换。但如果缺少必要参数，必须返回错误让模型重试，不能默默用默认值填充。

技巧三：设置合理的超时。模型调用和工具执行都要设超时。模型调用我设 60 秒，工具执行设 30 秒。超时后返回错误信息给模型，让它决定是重试还是换一种方式。

技巧四：用低 temperature 加少量示例。在系统提示里给一两个工具调用的示例，能显著提升模型输出格式的正确率。示例不需要多，一个简单场景加一个复合场景就够了。

5.4 扩展方向与后续优化思路

这个最小 Agent 跑通之后，可以往几个方向扩展。一是加记忆持久化，把消息历史存到文件或数据库，支持跨会话的上下文。二是加流式输出，让模型回复逐字显示，提升交互体验。三是加多轮工具调用的依赖处理，比如前一个工具的输出作为后一个工具的输入。四是加工具的动态加载，从配置文件或目录自动发现和注册工具。

每个方向都可以独立展开，但核心循环的逻辑不需要大改。这也是薄抽象设计的好处——扩展点清晰，改动范围可控。

我个人在实际操作中的体会是，写 Agent 最难的不是代码本身，而是对模型行为的理解和引导。同样的工具定义，换一种描述方式，模型的调用准确率可能从 60% 跳到 90%。这需要反复测试和调整，没有一劳永逸的方案。建议你在自己的场景里多准备一些测试用例，每次调整提示或工具描述后跑一遍，用数据来指导优化方向。