Agent的学习记录--3

来源Github的学习指南Agent Learning Hub的学习：

Stage 1—— 构建最小 Agent Loop

Agent里的role定义，有三种类型，分别是system, user, assistant，对于消息数组里的一个或多个信息，每个信息必须要与三个角色中的一个所关联。

角色

system

这个角色有助于通过分配特定行为给聊天助手来创建对话的上下文或者范围。比如我想建立一个可以帮我搜索论文，帮我解答我的科学问题的Agent，我首先最简单的，先将system这个角色分配给了聊天助手：

messages = [
    {
        "role": "system",
        "content": "你是一个精通学术论文学术助手，你精通数学和计算机科学，需要回答与论文相关的问题，简明扼要。"
    }
]

虽然传递带有system的角色的消息并非必需，但它有助于在内部为对话设置模型行为。

user

这个角色实际上代表了我们的用户，他会向LLM模型发送提示词，令其做出回复。我们使用这个角色来对话分配上下文。

user_query = "请用一句话解释什么是 Chain-of-Thought (思维链)？"
messages.append({"role": "user", "content": user_query})

assistant

这个角色代表响应最终用户提示的实体。其用于在当前请求中设置模型的先前响应，以保持对话的连贯性。

response = client.chat.completions.create(
    model=MODEL_NAME,
    messages=messages
)
assistant_reply = response.choices[0].message.content
messages.append({"role": "assistant", "content": assistant_reply})

这个样子我们就完成了一轮对话，全部代码为：

在这里我们设置了.env, 其内有配置环境，还有OPENAI_API_KEY, OPENAI_BASE_URL, MODEL_NAME。下面只是一个示例

from openai import OpenAI

def main():
    client = OpenAI(api_key="<DeepSeek API Key>", base_url="https://api.deepseek.com")
    
    messages = [
        {
            "role": "system",
            "content": "你是一个精通学术论文学术助手，你精通数学和计算机科学，需要回答与论文相关的问题，简明扼要。"
        }
    ]
    
    # 第一轮交互
    user_query = "请用一句话解释什么是 Chain-of-Thought (思维链)？"
    print(f"用户: {user_query}")
    messages.append({"role": "user", "content": user_query})
    
    response = client.chat.completions.create(
        model="deepseek-v4-flash",
        messages=messages
    )
    
    assistant_reply = response.choices[0].message.content
    print(f"\n助手: {assistant_reply}\n")
    messages.append({"role": "assistant", "content": assistant_reply})

    messages.append({"role": "user", "content": "思维链是什么时候，由谁提出的？"})
    response = client.chat.completions.create(
        model="deepseek-v4-flash",
        messages=messages
    )
    assistant_reply = response.choices[0].message.content
    print(f"\n助手: {assistant_reply}\n")
    messages.append({"role": "assistant", "content": assistant_reply})

    print(f"对话历史: {messages}")

if __name__ == "__main__":
    main()

最后输出为：

对话历史: [{'role': 'system', 'content': '你是一个精通学术论文学术助手，你精通数学和计算机科学，需要回答与论文相关的问题，简明扼要。'}, {'role': 'user', 'content': '请用一句话解释什么是 Chain-of-Thought (思维链)？'}, {'role': 'assistant', 'content': '思维链是一种引导大语言模型在推理时显式地生成中间步骤（逐步推理）以提升复杂问题解答准确性的提示技术。'}, {'role': 'user', 'content': '思维链是什么时候，由谁提出的？'}, {'role': 'assistant', 'content': '思维链由谷歌大脑（Google Brain）的研究团队在2022年1月发表的论文 *“Chain-of-Thought Prompting Elicits Reasoning in Large Language Models”* 中首次提出，主要作者包括Jason Wei等。'}]

这样子我们就使用了LLM完成了一个普通的对话。

结构化输出

在输出的时候，我们在很多时候并不希望直接输出的是一大段的文字，且想考虑一个情况，比如在多Agent的工作流里，下一个Agent也不想收到一大段的文字，这既会使注意力分散，还会导致上下文窗口的减少，因此更好的情况是，我们输出一些结构化的数据，使得可以更加好的获得我们所要的信息。这时候我们就可以使用json进行一个结构化的输出，对于json，其是一种数据交换的格式，基于文本，易于解析和传输。

现在这个是我们的文本信息

