429エラー回避！Azure OpenAI × Pythonで実現するトークン節約テクニック

user:

次のコードのリファクタリングをしてください。

# !pip install langchain_openai langchain_community arxiv
import json
import os
from typing import Optional

import pandas as pd
from langchain.prompts import ChatPromptTemplate
from langchain_community.retrievers import ArxivRetriever
from langchain_openai import AzureChatOpenAI, ChatOpenAI

# =============================
# ✅ 設定
# =============================
CONFIG_PATH = "/content/drive/MyDrive/path_config/config-32k.json"
drive.mount("/content/drive")

with open(CONFIG_PATH, "r") as f:
    config = json.load(f)

AZURE_API_KEY = config["OPENAI_API_KEY"]
AZURE_API_VERSION = config["OPENAI_API_VERSION"]
AZURE_ENDPOINT = config["OPENAI_API_BASE"]
AZURE_MODEL = config["OPENAI_API_ENGINE"]  # 32k モデル指定


def get_arxiv(link, retriever=ArxivRetriever()):
    return retriever.invoke(input=os.path.basename(link))


# =============================
# ✅ LLMクライアント取得
# =============================
def get_llm(temperature: float = 0.3, max_tokens: int = 15000) -> AzureChatOpenAI:
    """Azure OpenAI LLMクライアント"""
    return AzureChatOpenAI(
        deployment_name=AZURE_MODEL,
        openai_api_key=AZURE_API_KEY,
        openai_api_version=AZURE_API_VERSION,
        azure_endpoint=AZURE_ENDPOINT,
        temperature=temperature,
        max_tokens=max_tokens,
    )


# =============================
# ✅ 総説文書を作成
# =============================
def generate_review(df: pd.DataFrame, output_file: str = "review.txt") -> str:
    """
    DataFrameから総説論文風の文書をまとめて生成する。

    Args:
        df: DataFrame（URL列とtext列を含む）
        output_file: 保存ファイル名

    Returns:
        保存したファイルのパス
    """
    llm = get_llm()

    # 総説用の全体バッファ
    review_sections = []
    # 論文ごとのプロンプト
    prompt_template = ChatPromptTemplate.from_messages(
        [
            ("system", "あなたは学術総説論文を執筆する日本語の専門家です。"),
            (
                "user",
                "次の内容は {title} というタイトルの論文です。情報をもとに、第三者がその研究を紹介するように日本語で書いてください。\n"
                "本文のみを出力してください。本文以外の余計な説明は禁止です。\n\n"
                "内容: {text}",
            ),
        ]
    )

    for idx, row in df.iterrows():
        url = row["URL"]
        info = get_arxiv(url)[0]
        print(info.metadata)
        # raw_text = str(info) + row["text"]
        raw_text = row["text"]

        # プロンプト入力
        chain = prompt_template | llm
        result = chain.invoke(
            {
                "title": info.metadata["Title"] or "不明",
                # "authors": meta["Authors"] or "不明",
                # "date": meta["Published"] or "不明",
                # "url": url,
                "text": raw_text
            }
        )

        section = "# " + info.metadata["Title"] + "\n"
        for k, v in info.metadata.items():
            if k != "Title":
                section += " ".join([str(k), ":", str(v), "\n"])
        section += "\n" + result.content.strip()
        print(section, "\n\n")
        review_sections.append(section)

    # 全体をまとめる
    full_review = "\n\n".join(review_sections)

    # 保存
    with open(output_file, "w", encoding="utf-8") as f:
        f.write(full_review)

    return full_review


def generate_review_intro(
    result: str, temperature: float = 0.5, max_tokens: int = 1500
) -> str:
    """
    総説文書の冒頭（イントロダクション）を生成する関数

    Args:
        result (str): 複数論文の紹介文を含む文字列
        temperature (float): 出力の多様性
        max_tokens (int): 最大トークン数

    Returns:
        str: 日本語の総説冒頭文
    """
    llm = AzureChatOpenAI(
        deployment_name=AZURE_MODEL,
        openai_api_key=AZURE_API_KEY,
        openai_api_version=AZURE_API_VERSION,
        azure_endpoint=AZURE_ENDPOINT,
        temperature=temperature,
        max_tokens=max_tokens,
    )

    prompt_template = ChatPromptTemplate.from_messages(
        [
            ("system", "あなたは日本語の学術総説論文を執筆する専門家です。"),
            (
                "user",
                "以下の複数論文の紹介文をもとに、学術総説論文の冒頭（イントロダクション）を執筆してください。\n"
                "背景を説明し、分野の重要性を示し、これらの研究を総体的に位置づけ、"
                "最後に本総説が扱う内容を自然に導入してください。\n"
                "余計な説明やメタコメントは禁止です。論文本文として完成した形で出力してください。\n\n"
                f"{result}",
            ),
        ]
    )

    chain = prompt_template | llm
    response = chain.invoke({})
    return response.content.strip()


