จำกัด HTTP Response สำหรับ AI Agent: นับ bytes ก่อน parse JSON

คำตอบสั้น

จำกัด HTTP response body ด้วยงบไบต์สะสมและเพดานจำนวนครั้งที่อ่าน ก่อน decode หรือ parse JSON ตรวจทุก chunk ด้วยเงื่อนไข len(chunk) <= max_bytes - accepted_bytes ก่อน append ถ้าจะเกินให้หยุดทันทีและทิ้ง buffer ทั้งหมด ผล COLLECTED หมายถึงอ่านถึง EOF ภายในงบเท่านั้น ไม่ได้ยืนยันว่า JSON ถูกต้อง ปลอดภัย หรือ agent ทำงานสำเร็จ

ผลลัพธ์การเรียนรู้

เมื่อจบบทนี้ คุณจะออกแบบ collector ที่จำกัด bytes สะสมและ pull attempts, แยก EOF ออกจากการเต็มเพดาน, จำแนกผลล้มเหลวโดยไม่คืน partial body และอธิบายสิ่งที่ adapter HTTP จริงยังต้องรับผิดชอบ เช่น timeout, cancellation และการปิด response

ทักษะนี้เหมาะกับผู้อ่านที่ใช้ LLM chat แล้วเริ่มสร้าง agent ด้วย tools ผู้อ่านจะได้เห็นการเปลี่ยนจากคำขอภาษาธรรมชาติให้ “อ่านข้อมูลนี้” ไปเป็น acquisition tool ที่มีงบ input, output contract, acceptance criteria และ stopping condition ระบุได้ชัด

ขอบเขต: อ่านข้อมูลไม่เท่ากับเชื่อข้อมูล

Agent อาจรับคำสั่งจากโมเดลแล้วเรียก tool ไปอ่านเว็บหรือ API ข้อมูลตอบกลับอาจใหญ่ผิดคาด แม้ต้นทางมีชื่อคุ้นเคยก็อาจส่ง payload ผิดพลาด ถูกเปลี่ยนแปลง หรือมีขนาดกินทรัพยากรมาก การ parse ไม่ใช่การตรวจความปลอดภัย: JSON ที่ถูกไวยากรณ์ยังมีรายการจำนวนมหาศาลได้ และข้อความภายในอาจพยายามชักจูงการตัดสินใจในขั้นถัดไป

แบ่งเส้นทางเป็นชั้นที่มีหน้าที่คนละอย่าง: transport รับข้อมูล, collector จำกัด bytes ที่สะสม, decoder แปลง encoding, parser สร้างโครงสร้าง, validator ตรวจ schema และ policy แล้วจึงส่งข้อมูลให้ agent ใช้ แต่ละชั้นตอบคำถามคนละข้อ การผ่าน collector บอกเพียงว่ารับข้อมูลครบภายในงบ ไม่ได้บอกว่าเนื้อหาถูกต้องหรือควรเชื่อ

OWASP อธิบายความเสี่ยง unrestricted resource consumption ว่าการไม่มีเพดานทรัพยากรและขนาด input ที่เหมาะสมทำให้ระบบถูกใช้งานหนักเกินคาดได้ บทนี้เลือก control ที่วัดได้หนึ่งอย่าง: นับ bytes ที่ยอมรับก่อน parse และกำหนด pull cap เพื่อจำกัดจำนวนครั้งที่ขอข้อมูล มันไม่ใช่การแก้ความเสี่ยง API ทั้งหมด OWASP API Security: Unrestricted Resource Consumption

• คำถามเรื่องขนาดตอบด้วยการนับ bytes ที่อ่านจริง
• คำถามเรื่องความครบถ้วนตอบด้วย EOF
• คำถามว่า parse ได้หรือไม่ตอบโดย decoder/parser
• คำถามว่าควรให้ agent ใช้หรือไม่ตอบด้วย validation และ policy

จากแชต สู่ acquisition tool ที่มีสัญญาชัด

ในหน้าต่างแชต คนอ่านข้อความยาวแล้วหยุดเองได้ โปรแกรมที่อ่าน response ต้องมีกติกาหยุดที่ตรวจได้ หาก agent ส่งทุก byte เข้า context โดยไม่กำหนดงบ ขนาด input เวลา และค่าใช้จ่ายจะคาดเดายาก การย้ายงานอ่านไปยัง deterministic acquisition tool ช่วยให้ระบบหยุดก่อนส่งข้อมูลที่ไม่มีขอบเขตต่อไปยังโมเดล

กำหนด input ของ tool เช่น resource reference ที่ผ่านนโยบายปลายทางแล้ว และ output แบบ {outcome, body, accepted_bytes, pulls} ขั้น acquisition รวบรวม bytes ตามงบ ส่วน decode, parse, schema validation, relevance filtering และการตัดสินใจของ agent เป็นขั้นถัดไป ผู้เรียกตรวจ outcome ก่อนอ่าน body

เลือก all-or-nothing โดยเจตนา: มี body ให้ใช้เมื่อสังเกต EOF ครบและทุก chunk เป็น bytes ภายในงบเท่านั้น เมื่อเกิน limit หรือยังไม่จบตาม pull cap ผลลัพธ์ไม่มี body วิธีนี้ลดโอกาสที่ caller จะส่ง prefix ที่ขาดท้ายให้ JSON parser แล้วเผลอใช้เป็นข้อมูลจริง

ชั้น	หน้าที่	ตัวอย่างคำถาม
Collector	จำกัดและรวบรวม bytes	รับครบในงบหรือไม่?
Decoder	แปลง bytes เป็นข้อความ	encoding ใช้ได้ไหม?
Parser	สร้างโครงสร้างข้อมูล	JSON syntax ถูกหรือไม่?
Validator	ตรวจ schema และ policy	field/ค่าที่อนุญาตไหม?
Agent	ใช้ผลที่ผ่านเงื่อนไข	ข้อมูลตอบโจทย์หรือไม่?