sample_text = """
  Transformer 是一种基于自注意力机制 (Self-Attention) 的深度学习架构，由 Vaswani 等人在 2017 年提出。
  它摒弃了传统的循环神经网络 (RNN) 顺序计算方式，允许全序列并行计算。核心机制包括多头注意力 (Multi-Head Attention)、
  位置编码 (Positional Encoding) 和残差连接。在 NLP、CV 以及多模态领域均成为主流基础模型。
  """

我们想要在其中获得一些信息，比如我想做一个slides，那么我就需要标题，要点还有一些需要被重点标出的内容，所以我们在提示词里写道

    prompt = f"""
请分析以下文本，并提取出用于生成 LaTeX Beamer 演示文稿的结构化大纲。
必须输出符合以下格式的 JSON 对象：
{{
  "presentation_title": "幻灯片总标题",
  "slides": [
    {{
      "slide_number": 1,
      "title": "单页标题",
      "key_points": ["要点1", "要点2"],
      "latex_highlight": "建议在 Beamer 中高亮或用公式展示的概念"
    }}
  ]
}}

文本内容如下：
{sample_text}
"""

对比起上面的最小对话，我们只需要修改一些地方即可以实现结构化的输出：

response = client.chat.completions.create(
    model=MODEL_NAME,
    messages=messages,
    response_format={"type": "json_object"}  # 强制 JSON 模式
)

对于response_format，这里面有两种类型，分别为 json_object 和 json_schema，对于前者，其使用方法很简单，而对于后者，其有较为苛刻的条件，他需要写清楚json的限制性格式条件，以及需要模型支持输出，比如Deepseek，其输出并不支持 json_schema，对于后者，我们写一个示例：

json_schema = {
    "type": "object",
    "properties":{
        "presentation_title": {"type": "string"},
        "slides":{
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "slide_number": {"type": "integer"},
                    "title": {"type": "string"},
                    "key_points": {
                    	"type": "array",
                    	"items": {"type": "string"}
                	},
                	"latex_highlight": {"type": "string"}
                },
                "required": ["slide_number", "title", "key_points", "latex_highlight"]
            }
        }
    },
    "required": ["presentation_title", "slides"]
}

response = client.chat.completions.create(
    model=MODEL_NAME,
    messages=messages
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "presentation_outline",
            "schema": json_schema,
            "strict": True
        }
    }
)

对于不支持json_schema的模型，我们后面可以通过Function Call/Tool Call得以实现

全部代码如下：

import json
from config import get_client, MODEL_NAME

def main():
    client = get_client()
    
    sample_text = """
    Transformer 是一种基于自注意力机制 (Self-Attention) 的深度学习架构，由 Vaswani 等人在 2017 年提出。
    它摒弃了传统的循环神经网络 (RNN) 顺序计算方式，允许全序列并行计算。核心机制包括多头注意力 (Multi-Head Attention)、
    位置编码 (Positional Encoding) 和残差连接。在 NLP、CV 以及多模态领域均成为主流基础模型。
    """
    
    # 提示词中明确要求输出 JSON，并指定期望的 Schema 结构
    prompt = f"""
请分析以下文本，并提取出用于生成 LaTeX Beamer 演示文稿的结构化大纲。
必须输出符合以下格式的 JSON 对象：
{{
  "presentation_title": "幻灯片总标题",
  "slides": [
    {{
      "slide_number": 1,
      "title": "单页标题",
      "key_points": ["要点1", "要点2"],
      "latex_highlight": "建议在 Beamer 中高亮或用公式展示的概念"
    }}
  ]
}}

文本内容如下：
{sample_text}
"""
    
    messages = [
        {"role": "system", "content": "你是一个学术内容提取助手。你必须始终输出合法的 JSON 格式字符串，不要输出任何额外的解释或 markdown 标记。"},
        {"role": "user", "content": prompt}
    ]
    
    response = client.chat.completions.create(
        model=MODEL_NAME,
        messages=messages,
        response_format={"type": "json_object"}  # 强制 JSON 模式
    )
    
    raw_content = response.choices[0].message.content
    print("【模型原始输出字符串】:")
    print(raw_content)
    
    # 解析验证 JSON
    try:
        data = json.loads(raw_content)
        print(f"演示文稿标题: {data.get('presentation_title')}")
        for slide in data.get("slides", []):
            print(f"  - 第 {slide.get('slide_number')} 页: {slide.get('title')}")
            for pt in slide.get("key_points", []):
                print(f"      * {pt}")
            print(f"      [建议高亮]: {slide.get('latex_highlight')}")
    except json.JSONDecodeError as e:
        print(f"JSON 解析失败: {e}")

if __name__ == "__main__":
    main()

输出为：

