从零开始的LLM 3. 大模型应用

发表于 更新于

前言

从零开始学习ai文章系列已完成《动手学深度学习》和《磨菇书》两本书的学习，新开的LLM系列课本来自《Happy-LLM》，但是内容和排版等个人重新进行整理，因此不会按照原来课本中的章节来写。如果哪里有错误的欢迎指正。或者不清晰的可以直接查看原文部分。

《Happy-LLM》原文（课本）：https://datawhalechina.github.io/happy-llm/#/

在前面的章节中，我们系统地介绍了大模型的基础知识、训练方法和微调技术。本章将重点探讨大模型在实际应用中的关键技术和框架，涵盖大模型评测、RAG（检索增强生成）以及Agent（智能体）等核心内容，帮助读者深入理解大模型的实际应用场景和实现方法。

1. LLM 的评测

1.1 LLM 的评测数据集

在大模型的评测过程中，使用标准化的评测集至关重要。目前，主流的大模型评测集主要从以下几个方面进行评估：

1. 通用评测集

  • MMLU（Massive Multitask Language Understanding）：覆盖历史、数学、物理、生物、法律等数十个学科，全面考察模型跨领域的知识储备与语言理解能力。

2. 工具使用评测集

  • BFCL V2：评测模型在复杂工具调用任务中的表现，重点考察多步骤操作的正确性与效率，典型任务包括数据库交互、API 调用及特定指令执行。

3. 数学评测集

  • GSM8K：包含小学数学应用题，测试模型的数学推理与逻辑分析能力，涵盖算术运算、简单方程求解、数字推理等任务，同时考验语义理解与数学运算的双重能力。
  • MATH：针对代数、几何、数论等高难度数学问题的评测集，全面测试模型的深度数学推理能力。

4. 推理评测集

  • ARC Challenge：评测模型在科学推理任务中的表现，聚焦常识性与科学性问题解答，适用于科学考试题解答与百科问答系统开发。
  • GPQA（Graduate-Level Google-Proof Q&A）：题目难度达到研究生水平，即便借助搜索引擎也难以轻易作答，用于衡量模型在零样本条件下的深度专业知识与推理上限。
  • HellaSwag：评测模型在复杂语境下选择最符合逻辑答案的能力，适用于故事续写、对话生成等需要高阶情境理解的场景。

5. 长文本理解评测集

  • InfiniteBench/En.MC：评测模型处理超长文本阅读理解的能力，侧重学术文献理解，适用于自动摘要、长篇报道分析等场景。
  • NIH/Multi-needle：在超长上下文中埋入多个关键信息点（"needles"），测试模型的全局信息检索与整合能力，应用于政府报告解读、企业长文档分析等场景。

6. 多语言评测集

  • MGSM（Multilingual Grade School Math）：跨语言评估模型的数学解题能力，考察多语言适应性，适用于国际化数学教育与跨语言技术支持场景。

1.2 主流的评测榜单

大模型的评测不仅限于特定的数据集，许多机构还会根据评测结果发布模型排行榜，为学术界和工业界提供重要参考。以下是三个主流评测榜单：

Open LLM Leaderboard

由 Hugging Face 维护的开放式排行榜，汇集大量开源大模型的评测结果，通过多个标准化测试集综合评估模型性能并持续更新，是开源模型横向对比的核心参考平台。

🔗 https://huggingface.co/spaces/open-llm-leaderboard/open_llm_leaderboard

Chatbot Arena Leaderboard

由 LMSYS 提供，采用真实用户与模型匿名对比交互（人类盲测投票）的方式评估对话质量，重点考察自然语言生成、上下文理解与用户满意度，能更真实地反映模型在实际对话场景中的表现。

🔗 https://huggingface.co/spaces/awacke1/lmsys-chatbot-arena-leaderboard

OpenCompass

国内权威大模型评测榜单，由上海人工智能实验室联合多家机构共同建设，结合中文语言理解与多语言能力测试，特别关注模型在中文语境下的准确性、鲁棒性与适应性，为国内企业和研究者提供本土化的权威参考。

🔗 https://rank.opencompass.org.cn/home

1.3 垂直领域专项榜单

除通用榜单外，多家学术机构还针对不同行业场景推出了专项评测榜单：