สัญญาของตัวอย่าง

ตัวอย่างที่วางแผนใช้ Python 3.12 ขึ้นไป, standard library และ iterable จำลองเท่านั้น ไม่มี socket, API call หรือ dependency install จึงเป็น mock collector สำหรับฝึก contract ไม่ใช่ production HTTP client ผลลัพธ์มี outcome, body ซึ่งเป็น bytes หรือ None, accepted_bytes และ pulls

รับ max_bytes ที่เป็น exact int ตั้งแต่ศูนย์ขึ้นไป และ max_pulls exact int ตั้งแต่หนึ่งขึ้นไป ค่าผิดโยน ValueError ก่อนเรียก iter(source) การตรวจ exact type ตั้งใจปฏิเสธ bool เพราะ bool เป็น subclass ของ int ใน Python แต่ไม่ใช่การกำหนดงบที่เจตนาชัดเจน

chunk ที่ไม่ใช่ bytes ให้ INVALID_CHUNK; chunk ที่จะทำให้ยอดเกินให้ BODY_LIMIT ก่อน append; ใช้ pull cap ก่อนพบ EOF ให้ PULL_LIMIT; พบ EOF ภายในงบจึง COLLECTED ทุก failure คืน body=None และไม่ retry

กรณี	outcome	body	ความหมาย
EOF ภายในงบ	COLLECTED	ครบ	รับครบเท่านั้น ไม่รับรอง JSON
chunk เกิน byte cap	BODY_LIMIT	None	ไม่ append chunk นั้น
หมด pull cap ก่อน EOF	PULL_LIMIT	None	ยังยืนยันความครบไม่ได้
ชนิด chunk ไม่ใช่ bytes	INVALID_CHUNK	None	source ผิด contract
iterator โยน exception	READ_ERROR	None	ไม่เปิดเผยข้อความ exception
config ผิด	ValueError	ไม่มีผล	ตรวจค่าก่อนแตะ source

อัลกอริทึม: ตรวจพื้นที่เหลือก่อน append

เริ่ม accepted_bytes=0 และ buffer ว่าง ก่อนเรียก next ให้ตรวจว่าจำนวน pulls ยังต่ำกว่า max_pulls จากนั้นนับ attempt เมื่อได้ chunk ตรวจชนิดว่าเป็น bytes แล้วคำนวณพื้นที่เหลือ max_bytes - accepted_bytes ถ้า len(chunk) มากกว่าพื้นที่เหลือ คืน BODY_LIMIT โดยไม่ append ถ้าพอดีหรือน้อยกว่า จึง append และเพิ่ม accepted_bytes

เมื่อต้องอ่านต่อและพบ EOF ให้คืน COLLECTED พร้อม body ครบ การเรียก next ที่พบ EOF ยังนับหนึ่ง pull ทำให้ pull cap จำกัด attempts จริง ไม่ใช่เฉพาะจำนวน chunk ที่ยอมรับแล้ว การใช้พื้นที่เหลือทำให้ตรวจขอบเขตแต่ละ chunk ได้ตรงไปตรงมา

อย่าตัด chunk ที่เกินให้พอดี เพราะ prefix ที่ถูกตัดอาจกลายเป็น JSON ขาดท้ายและถูก caller ใช้ผิด ผลที่ชัดเจนคือ body ครบหรือไม่มี body ส่วน error counters ช่วยระบุเหตุโดยไม่เผยเนื้อหา response

Worked trace: นับ bytes และ pulls

กำหนด chunks เป็น b'{"v"' ยาว 4 bytes, b':123' ยาว 4 bytes และ b'45}' ยาว 3 bytes รวม 11 bytes ถ้า iterator จบหลังสาม chunks การตรวจ EOF ต้องใช้ pull ที่สี่

cap 10: pull 1 สะสม 4, pull 2 สะสม 8, pull 3 ได้ chunk ยาว 3 ซึ่งเกินพื้นที่เหลือ 2 ผลคือ BODY_LIMIT, accepted_bytes=8, pulls=3, body=None chunk ที่เกินถูกดึงจาก source แล้ว แต่ไม่ append

cap 11 และ max_pulls 4: รับครบ 11 bytes จากสาม chunks แล้ว pull 4 พบ EOF ผลคือ COLLECTED, accepted_bytes=11, pulls=4 และ body ครบ

cap 11 และ max_pulls 3: รับครบ 11 bytes แต่ไม่มี pull เหลือเพื่อสังเกต EOF ผลคือ PULL_LIMIT, accepted_bytes=11, pulls=3, body=None จำนวน bytes ที่พอดีเพดานยังไม่พิสูจน์ว่า source จบ

แถว BODY_LIMIT และ PULL_LIMIT สะสม bytes บางส่วนไว้ภายใน แต่ไม่คืน prefix ผลลัพธ์ที่แสดงเป็น expected output ที่คำนวณจากสัญญา ไม่ใช่ผลจากการรันโปรแกรม

max_bytes	max_pulls	ผลและเหตุการณ์
10	4 หรือมากกว่า	BODY_LIMIT; accepted=8, pulls=3, body=None
11	4	COLLECTED; accepted=11, pulls=4, body ครบ
11	3	PULL_LIMIT; accepted=11, pulls=3, body=None

ขอบเขตเป็น bytes ไม่ใช่อักขระ

HTTP body อ่านเป็น bytes และขอบเขต chunk ไม่จำเป็นต้องตรงกับขอบเขต UTF-8 character ตัวอักษรหนึ่งอาจประกอบหลาย bytes และถูกแบ่งคาบระหว่าง chunks ได้ การ decode ทีละ chunk จึงอาจผิดพลาดทั้งที่ body รวมกันถูกต้อง ให้สะสม bytes ก่อน แล้ว decode หลังผ่าน limit และสังเกต EOF เท่านั้น