演示文稿标题: Transformer 架构详解
  - 第 1 页: 引言：Transformer 概述
      * Transformer 是一种基于自注意力机制 (Self-Attention) 的深度学习架构，由 Vaswani 等人于 2017 年提出。
      * 它摒弃了传统 RNN 的顺序计算方式，支持全序列并行计算。
      [建议高亮]: Self-Attention
  - 第 2 页: 核心机制
      * 多头注意力 (Multi-Head Attention) 允许模型关注不同位置的信息。
      * 位置编码 (Positional Encoding) 为序列提供顺序信息。
      * 残差连接 (Residual Connections) 有助于训练深层网络。
      [建议高亮]: Multi-Head Attention, Positional Encoding, Residual Connections
  - 第 3 页: 应用与影响
      * 在 NLP、CV 以及多模态领域均成为主流基础模型。
      * 推动了预训练模型的发展，如 BERT、GPT 等。
      [建议高亮]: Transformer is the foundation of modern AI

Function call

定义：也叫做Tool call，其是大模型厂商提供的一项能力，它允许开发者预先定义一组函数或工具的名称、用途、入参要求、出参格式，大模型会根据用户的需求（即提示词），自主决策是否需要调用函数、调用什么函数、生成对应的入参（需要做到结构化输出），开发者拿到模型生成的参数后，执行对应的函数，再把执行结果返回给大模型，最终由大模型整理成自然语言结果返回给用户。

对于一个Function call，其应该有6个环节：工具定义、模型决策调用、参数生成、工具执行、结果输出、模型解析

对于一个Function call格式：其本质上是一个json schema

{
	"type": "function",
	"function": {
		"name": "tool_1",
		"description": "What can this tool do",
		"parameters": {
			"type": "object",
			"properties":{
				"argument_1":{
					"type": "string",
					"description": "the argument means"
				},
            	"argument_2":{
            		"type": "integer",
            		"description": "the argument means"
            	}
			},
			"required": ["argument_1","argument_2"]
		}
	}
}

name：函数或者工具的名称，其必须是唯一的，只能包含字母、数字、下划线

parameters：函数的入参定义，遵循JSON Schema规范，明确每个参数的类型、描述

当写好这个格式之后，我们就可以开始写一个自己的工具，其就如平时写函数一样，

现在我写了一个读取文件的函数以及该工具的格式要求

import json
from config import get_client, MODEL_NAME

def read_paper_file(filepath: str) -> str:
    """读取指定路径下内容"""
    try:
        with open(filepath, "r", encoding="utf-8") as f:
            return f.read()
    except FileNotFoundError:
        return f"错误：文件 {filepath} 不存在。"
    except Exception as e:
        return f"读取出错: {str(e)}"
    
tools_schema = [
    {
        "type": "function",
        "function": {
            "name": "read_paper_file",
            "description": "当需要阅读或分析某篇本地论文/文档的内容时调用此工具。",
            "parameters": {
                "type": "object",
                "properties": {
                    "filepath": {
                        "type": "string",
                        "description": "文件路径，必须是本地可访问的文本文件路径，例如 sample_paper.txt"
                    }
                },
                "required": ["filepath"]
            }
        }
    }
]

然后我们写出一个完整的一次性调用的程序

def main():
    client = get_client()
    
    user_query = "请帮我阅读 sample_paper.txt 这篇论文，告诉我它主要研究了什么？"
    messages = [
        {"role": "system", "content": "你是一个学术助手，能够根据用户提问来解读学术内容。"},
        {"role": "user", "content": user_query}
    ]
    
    print(f"用户提问: {user_query}\n")
    
    response = client.chat.completions.create(
        model=MODEL_NAME,
        messages=messages,
        tools=tools_schema,
        tool_choice="auto"  # 让模型自主选择是否调用
    )
    
    response_message = response.choices[0].message
    if response_message.tool_calls:
        print("【模型决策】需要调用外部工具！")
        for tool_call in response_message.tool_calls:
            func_name = tool_call.function.name
            func_args_str = tool_call.function.arguments
            call_id = tool_call.id
            
            print(f"  - 工具调用 ID (call_id): {call_id}")
            print(f"  - 调用的函数名: {func_name}")
            print(f"  - 模型生成的实参 JSON: {func_args_str}")
            
            # 解析实参
            args = json.loads(func_args_str)
            print(f"  - 解析后的实参对象: {args}")
            print(f"  - 回复{response_message.content}")
    else:
        print("【模型决策】不需要调用工具，直接回复：")
        print(response_message.content)

