在做RAG知识增强或者微调模型的时候，很多团队都会碰到一个共同的问题：直接拿大段文本往里灌，效果往往不尽如人意。相比之下，结构化的问答对要高效得多——无论是存入向量数据库，还是作为微调的训练语料。

那么问题来了，怎么从杂乱的长文本里，批量地、自动化地生成这些问答对？尤其麻烦的是，这些问答对必须是结构化的，能存进一个list里，每个元素都是格式统一的dict。这篇文章就基于Agently框架，给出了一条非常直接的实现路径。

案例概述

先看整体流程。数据流大致是：把一份文档（本地或线上）切分成若干文本块，然后对每一块调用大模型生成问答对，最后把结果拼接起来。整个过程的控制精度很高——文本块大小可以自定义，请求间隔可控，输出的数据结构也是预先定义好的。

下面是完整的代码样例。注意，这里的模型建议使用有16k或更大上下文窗口的版本，这样在处理长段落时不容易丢信息。

import Agently
import requests
import time

# Model Settings
agent_factory = Agently.AgentFactory()
.set_settings("model.OpenAI.auth", { "api_key": "" })
.set_settings("model.OpenAI.options", { "model": "gpt-3.5-turbo-16k" })
# recommend using 16k or larger context model for this kind of tasks

# Download document
document_link = "https://raw.githubusercontent.com/Maplemx/Agently/main/README.md"
document_content = ""
response = requests.get(document_link)
if response.status_code == 200:
    document_content = response.content.decode("utf-8")

# Work Settings
piece_length_control = 1000
sleep_time = 5 # sleep for a while in case of reaching API request limit

# Chop document
chunks = document_content.split("

")
paragraphs = []
paragraph_num = -1
for chunk in chunks:
    if chunk.startswith("#"):
        paragraphs.append(chunk + "

")
        paragraph_num += 1
    else:
        paragraphs[paragraph_num] += chunk + "

"

text_pieces = []
text_piece_num = 0
for paragraph in paragraphs:
    if len(text_pieces) == 0:
        text_pieces.append(paragraph)
    else:
        if len(text_pieces[text_piece_num] + paragraph) > piece_length_control:
            text_pieces.append(paragraph)
            text_piece_num += 1
        else:
            text_pieces[text_piece_num] += paragraph

# Generate QA Pairs
qa_pairs = []
agent = agent_factory.create_agent()
for text_piece in text_pieces:
    print("[Working on]: ", text_piece.split("
")[0])
    result = agent
        .input({"text": text_piece })
        .instruct("Generate at least 5 question and answer pairs about {text}")
        .output([{
            "question": ("String", "Question you may ask about {text}"),
            "answer": ("String", "Your answer to {question} according {text}"),
        }])
        .start()
    qa_pairs.append({
        "origin_piece": text_piece,
        "qa_pairs": result,
    })
    print("[Done] Start next work in " + str(sleep_time) + " seconds.")
    time.sleep(sleep_time)
print("[All Works Done]
")

# Print QA Paris
for item in qa_pairs:
    print("[Origin Text Piece]: 
", item["origin_piece"], end="
")
    for qa in item["qa_pairs"]:
        print("Question: ", qa["question"])
        print("Answer: ", qa["answer"], end="
")
    print("------")

代码一次性跑完，输出的是结构化数据，每个问答对都绑定了原始文本块。后面无论是存数据库还是做进一步分析，都非常方便。

关键问题1：文本内容怎么读？

文本来源无非两种：本地文档和在线文档。

本地文档最简单，Python自带的open()函数直接搞定。如果是PDF格式，用PyPDF2库先提取文本，再走同样流程：

import PyPDF2

document_piece = []
with open("./README.pdf", "rb") as file:
    reader = PyPDF2.PdfReader(file)
    for page in reader.pages:
        document_piece.append(page.extract_text())

在线文档则需要用requests库下载。如果你是从GitHub上取文件，注意URL格式：https://raw.githubusercontent.com/<账户名>/<项目名>/<分支名>/<文件名>。直接用这个地址请求，就能拿到raw内容。

import requests

document_link = "https://raw.githubusercontent.com/Maplemx/Agently/main/README.md"
document_content = ""
response = requests.get(document_link)
if response.status_code == 200:
    document_content = response.content.decode("utf-8")

关键问题2：长文本怎么切块？

切块是整条流水线的核心技术点。切得太碎，句子可能断裂，语义不完整；切得太大，又容易超出模型上下文窗口。

本例处理的是markdown文档，所以巧用了markdown的标题结构。具体分两步：

第一步，用 做初始拆行，然后逐行遍历。每遇到以#开头的行，就新建一个段落；否则，把当前行追加到上一个段落末尾。这样能保证每一段都是一个相对完整的语义单元。

chunks = document_content.split("

")
paragraphs = []
paragraph_num = -1
for chunk in chunks:
    if chunk.startswith("#"):
        paragraphs.append(chunk + "

")
        paragraph_num += 1
    else:
        paragraphs[paragraph_num] += chunk + "

"

第二步，把这些段落再尽量拼装起来，只要拼装后的总长度不超过piece_length_control（本例设为1000字符）就继续拼。这样既能减少对大模型的调用次数，又能保证每个文本块的完整性。

text_pieces = []
text_piece_num = 0
for paragraph in paragraphs:
    if len(text_pieces) == 0:
        text_pieces.append(paragraph)
    else:
        if len(text_pieces[text_piece_num] + paragraph) > piece_length_control:
            text_pieces.append(paragraph)
            text_piece_num += 1
        else:
            text_pieces[text_piece_num] += paragraph

这个策略的核心在于：让切分方法匹配文档的结构。如果是HTML文档，可以用

、

标签做切分点；如果是PDF论文，可以按段落自然换行处理。关键是不能机械地按固定字符数硬切。

关键问题3：怎么快捷生成结构化数据？

这部分正是Agently框架的强项——用工程化的语言，把复杂的大模型调用包装成直观的Python操作。

• 输入文本块：
  直接在.input()中传入{ "text": text_piece }，就像给函数传参一样自然。

• 定义输出结构：
  在.output()中声明想要的JSON结构。比如本例中，每个问答对就是一个dict，包含question和answer两字段。每个字段还可以通过("String", "描述")的方式，进一步告诉模型这个字段的用途和格式要求。

• 额外指令：
  在.instruct()中加入“对每一段文本至少生成5个问答对”这样的指导。你甚至可以传入一个list，里面写上“output language: Chinese”，看看效果有什么不同。

qa_pairs = []
agent = agent_factory.create_agent()
for text_piece in text_pieces:
    print("[Working on]: ", text_piece.split("
")[0])
    result = agent
        .input({"text": text_piece })
        .instruct("Generate at least 5 question and answer pairs about {text}")
        .output([{
            "question": ("String", "Question you may ask about {text}"),
            "answer": ("String", "Your answer to {question} according {text}"),
        }])
        .start()
    qa_pairs.append({
        "origin_piece": text_piece,
        "qa_pairs": result,
    })
    print("[Done] Start next work in " + str(sleep_time) + " seconds.")
    time.sleep(sleep_time)
print("[All Works Done]
")

最后还有一点细节：批量调用时注意接口频率限制。通过sleep_time变量控制请求间隔，避免被限流。整个流程跑完后，每个文本块对应的问答对和原始文本块一一映射，后续存储和检索都一目了然。

简单总结一句：这套方案的核心价值不在于代码有多复杂，而在于把“从长文本到结构化问答”这条路径，用最少的样板代码实现了闭环。剩下的，就是根据你自己的文档场景，调整切块策略和输出格式即可。