榜单 评测基准 评测维度 主办机构
金融榜 CFBenchmark 金融 NLP、预测计算、金融分析与安全检查 同济大学、上海人工智能实验室、东方财经
安全榜 Flames 公平性、安全性、数据保护、合法性四大维度抗性 上海人工智能实验室、复旦大学
通识榜 BotChat 日常多轮对话生成，判断模型是否具备类人对话水平 上海人工智能实验室
法律榜 LawBench 法律问答、文本生成、判例分析等法律理解与推理 南京大学
医疗榜 MedBench 医学知识问答、安全伦理理解等医疗专业能力 上海人工智能实验室

2. 检索增强生成（Retrieval-Augmented Generation，RAG）

2.1 RAG 简介

大语言模型（LLM）在生成内容时，虽然具备强大的语言理解和生成能力，但也面临一些固有挑战：

  • 幻觉问题：LLM 有时会生成不准确或误导性的内容，即所谓的"幻觉"。
  • 知识时效性：模型的训练数据存在截止日期，面对最新信息时，生成结果的准确性难以保证。
  • 领域知识局限：对于特定领域的专业知识，LLM 的处理深度有限，难以覆盖复杂的垂直场景。

检索增强生成（RAG）技术正是为解决上述问题而生。RAG 在生成答案之前，首先从外部文档数据库中检索相关信息，并将其融入生成过程，从而指导和优化语言模型的输出。这一机制不仅显著提升了内容生成的准确性与时效性，还使答案具备可追溯性——生成内容建立在真实文档的基础之上，可信度更高。

RAG 的核心原理是将检索与生成解耦并串联：用户提出查询后，系统先通过检索模块找到相关文本片段，再将这些片段作为上下文传递给语言模型，模型据此生成更精准、更可靠的回答。

2.2 搭建一个 RAG 框架

本节将带领大家一步步实现一个简化版的 RAG 模型——Tiny-RAG。它保留了 RAG 的核心功能（检索与生成），旨在帮助大家深入理解 RAG 的原理与实现细节。

完整代码可在 Happy-LLM Chapter7 RAG 中找到。

Step 1：RAG 流程介绍

Tiny-RAG 由以下五个核心模块构成：

模块 功能
文档加载与切分模块 加载文档并切分为文档片段
向量化模块 将文档片段编码为向量表示
向量数据库 存储文档片段及其对应向量
检索模块 根据用户 Query 检索相关文档片段
大模型模块 基于检索结果回答用户问题

RAG 的整体流程分为三个阶段：

  1. 索引：将文档库分割为较短片段，通过编码器构建向量索引。
  2. 检索：根据问题与片段的语义相似度，检索最相关的文档片段。
  3. 生成：以检索到的上下文为条件，生成最终回答。

Step 2：文档加载与切分

文档加载模块支持 PDF、Markdown、TXT 等常见格式，根据文件扩展名自动选择对应的读取方法：

1
2
3
4
5
6
7
8
9
def read_file_content(cls, file_path: str):
    if file_path.endswith('.pdf'):
        return cls.read_pdf(file_path)
    elif file_path.endswith('.md'):
        return cls.read_markdown(file_path)
    elif file_path.endswith('.txt'):
        return cls.read_text(file_path)
    else:
        raise ValueError("Unsupported file type")

文档读取后需要进行切分。切分策略采用滑动窗口方式，设置最大 Token 长度（max_token_len=600）与重叠长度（cover_content=150），确保片段之间有内容重叠，以提升检索的连贯性与准确性。

切分逻辑说明：

  • 每块最多 600 token，其中 150 token 与上一块重叠，实际新增内容为 450 token。
  • 按换行符粗切分后，分三种情况处理：单行超长则强制拆块；当前行可并入则追加；否则保存当前块并开启新块。

Claude responded: 切分逻辑的三种情况详细说明如下：

切分逻辑的三种情况详细说明如下：

切分参数说明：max_token_len=600（每块上限），cover_content=150（块间重叠），实际每块新增内容为 token_len = 600 - 150 = 450 token。

情况 1：单行长度超过 max_token_len（600 token）

该行本身就超过了一个块的上限，无法整行放入任何块中，必须强制拆分。处理步骤如下：

  1. 若当前 curr_chunk 有内容，先将其保存到 chunk_text，然后清空。
  2. 将该行编码为 token 列表，按 token_len（450）为步长切成若干子块。
  3. 从第二个子块起，取上一块末尾的 150 个 token 作为前缀拼入，保证块间语义连贯。
  4. 每个子块保存到 chunk_text，处理完毕后重置 curr_chunk。