ถ้า decode ไม่ได้ ให้ผลเป็น error ของ decoding ในชั้นถัดไป ไม่เปลี่ยนผล collector ให้เหมือนสำเร็จและไม่ส่ง prefix กลับ ผู้เรียกต้องแยกสถานะการรวบรวมกับสถานะการแปลง encoding ออกจากกัน

Content-Length ถ้ามีเป็น hint สำหรับปฏิเสธเร็วเมื่อค่าที่ประกาศใหญ่กว่างบ ใน measurement domain เดียวกับ limit เช่น bytes ที่ client ส่งต่อให้ collector อย่าเชื่อ header ว่าพิสูจน์ขนาดจริง เพราะอาจไม่มีหรือไม่ตรงกับ bytes ที่ iterator ส่ง การบังคับจริงยังต้องนับ bytes ที่อ่านได้

encoded กับ decoded body มีขนาดต่างกันได้ HTTPX iter_bytes() ให้ decoded content ขณะที่ iter_raw() ให้ raw bytes; เลือกตำแหน่งวัดให้ตรงกับความเสี่ยงที่ต้องควบคุม หากต้องจำกัดทั้ง compressed input และ decompressed output ต้องตรวจทั้งสอง layer ด้วย adapter ที่เหมาะสม HTTPX streaming responses

อย่าข้ามข้อผิดพลาดของ HTTP framing หรือ header ที่ client ตรวจพบ การนับ bytes ใน collector ไม่ได้แทนการตรวจ protocol: RFC 9110 ระบุว่า Content-Length เกี่ยวข้องกับจำนวน octets และการแบ่งขอบเขตข้อความ จึงต้องให้ transport จัดการค่าที่ผิดหรือไม่สอดคล้องอย่างถูกต้อง RFC 9110 §8.6

จุดวัด	ความหมาย	ข้อควรระวัง
raw bytes	ข้อมูลก่อนถอด content encoding	ไม่เท่ากับขนาด decoded
decoded bytes	ข้อมูลหลัง client decompress	อาจขยายมากกว่า raw
Content-Length	ค่าที่ต้นทางประกาศ	ใช้เป็น hint ไม่แทนการนับ

โค้ด collector และ tests สำหรับฝึกด้วย mock

บันทึกโมดูลด้านล่างเป็น response_budget.py และชุดทดสอบเป็น test_response_budget.py ในโฟลเดอร์เดียวกัน ใช้ Python 3.12 ขึ้นไปและ standard library เท่านั้น ไม่ต้องติดตั้ง package เพิ่ม ไม่มี environment variable หรือ API key ตัวอย่างใช้ข้อมูลจำลองในหน่วยความจำและพิมพ์ counters ไม่มีค่าบริการ API แต่ยังใช้ทรัพยากรเครื่องตามปกติ

ผู้เรียกต้องตรวจ outcome ก่อนเข้าถึง body; body มีค่าเฉพาะ COLLECTED จากนั้น decode ด้วย encoding ที่กำหนด, จัดการ decode error, parse JSON และ validate schema แยกกัน ห้ามตีความ COLLECTED ว่า JSON_VALID, SAFE หรือ TASK_SUCCESS

เอกสาร Python เตือนให้ระวังการ parse JSON จากแหล่งที่ไม่น่าเชื่อถือเพราะข้อมูลอาจกิน CPU และ memory มาก การจำกัด input bytes ลดขนาด input ที่ parser เห็น แต่ไม่ได้จำกัดทรัพยากรหลัง parse โดยตรง Python 3.12 json

ผลของ mock iterable บอกเพียง collector behavior ตามสัญญา ไม่ได้ทดสอบ HTTP framing, socket timeout, connection pooling หรือ cancellation การเชื่อมต่อจริงต้องใช้ client ที่จัดการ protocol และอ่าน stream ตาม API ของมัน อย่าประดิษฐ์ parser HTTP เองจาก bytes โดยละเลย framing

"""Bounded acquisition of bytes from a finite, caller-provided iterable.

This module deliberately has no HTTP, parsing, retry, logging, or socket code.
"""
from dataclasses import dataclass
from typing import Iterable


@dataclass(frozen=True)
class Result:
    outcome: str
    body: bytes | None
    accepted_bytes: int
    pulls: int


def collect_body(chunks: Iterable[bytes], max_bytes: int, max_pulls: int) -> Result:
    """Collect bytes under a retained-body and iterator-pull limit.

    Pull count includes every attempted next(), including the one that observes
    EOF. A PULL_LIMIT means EOF was not observed within the configured budget.
    """
    if type(max_bytes) is not int or max_bytes < 0:
        raise ValueError("max_bytes must be an int >= 0")
    if type(max_pulls) is not int or max_pulls < 1:
        raise ValueError("max_pulls must be an int >= 1")

    try:
        iterator = iter(chunks)
    except Exception:
        return Result("READ_ERROR", None, 0, 0)

    buffer = bytearray()
    pulls = 0
    while pulls < max_pulls:
        pulls += 1
        try:
            chunk = next(iterator)
        except StopIteration:
            return Result("COLLECTED", bytes(buffer), len(buffer), pulls)
        except Exception:
            return Result("READ_ERROR", None, len(buffer), pulls)

        if type(chunk) is not bytes:
            return Result("INVALID_CHUNK", None, len(buffer), pulls)
        if len(chunk) > max_bytes - len(buffer):
            return Result("BODY_LIMIT", None, len(buffer), pulls)
        buffer.extend(chunk)

    return Result("PULL_LIMIT", None, len(buffer), pulls)


