我们先来仔细阅读一下 MetaGPT 源码，从中找到获取灵感，从而帮助设计自己的 Agent 框架—aZent。
项目更新 github 上供大家学习，如果喜欢的话，可以给点个赞，相关视频也可以在 github 上找到链接。github zideajang 项目名称为 tiny_agent_service。
我们先把问题简化
我们先抛开 Agent 来看现在我们使用 GPT 这样的大语言模型的现状，简单来看就是一问一答，根据问题来问大模型，大模型根据自己的知识储备给予我们一个回答。这样根据问题解决程度逐渐递进的一问一答的这样方式来使用大模型来协助我们解决问题，从一个侧面也可以看做一种更快捷更准确的搜索方式，为什么要引入 Agent 来切入到一个一问一答关节中，Agent 能够解决那些现存的问题，以及如何解决这些问题，这是我们设计 Agent 时要重点考虑的问题。
首先我们要实现就是对这个一问一答流程进行抽象，关于抽象下面画了一张图来表示。首先关于问答有几种形式，分别维 ask、ask_batch 和 ask_code ,这里 ask_batch 是递增的，
from abc import ABC, abstractmethodfrom dataclasses import dataclass@dataclassclass BaseChatbot(ABC):mode: str = "API"@abstractmethoddef ask(self, msg: str) -> str:"""Ask LLM a question and get an answer"""@abstractmethoddef ask_batch(self,msgs:list) -> str:"""ask LLM a series of questions and get a series of answers"""@abstractmethoddef ask_code(self, msgs: list) -> str:"""Ask GPT multiple questions and get a piece of code"""
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_msg 一并进行输入作下一个 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)
我们在调用 openAI 的 chatGPT 时候需要传入一些 {角色和内容} 然后 chatGPT 专注到我们感兴趣领域，也就是经常听到催眠 GPT。也就是让 chatGPT 更加聚焦我们感兴趣的领域，不会天马行空是无忌惮回答问题，给出的答案更加专业，准确。
这里我们还是要说既然我们调用 chatGPT 格式，所以还是推荐大家有时间去看一看 openAI 提供 Api 请求的数据格式以及返回的数据格式，毕竟现在大家都向 openAi 的 chatGPT 看齐。
Agent 也是要加 token，加 token
计算开销这个功能可不是小事，这是需要计算成本的，这里用到了 tiktoken 这个库，这个库帮助来计算文本或者说字节到token的转换。语言模型不像我们人类那样看待文本，而是将其视为一系列的数字（称为 token），实际上，平均每个 token 对应大约 4 个字节，这个从文本到 token 这个过程是可逆的。让模型识别常见的 subword。例如，在英语中，'ing' 是一个常见的 subword，所以 BPE 编码通常会将 'encoding' 拆分为 'encod' 和 'ing' 这样的 token (而不是比如 'enc' 和 'oding' 这样来划分单词)，不同模型对不同语言的文本划分为 token 方式是不同的。
当然，这段文字详细地描述了字节对编码（BPE）以及它在文本转换为令牌（tokens）时的作用。我会逐行翻译：
这段描述清晰地阐释了 BPE 的工作原理以及它对于处理和理解文本的重要性。
import tiktokendef 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 tokentokens_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 计算这里不同模型有不同计算方式，这里就不赘述，感兴趣可以深入去了解一下。
其实这些方法可能在实际计算开销用不到，第一个调用这些方法是耗时的，第二个好像调用 chatGPT 在返回对象中是可以获取这些 token 数量的信息的。
class Costs(NamedTuple):total_prompt_tokens: inttotal_completion_tokens: inttotal_cost: floattotal_budget: float
class CostManager(metaclass=Singleton):def __init__(self):self.total_prompt_tokens = 0self.total_completion_tokens = 0self.total_cost = 0self.total_budget = 0self.config = Config()def update_cost(self, prompt_tokens, completion_tokens, model):self.total_prompt_tokens += prompt_tokensself.total_completion_tokens += completion_tokenscost = (prompt_tokens * TOKEN_COSTS[model]["prompt"]+ completion_tokens * TOKEN_COSTS[model]["completion"]) / 1000self.total_cost += costlogger.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_costdef get_total_prompt_tokens(self):return self.total_prompt_tokensdef get_total_completion_tokens(self):return self.total_completion_tokensdef get_total_cost(self):return self.total_costdef 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 使用情况，所以无需使用 tiktoken 库来计算 token 的数量，只要用返回 token 数量乘以对应 token 单价就可以计算项目开销。