情况 2：当前行加入后，curr_chunk 总长不超过 token_len（450 token）

当前行可以安全并入正在构建的块，直接追加即可：

  1. 若 curr_chunk 已有内容，先追加换行符 \n（+1 token）。
  2. 将当前行拼入 curr_chunk，累加 curr_len。

情况 3：当前行加入后，curr_chunk 总长会超过 token_len（450 token）

当前块已满，需要收口并开启新块：

  1. 将当前 curr_chunk 保存到 chunk_text。
  2. 取刚保存块的末尾 150 个字符（cover_content）作为重叠前缀 cover_part。
  3. 新块初始内容为 cover_part + '\n' + 当前行，从而保留上下文衔接。
  4. 重置 curr_len 为新块的实际 token 数，继续处理后续行。

三种情况的判断优先级依次为：先判断单行是否超长（情况 1），再判断能否并入当前块（情况 2），最后才触发换块逻辑（情况 3）。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
import tiktoken
enc = tiktoken.get_encoding("cl100k_base")

def get_chunk(cls, text: str, max_token_len: int = 600, cover_content: int = 150):
    chunk_text = []
    curr_len = 0
    curr_chunk = ''
    token_len = max_token_len - cover_content
    lines = text.splitlines()
    for line in lines:
        line = line.strip()
        line_len = len(enc.encode(line))

        if line_len > max_token_len:
            # 情况1：单行超长，强制按 token_len 拆块
            if curr_chunk:
                chunk_text.append(curr_chunk)
                curr_chunk = ''
                curr_len = 0
            line_tokens = enc.encode(line)
            num_chunks = (len(line_tokens) + token_len - 1) // token_len
            for i in range(num_chunks):
                start_token = i * token_len
                end_token = min(start_token + token_len, len(line_tokens))
                chunk_part = enc.decode(line_tokens[start_token:end_token])
                if i > 0 and chunk_text:
                    prev_tokens = enc.encode(chunk_text[-1])
                    cover_part = enc.decode(prev_tokens[-cover_content:])
                    chunk_part = cover_part + chunk_part
                chunk_text.append(chunk_part)
            curr_chunk = ''
            curr_len = 0

        elif curr_len + line_len + 1 <= token_len:
            # 情况2：可并入当前块
            if curr_chunk:
                curr_chunk += '\n'
                curr_len += 1
            curr_chunk += line
            curr_len += line_len

        else:
            # 情况3：开启新块，添加重叠内容
            if curr_chunk:
                chunk_text.append(curr_chunk)
            if chunk_text:
                prev_chunk = chunk_text[-1]
                cover_part = prev_chunk[-cover_content:] if len(prev_chunk) > cover_content else prev_chunk
                curr_chunk = cover_part + '\n' + line
                curr_len = len(enc.encode(cover_part)) + 1 + line_len
            else:
                curr_chunk = line
                curr_len = line_len

    if curr_chunk:
        chunk_text.append(curr_chunk)

    return chunk_text

Step 3：向量化

向量化模块将文档片段编码为稠密向量，是 RAG 实现语义检索的基础。我们首先定义 BaseEmbeddings 基类，便于后续扩展不同的 Embedding 模型：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import numpy as np
from typing import List

class BaseEmbeddings:
    def __init__(self, path: str, is_api: bool) -> None:
        self.path = path
        self.is_api = is_api  # True：调用在线 API；False：加载本地模型
    def get_embedding(self, text: str, model: str) -> List[float]:
        raise NotImplementedError

    @classmethod
    def cosine_similarity(cls, vector1: List[float], vector2: List[float]) -> float:
        v1 = np.array(vector1, dtype=np.float32)
        v2 = np.array(vector2, dtype=np.float32)
        if not np.all(np.isfinite(v1)) or not np.all(np.isfinite(v2)):
            return 0.0
        dot_product = np.dot(v1, v2)
        magnitude = np.linalg.norm(v1) * np.linalg.norm(v2)
        if magnitude == 0:
            return 0.0
        return dot_product / magnitude
        # 数学公式：cos θ = A·B / (|A||B|)
        # 语义相近的文本，其向量方向接近，余弦值趋近于 1

BaseEmbeddings 提供两个核心方法：get_embedding 获取文本向量表示（子类必须实现），cosine_similarity 计算两向量间的余弦相似度（继承复用）。

Step 4：向量数据库与检索

