Graph Engineering: วิธีรัน AI Agents 1,000 ตัวแบบขนานจาก Prompt เดียว

ทุกคนที่สร้างระบบหลายเอเจนต์ในปี 2026 ยังคงเขียนเส้นตรงอยู่ ก้าวแรก แล้วก้าวที่สอง แล้วก้าวที่สาม - แต่ละขั้นรอคอยขั้นก่อนหน้า นี่คือสาเหตุที่มันช้า และวิธีแก้ไข

ปัญหาที่ไม่มีใครตรวจสอบ

คุณสร้างเอเจนต์แบบหลายขั้นตอน มันทำงานได้ และมันก็ช้าด้วย

คุณคิดว่าตัวโมเดลคือคอขวด มันไม่ใช่

คอขวดคือรูปร่างที่คุณวาดขึ้นมา สายโซ่ - ขั้นตอนที่ 1 รอขั้นตอนที่ 2 ขั้นตอนที่ 2 รอขั้นตอนที่ 3 - บังคับให้ทำงานตามลำดับแม้ว่าครึ่งหนึ่งของขั้นตอนเหล่านั้นจะไม่เกี่ยวข้องกันเลยก็ตาม

"สรุปเอกสารนี้ แล้วตรวจสอบสภาพอากาศ" คืองานอิสระสองงานที่สวมเสื้อคลุมตัวเดียวกันปลอมตัวเป็นเวิร์กโฟลว์เดียว งานสภาพอากาศไม่ต้องการข้อมูลสรุป มันไม่เคยต้องการ แต่ถ้าคุณเขียนมันเป็นสายโซ่ มันก็ต้องรออยู่ดี

การรอคอยที่สูญเปล่า คูณด้วยหลายสิบขั้นตอน คือที่ที่รันไทม์ส่วนใหญ่ของคุณหายไป

บทที่ 1 - ลูป vs กราฟ

ลูปคือหน่วยหนึ่งของการปรับปรุงตัวเอง:

1ลองทำอะไรสักอย่าง → ตรวจสอบผลลัพธ์ → ปรับปรุง → ลองอีกครั้ง

นั่นคืออะตอม เอเจนต์หนึ่งตัว ตัวชี้วัดหนึ่งตัว วนซ้ำจนกว่าจะบรรจบกัน

ลูปมีรูปแบบความล้มเหลวที่รู้จักกันดี: พวกมันปรับให้เหมาะสมกับสิ่งที่คุณวัดเท่านั้นและไม่สนใจสิ่งอื่นใด บอทสนับสนุนที่ถูกปรับแต่งให้ปิดตั๋วอย่างรวดเร็ว จะปิดตั๋วอย่างรวดเร็ว - ในขณะที่ความพึงพอใจลดลงอย่างเงียบๆ ลูปไม่สามารถมองเห็นเกินตัวชี้วัดของตัวเองได้ นั่นคือกฎของ Goodhart ที่ปรากฏในสถาปัตยกรรมเอเจนต์ของคุณ

กราฟแก้ไขปัญหานี้ด้วยการออกแบบ แทนที่จะเป็นลูปเดียวที่ไล่ตามตัวเลขตัวเดียว คุณสร้างเครือข่ายของลูปที่คอยเฝ้าดูและแก้ไขซึ่งกันและกัน ผลลัพธ์ของโหนด A ป้อนให้โหนด B โหนด C ทำงานอิสระและตรวจสอบทั้งคู่ ไม่มีตัวชี้วัดตัวเดียวที่ขับเคลื่อนทั้งระบบ - โครงสร้างต่างหากที่ทำ

สำหรับระบบเอเจนต์ นี่หมายถึงการเปลี่ยนแปลงที่เป็นรูปธรรมอย่างหนึ่ง: หยุดเขียนเอเจนต์ตัวเดียวที่ทำทุกอย่างตั้งแต่ต้นจนจบ ออกแบบรูปร่างของงานก่อน - อะไรต้องเกิดขึ้นก่อนอะไร อะไรที่สามารถทำงานพร้อมกันได้ อะไรที่ต้องรอจริงๆ

บทที่ 2 - โหนด, ขอบ, และการทดสอบที่แยกพวกมันออกจากกัน

กราฟมีส่วนประกอบเพียงสองอย่าง:

โหนด - หน่วยของงานหนึ่งหน่วย เอเจนต์หนึ่งตัว งานหนึ่งชิ้น อินพุตหนึ่งชุด เอาต์พุตหนึ่งชุด

ขอบ - การพึ่งพาที่แท้จริง อินพุตของโหนด B ต้องการเอาต์พุตของโหนด A

ความผิดพลาดที่เกือบทุกคนทำ: การปฏิบัติต่อ "แล้วก็" เป็นขอบโดยค่าเริ่มต้น

1"อ่านโค้ดเบสนี้ แล้วเขียนบันทึกการเปลี่ยนแปลง"
2"ดึงหน้าราคา แล้วสรุปฟีเจอร์ของคู่แข่ง"

ถามหนึ่งคำถามสำหรับทุก "แล้วก็" ในเวิร์กโฟลว์ของคุณ:

ขั้นตอนถัดไปอ่านผลลัพธ์ของขั้นตอนก่อนหน้าจริงหรือไม่?

ถ้าใช่ → ขอบจริง รักษาลำดับตามลำดับไว้ ถ้าไม่ใช่ → ไม่มีขอบ การรอคอยนั้นสูญเปล่า ให้รันพวกมันแบบขนาน

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

นี่คือการทดสอบที่นำไปใช้ในโค้ด:

1from dataclasses import dataclass
2
3@dataclass
4class TaskNode:
5    id: str
6    prompt: str
7    depends_on: list[str]  # IDs of nodes this one actually needs
8
9def has_real_edge(node_a: TaskNode, node_b: TaskNode) -> bool:
10    """
11    The core graph engineering test:
12    does node_b's prompt actually require node_a's output?
13    """
14    return node_a.id in node_b.depends_on
15
16# Example: most "chains" collapse into 2-3 real dependency groups
17nodes = [
18    TaskNode("audit_routes", "List all API route files", []),
19    TaskNode("check_auth", "Check auth middleware coverage", []),
20    TaskNode("fetch_weather", "Get today's weather", []),
21    TaskNode("summarize", "Summarize route + auth findings",
22              depends_on=["audit_routes", "check_auth"]),
23]
24
25# audit_routes, check_auth, fetch_weather have NO edges between them
26# They run in parallel. Only "summarize" has real edges -- it waits.

เอเจนต์ "ทำ A, แล้ว B, แล้ว C" ในปัจจุบันของคุณ ในทางเทคนิคแล้วเป็นกราฟอยู่แล้ว มันเป็นเพียงกราฟที่แย่ที่สุดเท่าที่จะเป็นไปได้ - สายโซ่เดี่ยวที่ถ้า C หยุดชะงัก ไม่มีอะไรที่อยู่ downstream จะทำงานได้อีก

บทที่ 3 - การสร้างกราฟแรกของคุณ

ข้อกำหนด:

• Claude Code (เวอร์ชันล่าสุดที่รองรับ Dynamic Workflows)
• แผน Max, Team หรือ Enterprise - เวิร์กโฟลว์เปิดใช้งานโดยค่าเริ่มต้น สำหรับ Pro ให้เปิดใช้งานด้วยตนเอง

เปิด repository จริง ไม่ใช่ตัวอย่างเล่นๆ - ผลลัพธ์จะเห็นได้ชัดเจนในขนาดจริงเท่านั้น

พรอมต์ที่เริ่มต้นกราฟแรกของคุณ:

1Create a workflow to audit every route file in this codebase.
2
3For each route file, check independently:
4- authentication middleware present
5- input validation on all params
6- rate limiting configured
7- error handling doesn't leak stack traces
8
9Run these checks in parallel across all route files —
10they don't depend on each other.
11
12After all files are checked, produce one consolidated
13report grouped by severity: critical, warning, info.
14
15The consolidation step should wait for all checks to complete.
16Everything before it should not.

สังเกตโครงสร้างที่ฝังอยู่ในพรอมต์เอง: งานแบบขนานที่ถูกระบุอย่างชัดเจน การพึ่งพาที่แท้จริงเพียงอย่างเดียว (การรวบรวมรอการตรวจสอบทั้งหมดเสร็จสิ้น) ถูกระบุชื่ออย่างชัดเจน คุณไม่ได้หวังให้เอเจนต์อนุมานกราฟ - คุณกำลังอธิบายมัน

สิ่งที่เกิดขึ้นภายใต้ฝาครอบ - เวอร์ชันที่เรียบง่ายของการจัดเตรียม:

1import asyncio
2from anthropic import Anthropic
3
4client = Anthropic()
5
6async def audit_route_file(filepath: str) -> dict:
7    """One node. Runs independently of every other route file."""
8    response = await client.messages.create(
9        model="claude-sonnet-5",
10        max_tokens=1000,
11        messages=[{
12            "role": "user",
13            "content": f"""Audit this route file for:
14            - auth middleware, input validation,
15              rate limiting, error handling
16
17            File: {filepath}
18
19            Return JSON: {{"file": "", "issues": [], "severity": ""}}"""
20        }]
21    )
22    return {"file": filepath, "result": response.content[0].text}
23
24async def consolidate(results: list[dict]) -> str:
25    """The one real edge -- waits for every audit node to finish."""
26    response = await client.messages.create(
27        model="claude-opus-4-8",
28        max_tokens=2000,
29        messages=[{
30            "role": "user",
31            "content": f"""Consolidate these {len(results)} route audits
32            into one report grouped by severity:
33
34            {results}"""
35        }]
36    )
37    return response.content[0].text
38
39async def run_graph(route_files: list[str]):
40    # Fan out -- all independent nodes run concurrently
41    audit_tasks = [audit_route_file(f) for f in route_files]
42    results = await asyncio.gather(*audit_tasks)
43
44    # Fan in -- the one node with a real dependency
45    report = await consolidate(results)
46    return report
47
48# 40 route files, one prompt, one parallel pass
49results = asyncio.run(run_graph([
50    f"routes/{f}.py" for f in ["auth", "users", "billing", "orders"]
51    # ...36 more

การเรียก API แบบตามลำดับ 40 ครั้ง ที่ครั้งละ ~8 วินาที รวมกันกว่า 5 นาที การเรียก 40 ครั้งเดียวกันแบบกระจายขนาน: ภายใน 15 วินาที ซึ่งถูกจำกัดด้วยไฟล์ที่ช้าที่สุดไฟล์เดียว ไม่ใช่ผลรวมของทั้งหมด

บทที่ 4 - ที่ที่กราฟพังจริงๆ

วิศวกรรมกราฟล้มเหลวในสามจุดที่คาดเดาได้ รู้ไว้ก่อนที่คุณจะเจอ

Context collapse. กระจาย 1,000 โหนดแล้วพยายามป้อนผลลัพธ์ทั้งหมด 1,000 รายการเข้าสู่ขั้นตอนการรวบรวมหนึ่งขั้นตอน คุณจะเกินหน้าต่างบริบทใดๆ ก่อนที่การสังเคราะห์จะเริ่มต้นด้วยซ้ำ วิธีแก้ไข: วางเลเยอร์ fan-in ของคุณ จัดกลุ่มโหนดเป็นชุดละ 20-50 สรุปแต่ละชุด จากนั้นรวบรวมสรุป - ไม่ใช่ผลลัพธ์ดิบ

1async def layered_consolidate(results: list[dict], batch_size: int = 30):
2    """Fan-in in layers -- never synthesize raw output at scale."""
3    batches = [results[i:i+batch_size]
4               for i in range(0, len(results), batch_size)]
5
6    batch_summaries = await asyncio.gather(*[
7        summarize_batch(batch) for batch in batches
8    ])
9
10    # Final consolidation works on summaries, not 1,000 raw results
11    return await consolidate(batch_summaries)

การพึ่งพาที่เป็นเท็จ คุณจะคิดว่าโหนดสองโหนดเป็นอิสระต่อกันเพราะพรอมต์ของพวกมันไม่ได้อ้างอิงถึงกัน - แต่พวกมันทั้งคู่เขียนไปยังไฟล์เดียวกัน หรือเรียก API ที่มี rate limit ร่วมกัน นั่นคือขอบที่ซ่อนอยู่ วิธีแก้ไข: ตรวจสอบทรัพยากรที่ใช้ร่วมกัน ไม่ใช่แค่ข้อมูลที่ใช้ร่วมกัน โหนดสองโหนดที่มีการขัดแย้งในการเขียนจำเป็นต้องมีขอบ แม้ว่าจะไม่มีการพึ่งพาข้อมูลเลยก็ตาม

ความล้มเหลวของโหนดแบบเงียบ ในสายโซ่ ความล้มเหลวหนึ่งครั้งหยุดทุกอย่าง - น่ารำคาญแต่ชัดเจน ในกราฟ โหนดที่ล้มเหลวหนึ่งโหนดจาก 200 โหนดสามารถหายไปในรายงานที่ดูสมบูรณ์ วิธีแก้ไข: ทุกขั้นตอน fan-in จะตรวจสอบจำนวนโหนดกับจำนวนที่คาดหวังก่อนการสังเคราะห์ และทำเครื่องหมายช่องว่างอย่างชัดเจนแทนที่จะทำงานเงียบๆ กับข้อมูลบางส่วน

1async def safe_consolidate(results: list[dict], expected_count: int):
2    if len(results) < expected_count:
3        missing = expected_count - len(results)
4        print(f"WARNING: {missing} nodes failed silently. "
5              f"Report will be incomplete.")
6    return await consolidate(results)

บทที่ 5 - การขยายขนาดไปสู่ฝูงจริง

เมื่อรูปแบบทำงานได้ที่ 40 โหนด การขยายเป็นหลายร้อยคือการเปลี่ยนแปลงการตั้งค่า ไม่ใช่การออกแบบใหม่ - โดยมีเงื่อนไขว่าคุณสร้างกราฟอย่างถูกต้องตั้งแต่บทที่ 2 เป็นต้นไป

รูปร่างการผลิตเต็มรูปแบบ:

1                    Orchestrator
2                         |
3        +--------+-------+-------+--------+
4        v        v       v       v        v
5     Node 1   Node 2  Node 3  ...      Node N
6   (parallel, no edges between any of them)
7        |        |       |               |
8        +--------+-------+-------+-------+
9                         v
10                 Batch Summary  <- layered fan-in
11                 (groups of 30)
12                         v
13                 Final Report   <- the one true edge

งานเดียวของ orchestrator: แยกย่อยงานเป็นโหนด ระบุขอบจริง และส่งมอบงาน มันไม่ได้ทำงานใดๆ ด้วยตัวเอง - มันวาดกราฟ

1async def orchestrate(task: str, resources: list[str]):
2    """
3    The orchestrator node -- decomposes, doesn't execute.
4    """
5    plan = await client.messages.create(
6        model="claude-opus-4-8",
7        max_tokens=2000,
8        messages=[{
9            "role": "user",
10            "content": f"""Task: {task}
11            Resources available: {resources}
12
13            Decompose into a graph:
14            - List each independent node (no shared edges)
15            - List any real dependencies between nodes
16            - Group nodes into fan-in batches if count > 50
17
18            Return JSON with: nodes, edges, batch_groups"""
19        }]
20    )
21
22    graph = parse_plan(plan.content[0].text)
23
24    # Execute independent nodes in parallel
25    node_results = await asyncio.gather(*[
26        execute_node(n) for n in graph["nodes"] if not n["depends_on"]
27    ])
28
29    # Then execute dependent nodes, respecting real edges only
30    final = await execute_dependent_chain(graph["edges"], node_results)
31
32    return final

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

สิ่งที่เปลี่ยนแปลงเมื่อคุณคิดในรูปแบบกราฟแทนที่จะเป็นเส้นตรง

เอเจนต์เชิงเส้นที่มี 40 ขั้นตอน มีจุดที่ล้มเหลวตามลำดับ 40 จุด และเวลาแฝง 40 เท่าของขั้นตอนที่ช้าที่สุดขั้นตอนเดียว

กราฟที่มีหน่วยงาน 40 หน่วยเดียวกัน มีจุดที่ล้มเหลวแบบขนานมากเท่ากับจำนวนการพึ่งพาจริงที่คุณมี - โดยปกติ 3 ถึง 5 จุดในเวิร์กโฟลว์ส่วนใหญ่ - และเวลาแฝงถูกจำกัดด้วยเลเยอร์ที่ช้าที่สุดของคุณ ไม่ใช่จำนวนขั้นตอนทั้งหมดของคุณ

นั่นไม่ใช่การเพิ่มความเร็วแบบเล็กน้อย มันคือความแตกต่างระหว่างเวิร์กโฟลว์ที่ใช้เวลา 5 นาที กับเวิร์กโฟลว์ที่ใช้เวลา 15 วินาที โดยทำงานพื้นฐานเดียวกันทุกประการ

โมเดลไม่เคยเป็นคอขวด เส้นที่คุณวาดต่างหากที่เป็น

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

ขอบคุณที่อ่านครับ