if __name__ == "__main__":
    samples = (
        ("exact-cap", (b'{"v"', b":123", b"45}"), 11, 4),
        ("over-cap", (b'{"v"', b":123", b"45}"), 10, 4),
    )
    for name, chunks, cap, pulls in samples:
        result = collect_body(chunks, cap, pulls)
        print(f"{name}: outcome={result.outcome} accepted_bytes={result.accepted_bytes} pulls={result.pulls}")

คำสั่งสำหรับผู้เรียน: python response_budget.py ผลสองบรรทัดด้านล่างเป็นผลที่คาดจากการอ่านโค้ด สถานะ NOT RUN

exact-cap: outcome=COLLECTED accepted_bytes=11 pulls=4
over-cap: outcome=BODY_LIMIT accepted_bytes=8 pulls=3

บันทึกชุดต่อไปนี้เป็น test_response_budget.py มี 17 test methods ที่ออกแบบไว้ ใช้ python -m unittest -v test_response_budget.py หากทุกกรณีผ่านจึงคาดว่าจะเห็น Ran 17 tests และ OK เวลาและผลจริงยังไม่มี เพราะ NOT RUN

"""Designed unittest coverage. NOT RUN."""
import unittest

from response_budget import Result, collect_body


class ExplodingIterable:
    def __iter__(self):
        raise RuntimeError("secret iterator detail")


class ExplodingReader:
    def __iter__(self):
        return self

    def __next__(self):
        raise RuntimeError("secret reader detail")


class ResponseBudgetTests(unittest.TestCase):
    def test_empty_body_observes_eof(self):
        self.assertEqual(collect_body([], 0, 1), Result("COLLECTED", b"", 0, 1))

    def test_exact_cap_then_eof(self):
        self.assertEqual(collect_body([b"ab", b"cd"], 4, 3), Result("COLLECTED", b"abcd", 4, 3))

    def test_over_cap_returns_no_prefix(self):
        self.assertEqual(collect_body([b"ab", b"c"], 2, 3), Result("BODY_LIMIT", None, 2, 2))

    def test_canonical_cap_10_trace(self):
        got = collect_body((b'{"v"', b":123", b"45}"), 10, 4)
        self.assertEqual(got, Result("BODY_LIMIT", None, 8, 3))

    def test_canonical_cap_11_trace(self):
        got = collect_body((b'{"v"', b":123", b"45}"), 11, 4)
        self.assertEqual(got, Result("COLLECTED", b'{"v":12345}', 11, 4))

    def test_empty_chunks_consume_pull_budget(self):
        def endless_empty():
            while True:
                yield b""
        self.assertEqual(collect_body(endless_empty(), 4, 3), Result("PULL_LIMIT", None, 0, 3))

    def test_string_chunk_is_invalid(self):
        self.assertEqual(collect_body(["text"], 10, 2), Result("INVALID_CHUNK", None, 0, 1))

    def test_bytearray_chunk_is_invalid(self):
        self.assertEqual(collect_body([bytearray(b"x")], 10, 2), Result("INVALID_CHUNK", None, 0, 1))

    def test_bool_byte_limit_rejected_before_iter(self):
        with self.assertRaises(ValueError):
            collect_body(ExplodingIterable(), True, 2)

    def test_negative_limit_rejected(self):
        with self.assertRaises(ValueError):
            collect_body([], -1, 2)

    def test_zero_pull_limit_rejected_before_iter(self):
        with self.assertRaises(ValueError):
            collect_body(ExplodingIterable(), 4, 0)

    def test_bool_pull_limit_rejected(self):
        with self.assertRaises(ValueError):
            collect_body([], 4, True)

    def test_iterator_creation_error_is_sanitized(self):
        self.assertEqual(collect_body(ExplodingIterable(), 4, 2), Result("READ_ERROR", None, 0, 0))

    def test_reader_error_is_sanitized_and_has_no_body(self):
        got = collect_body(ExplodingReader(), 4, 2)
        self.assertEqual(got, Result("READ_ERROR", None, 0, 1))
        self.assertNotIn("secret", repr(got))

    def test_eof_attempt_counts_as_pull(self):
        self.assertEqual(collect_body([b"x"], 4, 2), Result("COLLECTED", b"x", 1, 2))

    def test_budget_exhaustion_without_observed_eof(self):
        self.assertEqual(collect_body([b"x"], 4, 1), Result("PULL_LIMIT", None, 1, 1))

    def test_split_utf8_reassembles_as_bytes(self):
        encoded = "ไทย".encode("utf-8")
        split = encoded[:2], encoded[2:5], encoded[5:]
        got = collect_body(split, len(encoded), 4)
        self.assertEqual(got.outcome, "COLLECTED")
        self.assertEqual(got.body, encoded)
        self.assertEqual(got.body.decode("utf-8"), "ไทย")


if __name__ == "__main__":
    unittest.main()

ความล้มเหลวและการกู้คืน

INVALID_CHUNK บอกว่า source ไม่ทำตาม contract ส่วน READ_ERROR บอกว่า iterator โยน exception ทั้งสองคืน body=None และไม่ใส่ exception text ใน result เพราะข้อความอาจเปิดเผยรายละเอียดภายในหรือเนื้อหาที่ไม่ควรส่งต่อ หากต้องบันทึก log ให้ใช้ outcome และบริบทที่ไม่เป็นความลับตามนโยบายระบบ

ความผิดพลาดจาก iter(source) เกิดก่อน pull จึง pulls=0; ความผิดพลาดจาก next() เกิดระหว่าง attempt ที่นับแล้ว ตัวอย่างจำลองนี้กำหนดผลให้แน่นอนเพื่อแยก source error ออกจาก limit ที่ collector ควบคุมได้

อย่าทำ retry ซ่อนอยู่ใน collector การลองใหม่เพิ่ม byte รวม, load และความเสี่ยงที่ผล response เปลี่ยน หากชั้นเรียก tool ต้อง retry ให้กำหนดจำนวน attempts, deadline, idempotency, aggregate byte budget และเงื่อนไขที่ห้าม retry แยกต่างหาก งานนี้ไม่มี retry และไม่มีข้อความ server exception