向量数据库负责存储文档片段及其向量，并提供检索接口。完整代码见 VectorBase.py。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
class VectorStore:
    def __init__(self, document: List[str] = ['']) -> None:
        self.document = document

    def get_vector(self, EmbeddingModel: BaseEmbeddings) -> List[List[float]]:
        # 将所有文档片段转为向量
        pass

    def persist(self, path: str = 'storage'):
        # 持久化保存向量数据库到本地
        pass

    def load_vector(self, path: str = 'storage'):
        # 从本地加载向量数据库
        pass

    def query(self, query: str, EmbeddingModel: BaseEmbeddings, k: int = 1) -> List[str]:
        # 将 query 向量化后，与数据库中所有向量计算相似度，返回 top-k 片段
        query_vector = EmbeddingModel.get_embedding(query)
        result = np.array([self.get_similarity(query_vector, vector) for vector in self.vectors])
        return np.array(self.document)[result.argsort()[-k:][::-1]].tolist()

query 方法的核心逻辑：将用户问题向量化，与数据库中所有片段向量逐一计算余弦相似度，返回相似度最高的 top-k 个文档片段。

Step 5：大模型模块

大模型模块负责基于检索结果生成最终回答。同样采用基类设计，便于扩展本地模型或不同 API 服务：

1
2
3
4
5
6
7
8
9
10
class BaseModel:
    def __init__(self, path: str = '') -> None:
        self.path = path

    def chat(self, prompt: str, history: List[dict], content: str) -> str:
        pass

    def load_model(self):
        # 本地开源模型需实现此方法；API 模型无需实现
        pass

以硅基流动平台的 Qwen2.5 为例，实现 RAG 专用的对话类：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
from openai import OpenAI

class OpenAIChat(BaseModel):
    def __init__(self, model: str = "Qwen/Qwen2.5-32B-Instruct") -> None:
        super().__init__()
        self.model = model

    def chat(self, prompt: str, history: List[dict], content: str) -> str:
        RAG_PROMPT_TEMPLATE = """
        使用以上下文来回答用户的问题。如果你不知道答案，就说你不知道。总是使用中文回答。
        问题: {question}
        可参考的上下文：
        ···
        {context}
        ···
        如果给定的上下文无法让你做出回答，请回答数据库中没有这个内容，你不知道。
        有用的回答:
        """
        client = OpenAI()
        client.api_key = os.getenv("OPENAI_API_KEY")
        client.base_url = os.getenv("OPENAI_BASE_URL")
        history.append({'role': 'user', 'content': RAG_PROMPT_TEMPLATE.format(question=prompt, context=content)})
        response = client.chat.completions.create(
            model=self.model,
            messages=history,
            max_tokens=2048,
            temperature=0.1
        )
        return response.choices[0].message.content

RAG 专用提示词模板的关键设计在于：明确要求模型仅基于提供的上下文作答，若上下文不足则如实告知，避免模型"发挥"产生幻觉。

Step 6：Tiny-RAG 完整示例

首次运行（构建并保存数据库）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 1. 加载文档并切分
docs = ReadFiles('./data').get_content(max_token_len=600, cover_content=150)

# 2. 构建向量数据库
vector = VectorStore(docs)
embedding = OpenAIEmbedding()
vector.get_vector(EmbeddingModel=embedding)
vector.persist(path='storage')  # 持久化到本地，下次可直接加载

# 3. 检索 + 生成
question = 'RAG 的原理是什么？'
content = vector.query(question, EmbeddingModel=embedding, k=1)[0]
chat = OpenAIChat(model='Qwen/Qwen2.5-32B-Instruct')
print(chat.chat(question, [], content))

再次运行（加载已有数据库）：

1
2
3
4
5
6
7
8
9
# 直接加载本地数据库，无需重新向量化
vector = VectorStore()
vector.load_vector('./storage')

question = 'RAG 的原理是什么？'
embedding = ZhipuEmbedding()
content = vector.query(question, EmbeddingModel=embedding, k=1)[0]
chat = OpenAIChat(model='Qwen/Qwen2.5-32B-Instruct')
print(chat.chat(question, [], content))

3. Agent

3.1 什么是 LLM Agent？