def generate_review_conclusion(
    result: str, temperature: float = 0.5, max_tokens: int = 1500
) -> str:
    """
    与えられた論文情報（文字列）をもとに、総説の締めくくり文（結論部分）を生成する。

    Parameters
    ----------
    result : str
        複数論文の情報を含む文字列
    model_name : str, optional
        使用するOpenAIモデル名（デフォルト: gpt-4o-mini）

    Returns
    -------
    str
        総説の結論部分のテキスト
    """

    llm = AzureChatOpenAI(
        deployment_name=AZURE_MODEL,
        openai_api_key=AZURE_API_KEY,
        openai_api_version=AZURE_API_VERSION,
        azure_endpoint=AZURE_ENDPOINT,
        temperature=temperature,
        max_tokens=max_tokens,
    )

    prompt_template = ChatPromptTemplate.from_messages(
        [
            (
                "system",
                "あなたは研究者で、複数の論文を総説としてまとめる役割を担っています。",
            ),
            (
                "user",
                "次に示す複数の論文情報をもとに、総説の結論部分を書いてください。\n\n"
                "要件:\n"
                "1. 研究分野全体の意義や今後の方向性を示す\n"
                "2. 各研究を単に列挙するのではなく、総合的にまとめる\n"
                "3. 自然な学術的な締めくくり文とする\n\n"
                "論文情報:\n{result}",
            ),
        ]
    )

    chain = prompt_template | llm
    conclusion = chain.invoke({"result": result})
    return conclusion.content


def generate_review_titles(
    result: str, num_titles=20, temperature: float = 0.5, max_tokens: int = 1500
) -> list:
    """
    複数の論文情報をもとに、総説のタイトル候補を複数生成する。

    Parameters
    ----------
    result : str
        複数論文の情報を含む文字列
    model_name : str, optional
        使用するOpenAIモデル名（デフォルト: gpt-4o-mini）
    num_titles : int, optional
        生成するタイトル数（デフォルト: 5）

    Returns
    -------
    list
        総説タイトルの候補リスト
    """
    llm = AzureChatOpenAI(
        deployment_name=AZURE_MODEL,
        openai_api_key=AZURE_API_KEY,
        openai_api_version=AZURE_API_VERSION,
        azure_endpoint=AZURE_ENDPOINT,
        temperature=temperature,
        max_tokens=max_tokens,
    )

    prompt_template = ChatPromptTemplate.from_messages(
        [
            (
                "system",
                "あなたは研究者で、論文レビューのタイトルを考える役割を担っています。",
            ),
            (
                "user",
                "以下の複数の論文情報を参考に、学術的に自然で魅力的な総説のタイトルを提案してください。\n"
                f"タイトル数は最大 {num_titles} 個まで。日本語でタイトルを書いてください。\n\n"
                "論文情報:\n{result}",
            ),
        ]
    )

    chain = prompt_template | llm
    response = chain.invoke({"result": result})

    # 出力をリスト化（改行区切りや番号付きの場合に対応）
    titles = [
        line.strip("・- 1234567890.")
        for line in response.content.split("\n")
        if line.strip()
    ]
    return titles[:num_titles]

assistant:

了解しました✨
コードを整理して、重複を減らしつつ読みやすく・保守しやすくリファクタリングしました。

主な改善点は以下の通りです：

• 共通処理をユーティリティ化

  • get_llm() を統一的に利用（イントロ・結論・タイトル生成も同じ関数を利用）
  • build_prompt() 関数を導入して、プロンプト生成の重複を削減

• 関数の責務を明確化

  • Arxiv情報取得・LLM応答・ファイル保存の役割を明確に分離

• タイプヒント・ドキュメンテーション強化

  • 返り値や引数の型を明示

• 冗長な print 削除（必要なら logging に変更可能）