สิ่งที่รับประกันและสิ่งที่ไม่รับประกัน

สัญญาที่ออกแบบไว้จำกัดความยาวข้อมูลที่ append ลง buffer ไม่เกิน max_bytes และจำนวน next attempts ไม่เกิน max_pulls พร้อมกำหนดให้ failure ไม่มี partial body สัญญานี้ตรวจสอบได้ด้วย input ที่กำหนดแน่นอน แต่บทความนี้ยังไม่ได้รันเพื่อยืนยันผล

ไม่ได้กำหนดเพดานหน่วยความจำรวม เพราะ source อาจจัดสรร chunk มาก่อนตรวจ และ bytearray มี allocation overhead ของตนเอง การแปลงเป็น bytes ตอนสำเร็จเพิ่มสำเนาข้อมูลอีกชุดยาวไม่เกิน B ขณะที่ buffer เดิมอาจยังอยู่ ส่วน decoder, decompression และ JSON parser อาจสร้างข้อมูลหรือ objects เพิ่ม เพดานความยาวข้อมูลที่สะสมจึงเป็นเพียง control หนึ่ง

ไม่ได้รับประกัน latency หรือการหยุดอ่านทันที next อาจ block ก่อนคืน chunk ดังนั้น pull cap ไม่ใช่ timeout adapter จริงต้องตั้ง timeout/deadline ที่ HTTP client รองรับ ปิด response เมื่อหยุดอ่าน และใช้ cancellation ตามความเหมาะสม หากหยุดทางตรรกะแต่ไม่ปิด stream อาจยังถือ connection หรือทรัพยากรอยู่

ไม่ได้ตรวจ framing, header syntax, status, redirect, TLS, compression bomb, content type, JSON schema, prompt injection หรือความถูกต้องเชิงความหมาย ให้ HTTP library จัดการ framing และ protocol errors อย่า parse raw HTTP เองโดยละเลย framing

ขอบเขต	ตัวอย่างนี้คุมไหม
bytes ที่ append	คุมได้
จำนวน next attempts	คุมได้
chunk allocation ก่อนตรวจ	ไม่คุม
เวลา block ใน next	ไม่คุม
memory หลัง parse	ไม่คุม
HTTP protocol / endpoint policy	ไม่คุม

Task brief แบบคัดลอกได้

ใช้ brief นี้มอบหมายงาน implementation ให้ทีมได้ ปรับ path ตาม repository แต่รักษาขอบเขตให้ตรวจซ้ำได้:

• เป้าหมาย: สร้าง Python 3.12+ collector จำกัด response bytes และ pull attempts ก่อน decode/parse JSON
• ข้อมูลเข้าและเครื่องมือ: iterable ของ exact bytes, max_bytes exact int >= 0, max_pulls exact int >= 1; standard library เท่านั้น; config ผิดต้อง ValueError ก่อน iter
• สิทธิ์และขอบเขต: mock ในหน่วยความจำ ไม่มี network, API, dependency install, retry หรือ deploy; HTTP adapter และ JSON parser อยู่นอกโมดูล
• สิ่งส่งมอบ: Result(outcome, body bytes|None, accepted_bytes, pulls) และ tests; outcomes มี COLLECTED, BODY_LIMIT, PULL_LIMIT, INVALID_CHUNK, READ_ERROR; ทุก failure ไม่คืน prefix หรือ exception text
• เกณฑ์รับงาน: exact cap + EOF, overflow ก่อน append, นับ EOF หนึ่ง pull, PULL_LIMIT เมื่อยังไม่เห็น EOF, invalid chunk, iter/next exceptions และ config ก่อนแตะ source ต้องตรงกับ cases
• จุดหยุด: ส่งโมดูลและหลักฐาน tests จาก environment ที่ได้รับอนุญาตให้คนทบทวน; ถ้ายังไม่รันให้ระบุ NOT RUN หยุดก่อนเรียก API จริง เพิ่ม dependency หรือ deploy ไม่อ้าง memory/latency guarantees

Acceptance criteria และวิธีใช้งาน

ทดสอบ exact boundary: body ที่ขนาดเท่ากับ cap และจบด้วย EOF ต้อง COLLECTED เมื่อ pull cap อนุญาตให้พบ EOF; ถ้าขนาดถัดไปเกิน ให้ BODY_LIMIT ก่อน append chunk ที่เกิน; ถ้าใช้ pulls หมดก่อน EOF ให้ PULL_LIMIT แม้ accepted_bytes เท่ากับ byte cap

ทดสอบ failure paths: chunk ชนิดผิด, iter(source) โยน, next โยน และค่าตั้งผิด ตรวจว่า failure body=None, exception text ไม่ถูกส่งคืน, ค่า config ผิดทำให้ source ไม่ถูกแตะ และ pulls ของ iter failure เป็นศูนย์ ส่วน error ที่เกิดใน next นับ attempt นั้น

คำสั่งที่เตรียมไว้คือ python response_budget.py และ python -m unittest -v test_response_budget.py บน Python 3.12+ คำสั่งเหล่านี้มีสถานะ NOT RUN ในการจัดทำบทความนี้ การทดลองระดับ Sandbox และ Live integration ยังไม่ได้ทำ

เงื่อนไข	ผลที่คาดตามสเปก
body เท่ากับ cap และพบ EOF	COLLECTED
chunk ถัดไปทำให้เกิน cap	BODY_LIMIT ก่อน append
bytes เต็มแต่ pull หมดก่อน EOF	PULL_LIMIT
config ผิด	ValueError ก่อน iter
iter ล้มเหลว	READ_ERROR, pulls=0

แบบฝึกหัดพร้อมเฉลยและ rubric