大模型 Agent 是一个以 LLM 为核心"大脑"，并赋予其自主规划、记忆和工具使用能力的系统。它不再仅仅被动地响应用户的提示，而是能够：

  1. 理解目标（Goal Understanding）：接收相对复杂或高层次的目标（例如"帮我规划一个周末去北京的旅游行程并预订机票酒店"）。
  2. 自主规划（Planning）：将大目标分解为一系列可执行的小步骤（如"搜索北京景点"、"查询天气"、"比较机票价格"、"调用预订 API"等）。
  3. 记忆（Memory）：拥有短期记忆（记住当前任务的上下文）和长期记忆（从历史交互或外部知识库中检索信息）。
  4. 工具使用（Tool Use）：调用外部 API、插件或代码执行环境来获取信息、执行操作或进行计算。
  5. 反思与迭代（Reflection & Iteration）：（高级 Agent）评估自身行为与结果，从中学习并调整后续计划。

传统 LLM 像一个知识渊博但只能纸上谈兵的图书馆员，而 LLM Agent 则更像一个全能的私人助理——不仅懂得多，还能跑腿办事，甚至能主动思考最优方案。

3.2 LLM Agent 的类型

  • 任务导向型（Task-Oriented）
    • 特点：专注完成特定领域的明确任务，有预设流程和固定工具集，LLM 主要负责理解意图与调用工具
    • 示例：餐厅预订机器人、GitHub Copilot 代码助手
  • 规划与推理型（Planning & Reasoning）
    • 特点：强调自主分解复杂任务、制定多步计划，常采用 ReAct（Reason+Act）或 CoT 思维框架
    • 示例：整合搜索、计算器、数据库的研究型 Agent
  • 多 Agent 系统（Multi-Agent）
    • 特点：多个不同角色的 Agent 协同工作，可通信、协作、辩论甚至竞争
    • 示例：AutoGen、ChatDev（模拟软件开发团队）
  • 探索与学习型（Exploration & Learning）
    • 特点：在与环境交互中主动学习新知识、优化策略，包含复杂的记忆与反思机制
    • 示例：自主探索未知软件环境的 Agent、游戏策略 Agent

3.3 动手构造一个 Tiny-Agent

本节基于 openai 库的 tool_calls 功能，构造一个简单的任务导向型 Agent，能够根据用户输入调用工具并回答问题。

效果如图所示：

Step 1：初始化客户端和模型

使用 openai 库，配置指向兼容 OpenAI API 的服务终端（以 SiliconFlow 为例）：

1
2
3
4
5
6
7
8
from openai import OpenAI

client = OpenAI(
    api_key="YOUR_API_KEY",        # 替换为你的 API Key
    base_url="https://api.siliconflow.cn/v1",
)

model_name = "Qwen/Qwen2.5-32B-Instruct"

Step 2：定义工具函数

在 src/tools.py 中定义 Agent 可调用的工具函数。每个函数需要有清晰的文档字符串（docstring），描述其功能和参数——这将用于自动生成工具的 JSON Schema：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
# src/tools.py
from datetime import datetime
import wikipedia

def get_current_datetime() -> str:
    """
    获取当前日期和时间。
    :return: 当前日期和时间的字符串表示。
    """
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

def count_letter_in_string(a: str, b: str) -> str:
    """
    统计字符串中某个字母的出现次数。
    :param a: 要搜索的字符串。
    :param b: 要统计的字母。
    :return: 字母在字符串中出现的次数。
    """
    return str(a.count(b))

def search_wikipedia(query: str) -> str:
    """
    在维基百科中搜索指定查询的前三个页面摘要。
    :param query: 要搜索的查询字符串。
    :return: 包含前三个页面摘要的字符串。
    """
    page_titles = wikipedia.search(query)
    summaries = []
    for page_title in page_titles[:3]:
        try:
            wiki_page = wikipedia.page(title=page_title, auto_suggest=False)
            summaries.append(f"页面: {page_title}\n摘要: {wiki_page.summary}")
        except (wikipedia.exceptions.PageError, wikipedia.exceptions.DisambiguationError):
            pass
    return "\n\n".join(summaries) if summaries else "维基百科没有搜索到合适的结果"

为让 API 理解这些工具，需通过 src/utils.py 中的 function_to_json 将函数转换为 JSON Schema：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# src/utils.py
import inspect

def function_to_json(func) -> dict:
    return {
        "type": "function",
        "function": {
            "name": func.__name__,
            "description": inspect.getdoc(func),
            "parameters": {
                "type": "object",
                "properties": parameters,
                "required": required,
            },
        },
    }

Step 3：构造 Agent 类