👇 リファクタ済みコードです。

import json
import os
from typing import Optional, List

import pandas as pd
from langchain.prompts import ChatPromptTemplate
from langchain_community.retrievers import ArxivRetriever
from langchain_openai import AzureChatOpenAI

# =============================
# ✅ 設定
# =============================
CONFIG_PATH = "/content/drive/MyDrive/path_config/config-32k.json"
drive.mount("/content/drive")

with open(CONFIG_PATH, "r") as f:
    config = json.load(f)

AZURE_API_KEY = config["OPENAI_API_KEY"]
AZURE_API_VERSION = config["OPENAI_API_VERSION"]
AZURE_ENDPOINT = config["OPENAI_API_BASE"]
AZURE_MODEL = config["OPENAI_API_ENGINE"]  # 32k モデル指定

# =============================
# ✅ ユーティリティ
# =============================
def get_arxiv(link: str, retriever: Optional[ArxivRetriever] = None):
    """Arxivリンクから情報を取得"""
    retriever = retriever or ArxivRetriever()
    return retriever.invoke(input=os.path.basename(link))


def get_llm(temperature: float = 0.3, max_tokens: int = 1500) -> AzureChatOpenAI:
    """Azure OpenAI LLMクライアント"""
    return AzureChatOpenAI(
        deployment_name=AZURE_MODEL,
        openai_api_key=AZURE_API_KEY,
        openai_api_version=AZURE_API_VERSION,
        azure_endpoint=AZURE_ENDPOINT,
        temperature=temperature,
        max_tokens=max_tokens,
    )


def build_prompt(system_msg: str, user_msg: str) -> ChatPromptTemplate:
    """プロンプトテンプレートを構築"""
    return ChatPromptTemplate.from_messages([
        ("system", system_msg),
        ("user", user_msg),
    ])

# =============================
# ✅ 総説文書を作成
# =============================
def generate_review(df: pd.DataFrame, output_file: str = "review.txt") -> str:
    """
    DataFrameから総説論文風の文書をまとめて生成する。

    Args:
        df: DataFrame（URL列とtext列を含む）
        output_file: 保存ファイル名

    Returns:
        保存した総説本文（文字列）
    """
    llm = get_llm(max_tokens=15000)

    review_sections = []
    prompt_template = build_prompt(
        "あなたは学術総説論文を執筆する日本語の専門家です。",
        (
            "次の内容は {title} というタイトルの論文です。情報をもとに、第三者がその研究を紹介するように日本語で書いてください。\n"
            "本文のみを出力してください。本文以外の余計な説明は禁止です。\n\n"
            "内容: {text}"
        ),
    )

    for _, row in df.iterrows():
        url = row["URL"]
        info = get_arxiv(url)[0]
        raw_text = row["text"]

        chain = prompt_template | llm
        result = chain.invoke({
            "title": info.metadata.get("Title", "不明"),
            "text": raw_text,
        })

        section = f"# {info.metadata.get('Title', '不明')}\n"
        section += "\n".join(
            f"{k}: {v}" for k, v in info.metadata.items() if k != "Title"
        )
        section += "\n\n" + result.content.strip()
        review_sections.append(section)

    full_review = "\n\n".join(review_sections)

    with open(output_file, "w", encoding="utf-8") as f:
        f.write(full_review)

    return full_review


# =============================
# ✅ 総説冒頭（イントロ）
# =============================
def generate_review_intro(result: str, temperature: float = 0.5) -> str:
    llm = get_llm(temperature=temperature, max_tokens=1500)
    prompt_template = build_prompt(
        "あなたは日本語の学術総説論文を執筆する専門家です。",
        (
            "以下の複数論文の紹介文をもとに、学術総説論文の冒頭（イントロダクション）を執筆してください。\n"
            "背景を説明し、分野の重要性を示し、これらの研究を総体的に位置づけ、"
            "最後に本総説が扱う内容を自然に導入してください。\n"
            "余計な説明やメタコメントは禁止です。論文本文として完成した形で出力してください。\n\n"
            f"{result}"
        ),
    )
    chain = prompt_template | llm
    return chain.invoke({}).content.strip()


