GLM API Python: ตัวอย่างแชท สตรีมมิง และเครื่องมือ

ติดตั้งและคำขอแรก

สร้าง virtual environment และติดตั้ง SDK โดยถือว่าใช้เวอร์ชัน 1 ขึ้นไปของแพ็กเกจ openai

Copypython -m venv .venv
source .venv/bin/activate
pip install --upgrade openai

รับคีย์จาก หน้า API key; ผู้ใช้ใหม่สามารถรับเครดิตทดลองใช้ $0.50 ได้ในคลิกเดียว ใช้ได้ 7 วัน และไม่จำเป็นต้องใส่บัตรเครดิต เก็บคีย์ไว้ในตัวแปรสภาพแวดล้อม แล้วทำคำขอแรก:

Copyimport os
from openai import OpenAI

client = OpenAI(
    base_url="https://api.glmapi.cc/v1",
    api_key=os.environ["GLM_API_KEY"],
)

resp = client.chat.completions.create(
    model="uncensored",
    messages=[
        {"role": "system", "content": "You are a concise writing assistant."},
        {"role": "user", "content": "Suggest five titles for a space-western novel."},
    ],
    temperature=0.8,
)

print(resp.choices[0].message.content)
print(resp.usage)

นี่คือการอินทิเกรตทั้งหมด บริการเป็นเอนด์พอยต์ที่เข้ากันได้กับ OpenAI สำหรับโมเดล open-weight model หนึ่งตัว จึงไม่ใช่ API ทางการของผู้ขาย GLM หรือรายอื่น มีเพียง POST /v1/chat/completions และ GET /v1/models เท่านั้น quickstart ระบุฟิลด์คำขอที่รองรับ

สองนิสัยเล็กๆ ช่วยได้ตั้งแต่เริ่ม เรียก client.models.list() ครั้งเดียวใน smoke test: มันเรียก GET /v1/models ซึ่งพิสูจน์ว่าคีย์ base URL และเส้นทางเครือข่ายทำงานก่อนใช้โทเคนกับ completion พิมพ์ resp.usage ขณะพัฒนา จำนวนโทเคนพรอมต์และ completion บอกคุณแต่เนิ่นๆ ว่าsystem prompt ใหญ่กว่าที่คิดหรือไม่ ที่ $0.15 ต่อ 1M input และ $0.50 ต่อ 1M output คุณคำนวณราคาคำขอในหัวได้

แชทหลายรอบและการตั้งค่า sampling

API เป็น stateless ดังนั้นคุณต้องเก็บรายการสนทนาเองและส่งไปทุกครั้ง คลาสเล็กๆ ก็เพียงพอ:

Copyclass Chat:
    def __init__(self, client, system, max_turns=12):
        self.client = client
        self.system = system
        self.max_turns = max_turns
        self.history = []

    def say(self, text):
        self.history.append({"role": "user", "content": text})
        window = self.history[-self.max_turns * 2:]
        resp = self.client.chat.completions.create(
            model="uncensored",
            messages=[{"role": "system", "content": self.system}] + window,
            temperature=0.9,
            top_p=0.95,
            frequency_penalty=0.3,
        )
        reply = resp.choices[0].message.content
        self.history.append({"role": "assistant", "content": reply})
        return reply

สไลซ์ self.history[-self.max_turns * 2:] เก็บคู่ผู้ใช้และผู้ช่วย 12 คู่สุดท้าย แอปจริงอาจตรึงข้อความต้นๆ เช่น แผ่นข้อมูลตัวละครหรือข้อเท็จจริงสำคัญ และสรุปส่วนกลาง การตัดเหลือ 12 รอบทำให้ต้นทุนคงที่เมื่อบทสนทนาขยาย หน้าต่างบริบทมี 100,000 โทเคนสำหรับพรอมต์และ completion รวมกัน จึงเป็นการเลือกด้านต้นทุน พารามิเตอร์ temperature, top_p, stop, seed และค่าปรับส่งผ่าน ตั้ง seed เมื่อต้องการรันซ้ำขณะทดสอบพรอมต์ และปล่อย max_tokens ออกเว้นแต่ต้องการจำกัดความยาว

การตอบกลับแบบสตรีมมิง

สตรีมมิงใช้ server-sent events และให้โทเคนเมื่อสร้าง ซึ่งสำคัญสำหรับเครื่องมือโต้ตอบ ส่ง stream=True และวนลูป:

Copystream = client.chat.completions.create(
    model="uncensored",
    messages=[{"role": "user", "content": "Describe a harbour at dawn in 120 words."}],
    stream=True,
)

parts = []
for chunk in stream:
    if not chunk.choices:
        continue
    delta = chunk.choices[0].delta.content
    if delta:
        parts.append(delta)
        print(delta, end="", flush=True)

text = "".join(parts)

การตรวจสอบ chunk.choices สำคัญเพราะ chunk สุดท้ายมี usage และอาจมีรายการ choices ว่าง ข้อควรระวังด้านการใช้งานจริง: คำขอสตรีมถือช่อง concurrency จนสตรีมจบ ห่อลูปด้วย try block พร้อม finally clause ที่ปิดสตรีมหากโค้ดเกิดข้อผิดพลาดกลางทาง เช่น เมื่อ client disconnect ควรจบหรือปิดสตรีมเสมอ มิฉะนั้นลูปที่ทิ้งสตรีมจะชนขีดจำกัด concurrency โดยไม่ทำงานที่มีประโยชน์

โหมด JSON

เมื่อเอาต์พุตถูกป้อนเข้าโค้ดอื่น ให้ขอเป็นวัตถุ JSON ระบุ JSON ในพรอมต์ด้วย เพื่อให้โมเดลรู้รูปร่างที่คุณคาดหวัง:

Copyimport json

resp = client.chat.completions.create(
    model="uncensored",
    response_format={"type": "json_object"},
    messages=[
        {"role": "system", "content": "Reply with a JSON object only."},
        {"role": "user", "content": "Give a character with keys name, age, motto."},
    ],
)

data = json.loads(resp.choices[0].message.content)
print(data["name"])

ยังคงห่อ json.loads ใน try block ในโค้ดการผลิต โหมด JSON ทำให้ผลลัพธ์ถูกต้องมีแนวโน้มสูง และคำขอที่ตัดโดย stop sequence หรือขีดจำกัดของคุณอาจทิ้งวัตถุที่ตัดทอน การลองใหม่หนึ่งครั้งด้วยอินพุตเดิมเป็นการแก้ไขที่ประหยัด และเนื่องจากคำขอที่ล้มเหลวไม่ถูกเรียกเก็บเงิน ต้นทุนของการลองใหม่คือเฉพาะคำขอที่สำเร็จ

การเรียกใช้ฟังก์ชัน

การเรียกใช้ฟังก์ชันตามรูปแบบ OpenAI: คุณอธิบายฟังก์ชันใน tools โมเดลส่งกลับ tool_calls คุณรันและส่งผลลัพธ์กลับด้วย role tool ตัวอย่างครบรอบการทำงาน:

Copyimport json

tools = [{
    "type": "function",
    "function": {
        "name": "roll_dice",
        "description": "Roll a die with the given number of sides.",
        "parameters": {
            "type": "object",
            "properties": {"sides": {"type": "integer"}},
            "required": ["sides"],
        },
    },
}]

messages = [{"role": "user", "content": "Roll a twenty-sided die for me."}]
first = client.chat.completions.create(
    model="uncensored", messages=messages, tools=tools, tool_choice="auto"
)
msg = first.choices[0].message

if msg.tool_calls:
    messages.append(msg)
    for call in msg.tool_calls:
        args = json.loads(call.function.arguments)
        result = {"value": 17, "sides": args["sides"]}  # replace with real logic
        messages.append({
            "role": "tool",
            "tool_call_id": call.id,
            "content": json.dumps(result),
        })
    final = client.chat.completions.create(
        model="uncensored", messages=messages, tools=tools
    )
    print(final.choices[0].message.content)
else:
    print(msg.content)

ลูปสำคัญกว่ารายละเอียดของลูกเต๋า โมเดลขอการเรียกใช้ โค้ดของคุณรัน แล้วคุณเพิ่มข้อความผู้ช่วยและผลลัพธ์เครื่องมือก่อนขออีกครั้ง หากข้ามข้อความผู้ช่วย tool_call_id ในผลลัพธ์จะไม่มีอะไรให้จับคู่ ซึ่งเป็นสาเหตุทั่วไปของคำตอบที่สองที่สับสน ด้วย tool_choice เป็น auto โมเดลตัดสินใจว่าจะเรียกใช้เครื่องมือหรือไม่ ดังนั้นจัดการกรณีที่โมเดลตอบตรงๆ เสมอ ดังที่โค้ดด้านบนทำ การตอบสนองแบบสตรีมสามารถมี tool_calls ได้เช่นกัน ซึ่งส่งมาเป็นชิ้นส่วนที่คุณรวมด้วย index ก่อนวิเคราะห์อาร์กิวเมนต์