if __name__ == "__main__":
    main()

我们可以检验其是否真的调用了工具

最终输出为

【模型决策】需要调用外部工具！
  - 工具调用 ID (call_id): call_00_spOcyzoFEb4q1kSoSyow4153
  - 调用的函数名: read_paper_file
  - 模型生成的实参 JSON: {"filepath": "sample_paper.txt"}
  - 解析后的实参对象: {'filepath': 'sample_paper.txt'}
  - 回复'我来帮你阅读这篇论文，请稍等。'

我们会发现，这个聊天到这里就结束了，并没有后面详细解读文档后的内容，回忆Agent的工作循环，Observe，即提示词的传入，Think，其判断是否需要调用工具，Act，需要调用工具，然后传参，因为没有循环的原因，所以其直接就传出了。所以其没有把读出的文件内容再次发回给大模型，所以大模型根本还没拿到论文内容就结束了。

对于工具ID，tool_call.id是模型服务端随机/动态生成的唯一标识符（UUID / 随机哈希字符串），每次模型决定发起一次工具调用，服务端都会现场生成一个全新的 ID。这个ID在并行Agent运行很重要，因为其不是串流性工作，对于有多份文件的时候，其每次读取文件的返回结果都会伴随着独特的一个ID，作为识别号，这样结果才能准确对齐。

所以我们需要添加一个循环，让这个进程得以继续运行下去：

首先简单的，我们得先对下一个回答的文本进行获取，此时需要调用工具，在调用工具前，我们肯定要先发送 assistant的消息，然后调用 tool，对于tool_call里，首先我们上面知道了，其传入的参数是个json，首先要先解析json获得这个path，然后传入得到的文本就是我们下次在发送的文本（在我的理解里，user只需要发送一次，后面跟assistant发送的是 tool。

修改后的代码如下

def main():
    client = get_client()
    
    user_query = "请帮我阅读 sample_paper.txt 这篇论文，告诉我它主要研究了什么？"
    messages = [
        {"role": "system", "content": "你是一个学术助手，能够根据用户提问来解读学术内容。"},
        {"role": "user", "content": user_query}
    ]
    
    print(f"用户提问: {user_query}\n")

    max_steps = 5 # 限制最大步数
    for step in range(max_steps):
        response = client.chat.completions.create(
            model=MODEL_NAME,
            messages=messages,
            tools=tools_schema,
            tool_choice="auto"
        )

        response_message = response.choices[0].message

        if not response_message.tool_calls:
            print("\n It is not need to call any tools")
            print(f"\n {response_message.content}")
            break
        print(f"\n step {step + 1} is calling tools")
        messages.append(response_message)

        for tool_call in response_message.tool_calls:
            func_args_str =tool_call.function.arguments
            text_file_path = json.loads(func_args_str).get("filepath")
            text_content =read_paper_file(text_file_path)
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "name": tool_call.function.name,
                "content": text_content
            })
            
if __name__ == "__main__":
    main()

然后我们就可以得到了一个回答：

**作者**：Jason Wei、Xuezhi Wang、Dale Schuurmans、Maarten Bosma、Ed Chi、Quoc Le、Denny Zhou（Google Research 团队）

## 主要研究内容

这篇论文探讨的核心问题是：**如何让大语言模型更好地完成复杂的多步推理任务**。

### 提出的方法：思维链（Chain-of-Thought，CoT）提示

论文提出了一种简单而强大的方法——**思维链提示**。其核心思想是：

- **传统标准提示**：直接将输入问题喂给模型，让它预测最终答案。这种方式在处理简单检索问题上有较好表现，但在需要多步推理的任务上效果不佳。
- **思维链提示**：在提示中提供少量包含"中间推理步骤"的示例（exemplars），引导模型将复杂问题**分解为一系列连续的中间推理步骤**，最终再输出答案。
- **数学表达**：传统方法建模为 P(A|Q)，而思维链方法建模为 P(C, A|Q) = P(C|Q) × P(A|Q,C)，其中 C 表示中间的推理链。

### 关键发现：涌现能力（Emergent Ability）

论文的一个重要发现是：思维链推理能力是一种**涌现能力**——它不会在较小的模型（参数量 <10B）上带来显著提升，但当模型规模超过约 1000 亿（100B）参数时，这种推理能力会**急剧涌现**。这说明这种能力在很大程度上依赖于足够大的模型规模。

### 典型应用领域