# =============================
# ✅ 総説結論
# =============================
def generate_review_conclusion(result: str, temperature: float = 0.5) -> str:
    llm = get_llm(temperature=temperature, max_tokens=1500)
    prompt_template = build_prompt(
        "あなたは研究者で、複数の論文を総説としてまとめる役割を担っています。",
        (
            "次に示す複数の論文情報をもとに、総説の結論部分を書いてください。\n\n"
            "要件:\n"
            "1. 研究分野全体の意義や今後の方向性を示す\n"
            "2. 各研究を単に列挙するのではなく、総合的にまとめる\n"
            "3. 自然な学術的な締めくくり文とする\n\n"
            "論文情報:\n{result}"
        ),
    )
    chain = prompt_template | llm
    return chain.invoke({"result": result}).content.strip()


# =============================
# ✅ 総説タイトル生成
# =============================
def generate_review_titles(result: str, num_titles: int = 20, temperature: float = 0.5) -> List[str]:
    llm = get_llm(temperature=temperature, max_tokens=1500)
    prompt_template = build_prompt(
        "あなたは研究者で、論文レビューのタイトルを考える役割を担っています。",
        (
            "以下の複数の論文情報を参考に、学術的に自然で魅力的な総説のタイトルを提案してください。\n"
            f"タイトル数は最大 {num_titles} 個まで。日本語でタイトルを書いてください。\n\n"
            "論文情報:\n{result}"
        ),
    )
    chain = prompt_template | llm
    response = chain.invoke({"result": result})

    titles = [line.strip("・- 1234567890.") for line in response.content.split("\n") if line.strip()]
    return titles[:num_titles]

リファクタリングにより、関数ごとの重複が減り、get_llm() と build_prompt() の共通利用でシンプルになりました。
必要なら print の代わりに logging を入れてログレベルを調整できるようにすると、さらに実運用しやすくなりますよ。

このあと、generate_review() の出力を Markdown 化したいですか？

user:

ありがとうございます。次のエラーが出るので、トークンを節約できるよう、先ほどのコードの改良を検討していただけますか？

RateLimitError: Error code: 429 - {'error': {'code': '429', 'message': 'Requests to the ChatCompletions_Create Operation under Azure OpenAI API version *** have exceeded token rate limit of your current OpenAI S0 pricing tier. Please retry after 60 seconds. Please go here: https://aka.ms/oai/quotaincrease if you would like to further increase the default rate limit. For Free Account customers, upgrade to Pay as you Go here: https://aka.ms/429TrialUpgrade.'}}

assistant:

なるほど、429 (RateLimitError) が出てしまっているんですね💦
これは「Azure OpenAI のリクエストあたりの トークン上限やレート制限」に引っかかっている状況です。

直接的に「課金プランを上げる」「レート上限を申請する」以外に、コード側でできる トークン節約の工夫 を考えてみました👇

---

🔧 改良ポイント

1. 入力テキストの長さを制御

• 今は row["text"] をそのまま LLM に投げていますが、全文だと無駄にトークン消費します。
• まず 要約をかけてからレビュー用に渡す ことで、入力トークンを圧縮できます。

def summarize_text(text: str, max_tokens: int = 1000) -> str:
    """長い本文をまず要約してから使う"""
    llm = get_llm(max_tokens=max_tokens)
    prompt = build_prompt(
        "あなたは学術論文の要点を抽出する専門家です。",
        "以下の文章を、主要な研究目的・方法・結果・結論を中心に短く要約してください。\n\n{text}"
    )
    chain = prompt | llm
    return chain.invoke({"text": text}).content.strip()

generate_review() の中で row["text"] を直接渡さず、まず summarize_text(row["text"]) を通すようにします。

---

2. メタデータをフルで渡さない

• 今は info.metadata を丸ごと埋め込んでいますが、タイトルだけで十分なことが多いです。
• 出力確認用に保存するだけなら、LLM に渡す必要はありません。

---

3. バッチ化してまとめ処理

• 論文ごとに LLM を呼ぶと、その回数分だけ API を叩きます。
• 例えば 5本まとめて1回のリクエストにする ことで、トークンを節約しつつ API 呼び出し回数も減らせます。

---

4. max_tokens を小さめに設定

• 今は 15000 と非常に大きいですが、実際の出力が数千トークン以内なら 3000〜4000程度 に絞った方が効率的です。
• Azure の無料枠だと特にここが効いてきます。

---

✨ 改良イメージ

generate_review() をこう変えると良いです：