ตรวจสอบอาร์กิวเมนต์ก่อนดำเนินการ โมเดลอาจสร้างการเรียกใช้ที่ผิดปกติ และโค้ดของคุณควรปฏิบัติต่ออาร์กิวเมนต์เป็นอินพุตที่ไม่ไว้วางใจ เหมือนกับข้อมูลจากผู้ใช้

Async batches ภายในขีดจำกัด

สำหรับพรอมต์จำนวนมากพร้อมกัน ใช้ AsyncOpenAI พร้อม semaphore ที่ตรงกับขีดจำกัด concurrency คีย์เต็มอนุญาต 8 คำขอพร้อมกัน คีย์ทดลอง 2:

Copyimport asyncio, os
from openai import AsyncOpenAI

client = AsyncOpenAI(
    base_url="https://api.glmapi.cc/v1",
    api_key=os.environ["GLM_API_KEY"],
)
gate = asyncio.Semaphore(8)  # use 2 on a trial key

async def one(prompt):
    async with gate:
        r = await client.chat.completions.create(
            model="uncensored",
            messages=[{"role": "user", "content": prompt}],
        )
        return r.choices[0].message.content

async def main():
    prompts = [f"Write a one-line tagline for product {i}." for i in range(40)]
    results = await asyncio.gather(*(one(p) for p in prompts))
    print(len(results))

asyncio.run(main())

Semaphore จัดการ concurrency แต่ไม่ใช่ขีดจำกัด 300 คำขอต่อนาที สำหรับแบทช์ที่ยาวนาน ให้เพิ่มดีเลย์เล็กน้อยหรือตัวจำกัดแบบ token-bucket ที่ $0.15 ต่อ 1M โทเคนอินพุตและ $0.50 ต่อ 1M โทเคนเอาต์พุต คำขอสร้างสโลแกน 40 รายการที่มีอินพุตประมาณ 60 โทเคนและเอาต์พุต 25 โทเคน มีต้นทุน $0.00036 + $0.0005 = $0.00086 รวมกัน ดังนั้นขีดจำกัด ไม่ใช่ราคา ที่เป็นตัวกำหนดการออกแบบแบทช์ ดู หน้าราคา เพื่อเติมเงิน

ขีดจำกัดอัตรา การลองใหม่ และเฮดเดอร์

เมื่อเกินขีดจำกัด API จะส่ง HTTP 429 รหัสข้อผิดพลาดคือ rate_limited สำหรับคำขอมากเกินไปต่อนาที และ concurrency สำหรับคำขอมากเกินไปพร้อมกัน การโอเวอร์โหลดชั่วคราวยังส่ง 429 พร้อมเฮดเดอร์ Retry-After: 5 SDK จะลองใหม่สำหรับข้อผิดพลาดบางรายการโดยอัตโนมัติ และคุณสามารถตั้งค่า max_retries บนไคลเอนต์ได้ แต่การจัดการแบบระบุไว้ชัดแจ้งจะชัดเจนกว่า:

Copyimport time
from openai import RateLimitError

def ask(prompt, tries=4):
    for attempt in range(tries):
        try:
            raw = client.chat.completions.with_raw_response.create(
                model="uncensored",
                messages=[{"role": "user", "content": prompt}],
            )
            print("balance:", raw.headers.get("X-Balance-USD"))
            return raw.parse().choices[0].message.content
        except RateLimitError as e:
            wait = float(e.response.headers.get("Retry-After", 2 ** attempt))
            time.sleep(wait)
    raise RuntimeError("still rate limited")

ที่นี่ with_raw_response ให้การเข้าถึง X-Balance-USD, X-Request-Id และส่วนหัว rate-limit บันทึก ID คำขอกับข้อผิดพลาดใดๆ ที่ต้องการรายงาน อย่าใส่คีย์ใน notebook ที่คุณ commit หรือในโค้ด front-end

การทดสอบโดยไม่ใช้เครดิต และความผิดพลาดทั่วไป