在 src/core.py 中定义 Agent 类，负责管理对话历史、调用 API、处理工具请求和执行工具函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
class Agent:
    def __init__(self, client: OpenAI, model: str = "Qwen/Qwen2.5-32B-Instruct",
                 tools: List = [], verbose: bool = True):
        self.client = client
        self.tools = tools       # Python 函数对象列表
        self.model = model
        self.messages = [{"role": "system", "content": SYSTEM_PROMPT}]
        self.verbose = verbose

    def get_tool_schema(self) -> List[Dict]:
        return [function_to_json(tool) for tool in self.tools]

    def handle_tool_call(self, tool_call):
        function_name = tool_call.function.name       # 函数名
        function_args = tool_call.function.arguments  # 函数参数（JSON 字符串）
        function_id   = tool_call.id                  # 调用 ID，用于追踪本次调用

        result = eval(f"{function_name}(**{function_args})")  # 执行函数

        return {"role": "tool", "content": result, "tool_call_id": function_id}

    def get_completion(self, prompt) -> str:
        # 1. 加入用户消息
        self.messages.append({"role": "user", "content": prompt})

        # 2. 第一次调用模型，判断是否需要工具
        response = self.client.chat.completions.create(
            model=self.model,
            messages=self.messages,
            tools=self.get_tool_schema(),
            stream=False,
        )
        # 将用户输入交给model，如果判断需要使用工具，
        # 例如：
        # 用户：北京天气怎么样
        # 模型会判断：
        # 我不能凭空知道实时天气
        # 我应该调用 get_weather
        # 于是返回：
        # message.tool_calls = [
        #   {
        #     "function":{
        #       "name":"get_weather",
        #       "arguments":"{\"city\":\"北京\"}"
        #     }
        #   }
        # ]
        # 模型不会真正执行函数，只会在 tool_calls 中输出"我要调用哪个函数、传什么参数"

        # 3. 若模型决定调用工具
        if response.choices[0].message.tool_calls:
            self.messages.append({"role": "assistant", "content": response.choices[0].message.content})

            tool_list = []
            for tool_call in response.choices[0].message.tool_calls:
                self.messages.append(self.handle_tool_call(tool_call))  # 执行工具并将结果写入 messages
                tool_list.append([tool_call.function.name, tool_call.function.arguments])

            # 此时 messages 结构大致为：
            # [{"role":"user",      "content":"北京天气怎么样"},
            #  {"role":"assistant", "content": None, "tool_calls":[...]},
            #  {"role":"tool",      "content":"25度", "tool_call_id":"..."}]

            if self.verbose:
                print("调用工具：", response.choices[0].message.content, tool_list)

            # 4. 携带工具结果，第二次调用模型生成最终回答
            response = self.client.chat.completions.create(
                model=self.model,
                messages=self.messages,
                tools=self.get_tool_schema(),
                stream=False,
            )

        # 5. 将最终回答写入 messages 并返回
        self.messages.append({"role": "assistant", "content": response.choices[0].message.content})
        return response.choices[0].message.content

Agent 的完整工作流程：

  1. 接收用户输入，加入对话历史。
  2. 携带工具 Schema 调用大模型，模型判断是否需要工具。
  3. 若需要，解析模型输出的工具调用请求，执行对应 Python 函数。
  4. 将工具执行结果写回对话历史，再次调用模型。
  5. 模型根据工具结果生成最终回复，返回给用户。

Step 4：运行 Agent

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# demo.py
if __name__ == "__main__":
    client = OpenAI(
        api_key="YOUR_API_KEY",
        base_url="https://api.siliconflow.cn/v1",
    )

    agent = Agent(
        client=client,
        model="Qwen/Qwen2.5-32B-Instruct",
        tools=[get_current_datetime, add, compare, count_letter_in_string],
        verbose=True,
    )

    while True:
        prompt = input("\033[94mUser: \033[0m")      # 蓝色显示用户输入
        if prompt.lower() == "exit":
            break
        response = agent.get_completion(prompt)
        print("\033[92mAssistant: \033[0m", response) # 绿色显示回答

本节实现的 Agent 类型说明

本节构造的属于最基础的 Tool-Calling Agent，其本质特征是：

  • 只负责"调用工具 + 拼接结果"，不更新模型参数
  • 不改变自身能力，不保存长期经验（上下文仅限当次 messages）
  • 每次对话都是"新大脑"，不会随使用变得更聪明

这是理解更复杂 Agent 架构（如带记忆的 Agent、多 Agent 系统）的基础。