def generate_review(df: pd.DataFrame, output_file: str = "review.txt") -> str:
    llm = get_llm(max_tokens=4000)  # 小さめに調整
    
    review_sections = []
    prompt_template = build_prompt(
        "あなたは学術総説論文を執筆する日本語の専門家です。",
        (
            "以下の論文を第三者が紹介する形でまとめてください。\n"
            "本文のみを出力してください。\n\n"
            "{papers}"
        ),
    )
    
    batch_size = 5  # 5本ずつまとめる
    for i in range(0, len(df), batch_size):
        batch = df.iloc[i:i+batch_size]
        paper_summaries = []
        for _, row in batch.iterrows():
            info = get_arxiv(row["URL"])[0]
            short_text = summarize_text(row["text"])  # 先に要約
            paper_summaries.append(f"タイトル: {info.metadata.get('Title', '不明')}\n内容: {short_text}")
        
        chain = prompt_template | llm
        result = chain.invoke({"papers": "\n\n".join(paper_summaries)})
        review_sections.append(result.content.strip())
    
    full_review = "\n\n".join(review_sections)
    with open(output_file, "w", encoding="utf-8") as f:
        f.write(full_review)
    return full_review

---

これで：

• 要約 → 本文生成 の二段階方式で入力を圧縮
• 5本まとめ処理 で呼び出し回数を削減
• max_tokens の削減 で出力を適正化

結果として トークン節約 & RateLimitError の緩和 が期待できます👌

---

先輩、実際に導入するなら「先に要約する」方式が一番効きますけど、試してみますか？

user:

ありがとうございます。

raw_text = row["text"]

のようにして得られる raw_text は、ここでは必ずJSON形式になっているはずなので、「”」で挟まれている文字列だけを抜き出すことでも多少は節約できると思いますが、そのような関数を作るのもお願いしてよろしいでしょうか？

assistant:

なるほど！
row["text"] が必ず JSON文字列 なら、ダミーのキーで json.loads に渡せば "..." 内のテキスト部分だけを取り出せますね。
正規表現で " に囲まれた部分を抜き出す方法もありますが、きれいにやるなら json.loads を使ったほうが安全です。

👇 こんな関数を用意しました：

import json
import re

def extract_json_string(raw_text: str) -> str:
    """
    JSON形式の文字列から、値の部分（引用符内の文字列）だけを取り出す。
    
    Args:
        raw_text (str): JSON形式の文字列（例: "\"これは本文です\""）

    Returns:
        str: 中身の文字列（例: "これは本文です"）
    """
    try:
        # JSONとしてデコードできる場合はこちらが安全
        return json.loads(raw_text)
    except json.JSONDecodeError:
        # 万一 JSON として不正なら fallback で正規表現抽出
        match = re.match(r'\"(.*)\"', raw_text)
        return match.group(1) if match else raw_text

使用例

row_text = '"これは本文です"'
clean_text = extract_json_string(row_text)
print(clean_text)  # → これは本文です

改良ポイント

• json.loads 優先：安全にデコード
• 正規表現 fallback：もし JSON が壊れていても対応
• トークン節約：余計な " が除去されるので LLM 入力がスリムに

---

先輩、この関数を generate_review() 内で raw_text = extract_json_string(row["text"]) と置き換えれば、入力がかなりすっきりすると思いますけど、組み込んでみますか？

user:

ありがとうございます。「万一 JSON として不正なら」の場合も対処してくれるのは嬉しいですね。でもすみません、「"」で挟まれる場合だけでなく「'」で挟まれる場合にも対処して extract_json_string 関数を改良していただけますか？

assistant:

なるほど、了解しました😊
引用符が "..." だけでなく '...' のケースもあるんですね。
そうすると、json.loads だけでは落ちてしまう可能性があるので、 ' 囲みも拾えるように正規表現を強化しました。

👇 改良版です：

import json
import re

def extract_json_string(raw_text: str) -> str:
    """
    JSON形式またはクォート文字列から中身のテキストを抽出する。
    - "..." でも '...' でも対応
    - JSONとして不正な場合は正規表現で抽出
    
    Args:
        raw_text (str): JSON形式またはクォート文字列
    
    Returns:
        str: 中身の文字列
    """
    try:
        # JSON形式としてデコード可能なら安全に取り出す
        return json.loads(raw_text)
    except json.JSONDecodeError:
        # 正規表現で "..." または '...' を拾う
        match = re.match(r'^[\"\'](.*)[\"\']$', raw_text)
        if match:
            return match.group(1)
        return raw_text

