知己知彼，深入解读 MetaGPT 的源码

先来看看MetaGPT的源码设计，从中我们能找到不少启发，用于构建自己的Agent框架——aZent。这篇文章将逐步拆解其中的核心思路，帮助大家理解如何将大语言模型从“一问一答”的简单工具，升级为真正能处理复杂任务的智能体。

先抛开Agent的概念，看看我们现在使用GPT这类大模型时的实际状况。说白了，核心交互方式就是“一问一答”：我们抛出一个问题，模型根据自己的知识储备给出一个答案。这种方式虽然够直接，但也意味着每次对话的深度和广度完全取决于问题的质量。如果我们把这种逐层递进的多轮对话看作一种更智能的搜索，那引入Agent的意义就很明确了——它不是要替代这个对话过程，而是要切入其中，解决现有模式的几个关键痛点。哪些问题？怎么解决？这才是设计Agent时首先要想清楚的事。

首先需要做的，就是对“一问一答”这个流程进行抽象化。下面这张图能直观展示这种抽象的思路。我们可以把问答行为归纳为几种基本形式：ask（单次提问）、ask_batch（连续对话）、ask_code（代码生成）。其中，ask_batch是递增式的交互，每一次回答都会融入后续的上下文。

from abc import ABC, abstractmethod
from dataclasses import dataclass

@dataclass
class BaseChatbot(ABC):
    mode: str = "API"

    @abstractmethod
    def ask(self, msg: str) -> str:
        """Ask LLM a question and get an answer"""

    @abstractmethod
    def ask_batch(self,msgs:list) -> str:
        """ask LLM a series of questions and get a series of answers"""

    @abstractmethod
    def ask_code(self, msgs: list) -> str:
        """Ask GPT multiple questions and get a piece of code"""

具体实现上，ask_batch 的思路很清晰：将用户输入的消息序列依次添加到上下文中，每次调用模型后，再把模型返回的内容作为 assistant 角色的消息追加回去，然后继续下一轮。这样就能自然地保持多轮对话的连贯性。

def ask_batch(self, msgs: list) -> str:
    context = []
    for msg in msgs:
        umsg = self._user_msg(msg)
        context.append(umsg)
        rsp = self.completion(context)
        rsp_text = self.get_choice_text(rsp)
        context.append(self._assistant_msg(rsp_text))
    return self._extract_assistant_rsp(context)

每次将 context.append(self._assistant_msg(rsp_text)) ，把返回值作为 assistant role 的输入，这样就能维持连续的多轮对话。

在此基础上，辅助方法也很直接：

def _user_msg(self, msg: str) -> dict[str, str]:
    return {"role": "user", "content": msg}

def _assistant_msg(self, msg: str) -> dict[str, str]:
    return {"role": "assistant", "content": msg}

def _system_msg(self, msg: str) -> dict[str, str]:
    return {"role": "system", "content": msg}

def _system_msgs(self, msgs: list[str]) -> list[dict[str, str]]:
    return [self._system_msg(msg) for msg in msgs]

def _default_system_msg(self):
    return self._system_msg(self.system_prompt)

当我们通过API调用 OpenAI 的 ChatGPT 时，核心就是传入包含角色和内容的对话消息。通过 system prompt 设置角色和关注领域，可以让模型更聚焦，不会天马行空，给出的答案也更专业、精准，这就是业内常说的“催眠 GPT”或“角色设定”。

既然调用格式基于 ChatGPT API，还是建议花点时间看看 OpenAI 官方的请求和返回数据格式，毕竟目前大部分大模型接口都在向这个标准对齐。

计算 token 开销可不是小事，这直接关系到项目的运营成本。这里用到了 tiktoken 这个库，它专门用于将文本或字节转换为 token 数量。语言模型不像我们那样直接看文字，而是将文本视为一系列数字（即 token）。平均每个 token 对应大约 4 个字节，这个过程是可逆的。模型通过识别常见的子词单元来理解文本。比如在英语中，“ing”是一个常见子词，BPE（字节对编码）会将 “encoding” 拆分为 “encod” 和 “ing”，而不是 “enc” 和 “oding”。不同模型在不同语言上的 token 划分方式也各有差异。

import tiktoken

def count_string_tokens(string: str, model_name: str) -> int:
    encoding = tiktoken.encoding_for_model(model_name)
    return len(encoding.encode(string))

if __name__ == "__main__":
    test_sentence = "hello world azent is coming soon" # 7 token
    tokens_len = count_string_tokens(test_sentence,"gpt-3.5-turbo")
    print(tokens_len)

if __name__ == "__main__":
    test_sentence = "您好, azent 框架是为了中小企业而生，是 LLM 走进您的企业和公司正确方式，感觉您的支持，虽然是一杯咖啡，却盛满对开发者支持!"
    tokens_len = count_string_tokens(test_sentence,"gpt-3.5-turbo")
    print(tokens_len)

开始实现之前，先看看 tiktoken。这个库可以根据传入的模型和文本，准确计算出所需的 token 数量。

TOKEN_COSTS = {
    "gpt-3.5-turbo": {"prompt": 0.002, "completion": 0.002},
    "gpt-3.5-turbo-0301": {"prompt": 0.002, "completion": 0.002},
    "gpt-4-0314": {"prompt": 0.03, "completion": 0.06},
    "gpt-4": {"prompt": 0.03, "completion": 0.06},
    "gpt-4-32k": {"prompt": 0.06, "completion": 0.12},
    "gpt-4-32k-0314": {"prompt": 0.06, "completion": 0.12},
    "text-embedding-ada-002": {"prompt": 0.0004, "completion": 0.0},
}

不同模型有不同的收费标准，这需要实时参考官方定价。这里只是一个通用示例，仅做参考。关于 token 计算，不同模型方法不同，这里不展开，感兴趣可以深入了解。

其实，在实际开发中，计算 token 开销时往往不需要每次都自己用 tiktoken 算一遍。因为 ChatGPT 的 API 返回对象中已经包含了本次请求使用的 token 数量信息，直接拿这些数据乘以单价就行。

class Costs(NamedTuple):
    total_prompt_tokens: int
    total_completion_tokens: int
    total_cost: float
    total_budget: float

class CostManager(metaclass=Singleton):
    def __init__(self):
        self.total_prompt_tokens = 0
        self.total_completion_tokens = 0
        self.total_cost = 0
        self.total_budget = 0
        self.config = Config()

    def update_cost(self, prompt_tokens, completion_tokens, model): 
        self.total_prompt_tokens += prompt_tokens
        self.total_completion_tokens += completion_tokens
        cost = (
            prompt_tokens * TOKEN_COSTS[model]["prompt"]
            + completion_tokens * TOKEN_COSTS[model]["completion"]
        ) / 1000
        self.total_cost += cost
        logger.info(f"Total running cost: ${self.total_cost:.3f} | Max budget: ${self.config.max_budget:.3f} | "
                     f"Current cost: ${cost:.3f}, {prompt_tokens=}, {completion_tokens=}")
        self.config.total_cost = self.total_cost

    def get_total_prompt_tokens(self):
        return self.total_prompt_tokens

    def get_total_completion_tokens(self):
        return self.total_completion_tokens

    def get_total_cost(self):
        return self.total_cost

    def get_costs(self) -> Costs:
        return Costs(self.total_prompt_tokens, self.total_completion_tokens, self.total_cost, self.total_budget)

从 update_cost 方法可以清楚看到，由于 OpenAI API 返回的信息已经包含了 token 使用情况，我们完全可以用返回的 token 数量直接乘以对应单价来精确计算项目开销，而不需要再额外调用 tiktoken 库。