แบบฝึกหัด: source ให้ b'ABCD', b'EF' แล้ว EOF ตั้ง max_bytes=6, max_pulls=3 จงระบุ outcome, body, accepted_bytes และ pulls จากนั้นลด max_pulls เหลือ 2 แล้วอธิบายว่าทำไมผลเปลี่ยน แม้ byte count จะเท่าเดิม

เฉลย: กรณีแรก pull ที่หนึ่งรับ 4 bytes, pull ที่สองรับ 2 bytes, pull ที่สามพบ EOF จึง COLLECTED, body=b'ABCDEF', accepted_bytes=6, pulls=3 กรณีสองสอง pulls ใช้กับ chunks ทั้งหมด แต่ยังไม่พบ EOF จึง PULL_LIMIT, body=None, accepted_bytes=6, pulls=2

Rubric 4 คะแนน: 1 คะแนนนับ bytes ถูก, 1 คะแนนนับ EOF attempt, 1 คะแนนอธิบายว่า bytes ครบไม่เท่ากับ stream จบ, 1 คะแนนระบุว่า PULL_LIMIT ไม่มี body ถ้าตอบว่ากรณีสองสำเร็จเพราะ body ยาวพอดี cap ให้ทบทวนว่า EOF เป็นหลักฐานความครบถ้วนใน contract นี้

3D Lab: สำรวจ resource boundary

CSS3D Lab ด้านล่างแยกสามด่าน Intake, Gate และ Parser ให้เลือกด่านด้วยปุ่มเพื่อดูเหตุและผล: chunk ถูกสร้างจากต้นทางก่อนตรวจเพดาน ส่วน parser ยังต้องรอการรับข้อมูลครบ ภาพมีความลึกจาก CSS perspective และ transform

ใช้ภาพจำลองเพื่ออ่าน architecture ไม่มี HTTP request หรือผลวัด memory/latency ปุ่มใช้โครงสร้าง HTML ที่รองรับ keyboard และมี CSS reduced-motion ส่วนผลการใช้งานบนอุปกรณ์จริงยังไม่ได้ตรวจ

Visual Interactive: ทดลอง byte และ pull caps

Interactive ในกรอบเดียวกันเป็นกิจกรรมแยกจากฉาก 3D ใช้ chunks คงที่ 4, 4, 3 bytes ปรับเพดาน byte และจำนวน pulls ด้วยแถบเลื่อน แล้วอ่านยอด accepted_bytes, pulls และ terminal outcome ก้อนที่เกินใช้ pull แต่ไม่เพิ่ม accepted_bytes

มี labels, aria-live และตารางกรณีอ้างอิงที่อยู่ใน HTML ให้อ่านแม้สคริปต์ไม่ทำงาน เป็นการออกแบบ simulation ยังไม่มีผลทดสอบ keyboard, screen reader หรือ browser ให้รายงาน

• ลอง cap 10 กับ chunks 4,4,3
• ลอง cap 11 และ pull cap 4 เพื่อเห็น EOF
• ลอง cap 11 และ pull cap 3 เพื่อเห็น PULL_LIMIT

Video Lab: วางแผนคลิปอธิบายงบ response

ใช้ Video Lab ด้านล่างวางแผนคลิปเรื่อง “นับ bytes ก่อน parse JSON” เพื่อทบทวนบทเรียน ลองเลือกความยาว 60 วินาทีและคำบรรยายไทย แล้วนำ prompt ที่ได้มาเทียบกับ brief หกส่วนต่อไปนี้ กิจกรรมนี้วางแผนสื่อ ไม่ได้อ่าน HTTP หรือเรียก endpoint จริง

Brief หกส่วนสำหรับคัดลอก: (1) ผู้ชม—คนที่ใช้ LLM chat และเริ่มสร้าง agent; (2) เป้าหมาย—อธิบาย byte cap, pull cap และ EOF; (3) ฉาก—chunks 4+4+3 bytes กับ caps 10/11; (4) ลำดับ—นับ pulls, คำนวณพื้นที่เหลือ, append หรือหยุด, ตรวจ EOF; (5) เกณฑ์—BODY_LIMIT 8 bytes/3 pulls, COLLECTED 11/4, PULL_LIMIT 11/3 ตรงกับ trace และติดป้ายว่า simulation; (6) ขอบเขต—ไม่มี API จริง ไม่มี memory/latency claim และหยุดก่อนแนะนำ production adapter เกินข้อมูล

ต้นทุนที่อาจแสดงใน Lab เป็น mock/synthetic assumption เพื่อฝึกวางแผนเท่านั้น ไม่ใช่ราคาจริงหรือผล render หากผู้เรียนกรอกค่า ให้ระบุ estimated และตรวจราคาจากผู้ให้บริการก่อนใช้งาน ไม่มีการสร้างวิดีโอหรือเรียกบริการเสียเงินในบทนี้

Shortcode นี้เชื่อมกับทักษะการเขียน brief และตรวจ acceptance ของวิดีโอเท่านั้น ส่วนเนื้อหา HTTP acquisition สอนด้วยตัวอย่างและ interactive ของบท ไม่กล่าวว่า Lab สร้างหรืออ่าน HTTP response

H✦HYBRID MINDINTERACTIVE EDITION LIVE EXPERIENCE

HYBRID MIND ORIGINAL / AI AGENT WORKFLOW / 2026

จากคลิปดิบ
สู่จักรวาลคอนเทนต์

ChatGPT ตัดต่อวิดีโอแทนคนได้จริงแค่ไหน? สำรวจระบบ AI Video Editing Agent ผ่านฉากสามมิติที่โต้ตอบได้ พร้อมแยกข้อเท็จจริง สร้างพรอมป์ต์ และจำลองต้นทุนก่อนลงมือทำจริง

RAW / FOOTAGE▥

AI / TIMELINE⟡

HYBRID MIND · 9:16▶RENDER / MP4

✦ AI-ASSISTED EDIT