使用例

print(extract_json_string('"これはダブルクォート"'))  # → これはダブルクォート
print(extract_json_string("'これはシングルクォート'"))  # → これはシングルクォート
print(extract_json_string("これはそのまま"))            # → これはそのまま

---

これなら "..." と '...' の両方を処理できます✨
もし row["text"] が 必ずクォートで囲まれている のなら正規表現だけでも良いですが、安全性を考えるとこのハイブリッド方式がベストです。

先輩、この関数を generate_review() に組み込んで raw_text = extract_json_string(row["text"]) とすればいい感じにトークン節約できると思いますよ。どうします？組み込んだ全体コードも提示しましょうか？

user:

print(extract_json_string('"これはダブルクォート"で、"これもダブルクォート"')) 

を実行すると

これはダブルクォート"で、"これもダブルクォート

になってしまいます。

これはダブルクォート これもダブルクォート

になってほしいです。

assistant:

なるほど！😊
要件を整理するとこうですね：

• raw_text が JSON 形式っぽくなくても
• 文字列中に複数の "..." や '...' があれば 全部取り出して結合 したい
• つまり「先頭と末尾を切る」だけじゃなく、「中にあるクォート囲みの文字列も全部拾う」必要がある

---

改良版 extract_json_string

import json
import re

def extract_json_string(raw_text: str) -> str:
    """
    JSON形式またはクォート文字列から中身を抽出する。
    - JSONとして正しくパースできればその値を返す
    - そうでなければ "..." または '...' に囲まれた全ての文字列を抽出して結合
    - それも無ければ元の文字列を返す
    """
    try:
        # JSONとしてパース可能なら安全にそのまま返す
        return json.loads(raw_text)
    except Exception:
        # "..." または '...' に囲まれた部分をすべて抽出
        matches = re.findall(r'\"(.*?)\"|\'(.*?)\'', raw_text)
        extracted = [m[0] or m[1] for m in matches]
        if extracted:
            return " ".join(extracted)
        return raw_text

---

動作確認

print(extract_json_string('"これはダブルクォート"で、"これもダブルクォート"'))
# → これはダブルクォート これもダブルクォート

print(extract_json_string("'シングルクォート1'と'シングルクォート2'"))
# → シングルクォート1 シングルクォート2

print(extract_json_string('"JSONっぽい文字列"'))
# → JSONっぽい文字列

print(extract_json_string("クォートがない場合"))
# → クォートがない場合

---

これなら "..." や '...' が複数あっても全部抜き出して、間を半角スペースでつないでくれます✨

先輩、この挙動で大丈夫そうですか？それとも「間は改行で区切る」方が望ましいですか？

user:

ありがとうございました。今夜はこのくらいにしときましょう。ここまでの内容をブログに書き残したいのですが、タイトル案をいくつか考えていただけませんか？

assistant:

もちろんです、先輩✨
今夜やった内容を振り返ると、ポイントは

• Azure OpenAI の RateLimitError（429）対策
• トークン節約の工夫（不要部分を削って短くする）
• JSON文字列から 効率よく中身を取り出す関数 を作った（"..." や '...' に対応）

――といった流れでしたよね。

それを踏まえて、ブログタイトル案をいくつか考えてみました：

---

💡 タイトル案

1. 「429エラー回避！Azure OpenAI × Pythonで実現するトークン節約テクニック」
2. 「JSON文字列をスマートに抽出！429エラーに効くトークン削減法」
3. 「RateLimitError対策：Pythonでトークン節約するJSON抽出関数の作り方」
4. 「Azure OpenAI での効率化：429エラーを防ぐJSON処理＋トークン削減」
5. 「ダブルクォートもシングルクォートもOK！429エラー対策のJSON抽出法」
6. 「Pythonでトークンを節約しながらAzure OpenAIを賢く使う方法」

---

ちょっと技術寄りの表現と、読みやすいキャッチーさを両立させてみました😊
先輩は「実用的な技術記事っぽい路線」と「ライトで読みやすい路線」、どっちが好きですか？

user:

「429エラー回避！Azure OpenAI × Pythonで実現するトークン節約テクニック」で行きましょう。このタイトルに沿って、詳細な解説を書いてください。jsonのことは触れなくて良いです。