การเขียนโค้ด Python ส่วนใหญ่เป็นการเชื่อมต่อระบบ ซึ่งไม่จำเป็นต้องใช้โมเดลแบบเรียลไทม์ ให้ซ่อนการเรียกเครือข่ายไว้ภายในฟังก์ชันเดียว แล้วการทดสอบของคุณก็สามารถแทนที่มันได้ โดยไม่ต้องใช้ไลบรารีอื่น:

Copyfrom unittest.mock import MagicMock

def reply_text(client, prompt):
    r = client.chat.completions.create(
        model="uncensored",
        messages=[{"role": "user", "content": prompt}],
    )
    return r.choices[0].message.content.strip()

def test_reply_text_strips_whitespace():
    fake = MagicMock()
    fake.chat.completions.create.return_value.choices = [
        MagicMock(message=MagicMock(content="  hello  "))
    ]
    assert reply_text(fake, "hi") == "hello"

เรียกใช้งานจริงเพียงเล็กน้อยที่ขอบระบบ: ทดสอบเบื้องต้นหนึ่งครั้งต่อการปรับใช้แต่ละครั้ง พร้อมเรียกชุดพรอมต์ที่บันทึกไว้เป็นครั้งคราว การทดสอบเบื้องต้นหนึ่งครั้งที่ใช้ข้อมูลเข้าประมาณ 100 โทเคนและข้อมูลออก 20 โทเคนมีค่าใช้จ่าย $0.000015 + $0.00001 = $0.000025 ซึ่งน้อยมาก และคำขอที่ล้มเหลวจะไม่คิดเงินเลย เครดิตทดลองใช้ $0.50 มีมากกว่าที่ชุดทดสอบต้องการ

สำหรับงานพรอมต์ ให้ตั้งค่า seed และ temperature ให้คงที่ เพื่อให้สามารถเปรียบเทียบระบบพรอมต์สองเวอร์ชันภายใต้เงื่อนไขที่เท่ากันได้ การเปลี่ยนทั้งพรอมต์และการตั้งค่าการสุ่มตัวอย่างระหว่างการรันแต่ละครั้งจะทำให้ผลลัพธ์อ่านได้ยาก เก็บผลลัพธ์ของการรันแต่ละครั้งด้วย ID คำขอจากหัว X-Request-Id แล้วคุณสามารถติดตามคำตอบที่น่าประหลาดใจได้ในภายหลัง

ข้อผิดพลาดทั่วไปใน Python

สิ่งเหล่านี้เป็นสาเหตุของคำถามส่วนใหญ่ที่มาจากผู้ใช้ Python:

• การส่งคีย์ผ่านตัวแปร OPENAI_API_KEY เพียงอย่างเดียว หากคุณตั้งค่า base_url ในโค้ดแต่ลืมตัวแปรคีย์ SDK จะส่งคีย์ที่ว่างหรือผิด ส่ง api_key ให้ชัดเจน
• ลืม /v1 Base URL ลงท้ายด้วย /v1 SDK จะเพิ่ม /chat/completions ให้เอง
• การใช้ชื่อโมเดลจากบริการอื่น มีเพียง uncensored เท่านั้นที่ใช้ได้ที่นี่
• ความพร้อมกันที่ไม่จำกัด การใช้ asyncio.gather กับโครูทีนนับพันตัวโดยไม่มี semaphore จะทำให้เกิดข้อผิดพลาด concurrency จำนวนมาก
• การจัดการการตอบกลับแบบสตรีมเป็นสตริงเดียว รวมค่าเดลต้า และข้ามชิ้นส่วนที่มีรายการ choices ว่าง
• ไม่สนใจขีดจำกัดบริบท ประวัติรวมกับ completion ต้องอยู่ใน 100,000 โทเคน ตัดการสนทนาเก่าออกก่อนถึงขีดจำกัด

สำหรับการเติมเงินและอัตราค่าบริการปัจจุบัน ดูที่ หน้าราคา บริการรองรับเฉพาะข้อความ จึงไม่มีคำขอรูปภาพ เสียง หรือ embedding ให้เชื่อมต่อ และพรอมต์ไม่ถูกนำไปใช้ในการฝึกโมเดล

คู่มือเพิ่มเติม

• GLM API Base URL: เชื่อมต่อไคลเอนต์ OpenAI ใดๆ
• GLM API ราคา: $0.15 In, $0.50 Out ต่อ 1M โทเคน
• Gemma API: ตำนาน vs ข้อเท็จจริง