สำรวจ 3D Workflow ↗สร้าง Prompt ของฉัน →

06 EDITING STAGES03 CONTENT FORMATS01 INTERACTIVE LAB

01 / EXPLORE

ออกแบบระบบตัดต่อในมุมมอง 3D

ลากบนฉากเพื่อหมุนมุมมอง แตะโหนดเพื่อดูหน้าที่ของ AI ในแต่ละขั้นตอน

ORBITFRONTTOPⅡ

แผนภาพ 6 ขั้นตอน: Footage, Story, Edit, Captions, Graphics, Render

AGENT PIPELINE / VIEWPORT ORBIT 3D

DRAG / SWIPE TO ROTATE
SELECT NODE TO INSPECT

01 / 06นำเข้าคลิปดิบ

จัดระเบียบไฟล์ ตรวจคุณภาพภาพและเสียง และกำหนดสิทธิ์เข้าถึงให้ AI Agent

● SOURCE

01 FOOTAGE02 STORY03 EDIT04 CAPTIONS05 GRAPHICS06 RENDER

จุดสำคัญ: AI Agent ทำหน้าที่วางแผนและควบคุมเครื่องมือ ส่วนการตัดต่อจริงต้องมีซอฟต์แวร์หรือระบบเรนเดอร์รองรับ ฉากนี้เป็น แบบจำลองเชิงอธิบาย ไม่ใช่ระบบอัปโหลดหรือตัดต่อวิดีโอ

02 / FACT CHECK

แยกความสามารถจริงจากคำโฆษณา

กดแต่ละประเด็นเพื่ออ่านรายละเอียดและแหล่งข้อมูล

EVIDENCE-FIRST

✓ChatGPT Work เข้าถึงไฟล์ในเครื่องได้ตรวจสอบแล้ว / มีเงื่อนไข+

รองรับการทำงานกับโฟลเดอร์หรือแอปบนคอมพิวเตอร์เดสก์ท็อป เมื่อแผนและการตั้งค่าอนุญาตและผู้ใช้ให้สิทธิ์ ส่วน Work บนมือถือหรือเว็บไม่ได้เข้าถึงโฟลเดอร์ในคอมพิวเตอร์ของคุณโดยตรง ที่มา: OpenAI Help ↗

✓Astra ใช้โปรแกรมคอมพิวเตอร์ทำงานได้ตรวจสอบแล้ว / ไม่ใช่โปรแกรมตัดต่อในตัว+

OpenAI อธิบายความสามารถด้าน computer use ของ Astra รวมถึงงานสื่อที่ใช้เครื่องมือ แต่ต้องแยกบทบาท AI ผู้ควบคุมออกจากซอฟต์แวร์ประมวลผลวิดีโอและการเรนเดอร์ไฟล์ ที่มา: OpenAI ↗

?ตัดเสร็จใน 5–10 นาทีทุกกรณียังพิสูจน์เป็นเกณฑ์ทั่วไปไม่ได้+

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

?ถูกกว่าจ้างคน 10 เท่าต้องกำหนดต้นทุนให้เทียบกันได้+

ต้องรวมเวลาตรวจแก้ ลิขสิทธิ์เสียง เครื่องมือที่จ่ายเพิ่ม และต้นทุนโมเดล ยังไม่มีหลักฐานว่าประหยัด 10 เท่าสำหรับทุกงาน ทดลองสมมติฐานส่วนตัวได้ใน Cost Simulator ด้านล่าง

03 / CREATOR STUDIO

หนึ่งไอเดีย สามรูปแบบคอนเทนต์

ปรับรูปแบบวิดีโอ แล้วดูกรอบภาพตัวอย่างและ Prompt เปลี่ยนตามแบบเรียลไทม์

WORKING INTERACTIVE

01 / เลือกแพลตฟอร์ม

TikTokReelsYouTube

02 / ประเภทเนื้อหาคลิปให้ความรู้ AIคลิปแนะนำสินค้า Affiliateตัดไฮไลต์จากไลฟ์

03 / หัวข้อของคลิป

04 / ความยาวเป้าหมาย 45 วินาที

15 SEC180 SEC

ซับไทยทั้งคลิปMotion Graphicsเพิ่มเสียงพากย์ (เมื่อมีเครื่องมือ)

สร้าง Prompt ↗คัดลอก Prompt

LIVE CREATIVE PREVIEW9:16 · 45 SEC

HYBRID MIND / AI

AI EDITOR / EXPLAINEDAI Agent
ช่วยตัดต่อ?

ใส่ซับตามประโยคได้อัตโนมัติ

▶◉↗

CONCEPT PREVIEW · NOT RENDERED VIDEO

VERTICAL · SHORT FORMVisual Mockup

YOUR GENERATED PROMPT

พร้อมนำไปใช้กับ AI Agent

COPY PROMPT ↗

ตัวสร้าง Prompt นี้ไม่อัปโหลดคลิป ไม่เรียก API เสียเงิน และไม่ตัดต่อวิดีโอจริง ต้องใช้กับเครื่องมือที่ผู้ใช้อนุญาตภายหลัง

04 / EXPERIMENTAL SIMULATOR

จำลองเวลาและต้นทุนของคุณเอง

เลื่อนตัวเลขเพื่อเปรียบเทียบสถานการณ์สมมติ ไม่ใช่ผล benchmark ที่ยืนยันแล้ว

SCENARIO · NOT BENCHMARK

จำนวนคลิปต่อเดือน 12 คลิปเวลาตัดเองต่อคลิป 90 นาทีเวลาทำงานด้วย AI รวมตรวจแก้ 35 นาทีมูลค่าเวลาของคุณต่อชั่วโมง ฿400ค่าเครื่องมือ AI ต่อคลิป ฿40

ESTIMATED TIME SAVED11.0 ชม./เดือน

MANUAL