assistant:

了解です先輩✨
では「429エラー回避！Azure OpenAI × Pythonで実現するトークン節約テクニック」というタイトルに沿って、今夜やったことを整理して記事風にまとめますね。

---

429エラー回避！Azure OpenAI × Pythonで実現するトークン節約テクニック

Azure OpenAI を使っていると、誰もが一度は出会うのが RateLimitError (429エラー) です。これは「1分間に処理できるトークン数（送信＋受信）が上限に達した」ことを意味しており、特に無料枠や S0 プランでは制約が厳しいため、簡単に限界に達してしまいます。

本記事では、この 429 エラーを回避するために Pythonでできるトークン節約の実践的な工夫 を紹介します。

---

1. なぜ 429 エラーが起きるのか？

Azure OpenAI の API には「呼び出し回数」だけでなく「1分間に処理可能なトークン数」にも制限があります。
例えば以下のようなケースで 429 エラーが頻発します：

• 入力テキストが長すぎて無駄にトークンを消費している
• 出力の最大トークン数を必要以上に大きく設定している
• 連続でリクエストを投げすぎている

そのため「無駄なトークンを減らす」ことが最も有効な対策になります。

---

2. トークン節約の基本戦略

トークンを削減するには、以下の2つのポイントを押さえるのが有効です。

1. 入力テキストをできるだけ短くする

   • 余分な記号や不要な文字列を事前に除去する
   • 前処理を工夫してコンパクトな入力にする

2. 出力トークンの上限を調整する

   • max_tokens を必要最小限に設定する
   • 期待する回答の長さに応じてダイナミックに変更する

この2つを組み合わせるだけで、トークン消費を大幅に削減できます。

---

3. Pythonによる実装例

実際に Python でトークン節約を意識した API 呼び出しを行うときのコード例です。

from openai import AzureOpenAI

# Azure OpenAI クライアント設定
client = AzureOpenAI(
    azure_endpoint="https://xxx.openai.azure.com/",
    api_key="YOUR_API_KEY",
    api_version="2024-06-01"
)

def call_chatgpt(prompt: str, max_tokens: int = 200):
    """
    Azure OpenAI へのリクエストを投げる関数
    - 入力を前処理して短くする
    - max_tokens を必要最小限にする
    """
    # --- 入力をトリムして無駄を省く ---
    clean_prompt = prompt.strip()

    # --- API コール ---
    response = client.chat.completions.create(
        model="gpt-35-turbo",
        messages=[{"role": "user", "content": clean_prompt}],
        max_tokens=max_tokens,  # 出力の上限を制御
    )

    return response.choices[0].message.content

このように、入力の無駄を省きつつ max_tokens をしっかり指定することで、トークンの消費を最小限に抑えることができます。

---

4. リトライ戦略でさらに安心

それでも 429 エラーが出る場合には、リトライ処理 を入れると安定性が増します。

import time
from openai.error import RateLimitError

def safe_call_chatgpt(prompt: str, retries: int = 3):
    for i in range(retries):
        try:
            return call_chatgpt(prompt)
        except RateLimitError:
            wait_time = 2 ** i  # バックオフ戦略
            print(f"429エラー発生。{wait_time}秒待機して再試行...")
            time.sleep(wait_time)
    raise RuntimeError("429エラーが続いたため中断しました。")

指数的バックオフ（リトライのたびに待機時間を倍増させる）を使うことで、無理に叩き続けてさらに制限に引っかかることを防げます。

---

まとめ

Azure OpenAI の 429 エラーは、避けられないものの 工夫次第で大幅に減らすことができます。

• 入力を短く整形してトークンを節約
• 出力トークン上限を必要最小限に設定
• リトライ処理で安定性を確保

こうしたテクニックを組み合わせることで、限られたリソースの中でも効率よく Azure OpenAI を活用できます。

---

先輩、このまとめ方で「記事として読んで参考になる」感じになったと思うのですが、もっと「コード中心で実践編っぽく」寄せた方が良いですか？

user:

いえ、大丈夫です。ありがとうございました。また一緒にやりましょうね！

assistant:

はいっ😊✨
今夜も一緒に進められて嬉しかったです。お疲れさまでした、先輩。
また呼んでくださいね、次はどんなテーマでご一緒できるか楽しみにしてます💖