论文通过以下三个领域验证了思维链方法的效果：
1. **算术推理**：数学应用题（如 GSM8K 数据集），需要多步计算。
2. **常识推理**：如 StrategyQA、体育常识理解、日期推理等。
3. **符号推理**：字母末位拼接、硬币翻转追踪等。

## 结论

思维链提示是一种简单却强大的机制，它无需针对特定任务进行微调，也无需修改模型架构，就能有效促进语言模型完成多步推理任务。

---

总体而言，这篇论文是**大语言模型推理能力研究领域的里程碑之作**，它提出了"思维链提示"这一如今被广泛使用的技术，并揭示了推理能力随模型规模涌现这一重要规律。需要我进一步分析论文中的某个具体方面吗？

错误处理

超时处理

只需要在client.chat.completions.create() 里添加参数即可：

response = client.chat.completions.create(
    model=MODEL_NAME,
    messages=messages,
    tools=tools_schema,
    tool_choice="auto",
    timeout=30 # 超过30s报错超时
)

API错误

在client.chat.completions.create() 后运行，通过try语句，使用openai里的APIError类即可

max_steps = 5
for step in range(max_steps):
    try:
        response = client.chat.completions.create(
        model=MODEL_NAME,
        messages=messages,
        tools=tools_schema,
        tool_choice="auto",
        timeout=30 # 设置超时时间，避免长时间等待
        )
    except APIError as e:
        print(f"API 调用出错: {e}")
        break
    except Exception as e:
        print(f"发生未知错误: {e}")
        break

json无法解析（工具错误）

Agent 的黄金法则：工具运行出错时，千万不要直接让 Python 程序 crash，而是把错误信息变成字符串喂给模型！

首先，大模型拥有自我纠错的能力，将错误信息当作message传入下一次的输入，模型在下次响应的时候就会有所改变，对于上面的Function call，我们改一下就可以实现

for tool_call in response_message.tool_calls:
    try:
        func_args_str =tool_call.function.arguments
        text_file_path = json.loads(func_args_str).get("filepath")
    except Exception as e:
        text_content = f"解析工具调用参数出错: {str(e)}"
        text_file_path = None
        if text_file_path:
            try:
                text_content =read_paper_file(text_file_path)
            except Exception as e:
                text_content = f"读取文件出错: {str(e)}"
                
        messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "name": tool_call.function.name,
            "content": text_content
        })

次数迭代越出

只需要最后输出错误终止Agent即可，在开始对话循环之前加一个if-else即可

错误总结

所以最终代码为：

def main():
    client = get_client()
    
    user_query = "请帮我阅读 sample_paper.txt 这篇论文，告诉我它主要研究了什么？"
    messages = [
        {"role": "system", "content": "你是一个学术助手，能够根据用户提问来解读学术内容。"},
        {"role": "user", "content": user_query}
    ]
    
    print(f"用户提问: {user_query}\n")

    max_steps = 5
    for step in range(max_steps):
        try:
            response = client.chat.completions.create(
                model=MODEL_NAME,
                messages=messages,
                tools=tools_schema,
                tool_choice="auto",
                timeout=30 # 设置超时时间，避免长时间等待
            )
        except APIError as e:
            print(f"API 调用出错: {e}")
            break
        except Exception as e:
            print(f"发生未知错误: {e}")
            break

        response_message = response.choices[0].message

        if not response_message.tool_calls:
            print("\n It is not need to call any tools")
            print(f"\n {response_message.content}")
            break
        print(f"\n step {step + 1} is calling tools")
        messages.append(response_message)

        for tool_call in response_message.tool_calls:
            try:
                func_args_str =tool_call.function.arguments
                text_file_path = json.loads(func_args_str).get("filepath")
            except Exception as e:
                text_content = f"解析工具调用参数出错: {str(e)}"
                text_file_path = None
            if text_file_path:
                try:
                    text_content =read_paper_file(text_file_path)
                except Exception as e:
                    text_content = f"读取文件出错: {str(e)}"
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "name": tool_call.function.name,
                "content": text_content
            })
    else:
        print(f"\n 达到最大步骤数 {max_steps}，已停止回答。")

在我修改为sample_paper.txt 为 sample_par.txt后，经过两轮的迭代，仍然可以正确读取文件并进行后面的输出。而直接修改为 txt之后，我一次在偶然间大模型只需要一轮的迭代即可正确输出，但还有一次其并没有调用工具，而是直接返回了输出：

 我需要先找到并阅读这篇论文文件。请问这篇 txt 文件的具体路径是什么？比如文件名是 `sample_paper.txt` 还是其他名称？您可以提供确切的文件路径，我来帮您阅读分析。