18.0 ชม.

AI-ASSISTED

7.0 ชม.

ส่วนต่างต้นทุนเวลา + เครื่องมือ (สมมติ)฿3,920

ตัวเลขเป็นตัวอย่างเพื่อช่วยตัดสินใจ ไม่ใช่ผลประหยัดที่รับประกัน

05 / THE DEEP DIVE

AI ที่ตัดคลิปเป็น ไม่ได้ทำงานลำพัง

สิ่งที่น่าสนใจของ GPT-6 Astra และ GPT-6.1 Sol ไม่ได้อยู่ที่ว่าตัวโมเดลมีปุ่ม Render MP4 ซ่อนอยู่ แต่คือความสามารถในการจัดการงานหลายขั้นตอนร่วมกับเครื่องมืออื่น: อ่านไฟล์ วางลำดับเรื่อง ใช้แอป ตรวจผลลัพธ์ และแก้งานตามข้อเสนอแนะของมนุษย์

สำหรับคนทำคอนเทนต์ ความต่างสำคัญคือการเปลี่ยนจาก “ส่งคลิปให้ AI แล้วหวังว่าจะได้ผลงาน” ไปเป็น Workflow ที่ตรวจรับได้ทุกขั้น ตั้งแต่การเลือกช่วงเด็ด การสร้างซับ การตรวจคำผิด จนถึงการเรนเดอร์ไฟล์ตามสัดส่วนของแต่ละแพลตฟอร์ม

ข้อควรระวังคือระบบ Work ที่รันบนเว็บหรือมือถือไม่ได้หมายความว่า AI เข้าถึงคลิปบนคอมพิวเตอร์บ้านได้ทันที ต้องทำงานในสภาพแวดล้อมที่เปิดให้เข้าถึงไฟล์จริง มีเครื่องมือตัดต่อ และมีสิทธิ์ใช้งานอย่างชัดเจน

01จัดคลิปให้พร้อม

รวบรวมไฟล์ในโฟลเดอร์ที่เข้าถึงได้ ระบุสิทธิ์และข้อจำกัด

02ตรวจสคริปต์และไทม์ไลน์

ให้ Agent เสนอ Hook, ช่วงตัด และโครงเรื่องก่อนตัดต่อ

03สร้างซับ + กราฟิก

ตรวจชื่อเฉพาะ ตัวเลข ภาษาไทย และความถูกต้องของคำกล่าวอ้าง

04ตรวจไฟล์ MP4 จริง

ตรวจความละเอียด เสียง ระยะเวลา safe zone และคำบรรยาย

FAQ

ถาม: ถ้า Content-Length น้อยกว่า cap เชื่อได้ไหม? ตอบ: ไม่ใช้แทนการนับจริง เป็น hint สำหรับ early rejection เท่านั้น

ถาม: ทำไมไม่คืน prefix เมื่อเกิน limit? ตอบ: prefix อาจเป็น JSON ขาดท้ายและ caller อาจนำไปใช้ผิด contract นี้เลือก all-or-nothing

ถาม: COLLECTED แปลว่าปลอดภัยไหม? ตอบ: ไม่ได้แปลเช่นนั้น ต้อง decode, parse, validate schema และประเมินเนื้อหาแยกกัน

ถาม: byte cap ป้องกัน memory exhaustion ทั้งหมดไหม? ตอบ: ไม่ upstream chunk, copies, decompression และ parsed objects อาจใช้ memory เพิ่ม

ถาม: pull cap กันการค้างไหม? ตอบ: ไม่ next อาจ block ต้องกำหนด timeout/deadline และปิด response ใน adapter จริง

แหล่งข้อมูลและบทเรียนต่อยอด

หลักการ resource boundary และสัญญา collector ในบทนี้เป็นการออกแบบตัวอย่างที่ผู้เขียนสร้างขึ้น แหล่งอ้างอิงรองรับพฤติกรรม streaming, ข้อควรระวัง JSON และ resource consumption ไม่ได้อ้างว่าแหล่งเหล่านี้เป็นที่มาของ implementation

• HTTPX: Streaming responses — ความต่างระหว่าง iter_bytes() ซึ่งให้ decoded content กับ iter_raw() ซึ่งให้ raw bytes; ตรวจเอกสาร 23 ตุลาคม 2026
• Python 3.12: json — เอกสารเตือนเรื่องการ parse JSON จาก input ที่ไม่น่าเชื่อถือและการใช้ CPU/memory; ตรวจเอกสาร 23 ตุลาคม 2026
• OWASP API Security 2023: Unrestricted Resource Consumption — การจำกัดทรัพยากรและปริมาณ input ตามความเหมาะสม; ตรวจเอกสาร 23 ตุลาคม 2026
• จำแนกข้อผิดพลาด HTTP ด้วย RFC 9457 Problem Details — บทต่อยอดการจำแนก error หลัง acquisition; บทนี้ไม่ทำซ้ำ RFC 9457
• RFC 9110 §8.6 Content-Length — ความหมาย octets และบทบาทต่อ framing; ตรวจ 23 ตุลาคม 2026

สถานะหลักฐานและขอบเขต

สเปกและ expected outputs เป็น authored design และ mock trace โค้ดตัวอย่างกับ tests มีสถานะ NOT RUN ส่วนการใช้งาน browser, keyboard, screen reader, reduced-motion และ Video Lab บนอุปกรณ์จริงยังเป็น UNKNOWN การอ่านเนื้อหาที่ WordPress บันทึกกลับมาเป็นหลักฐาน saved content เท่านั้น

ก่อนต่อ HTTP client จริง ให้กำหนด timeout, cancellation, close response, framing, compression, status/error handling และปลายทางที่อนุญาตแยกต่างหาก บทนี้ไม่ให้สิทธิ์เรียก API จริง ไม่ติดตั้ง dependencies และไม่ deploy ระบบ