AI Agents · บทเรียน

โพรโทคอลการสื่อสาร (บัสข้อความ)

แยกเอเจนต์ออกจากกันด้วยบัสข้อความ เช่น Redis, NATS และ Kafka เพื่อให้ขยายระบบและล้มเหลวได้อย่างอิสระ

โพรโทคอลการสื่อสาร (บัสข้อความ) เป็นบทเรียน AI Agents ฟรีบน CoddyKit นี่คือบทเรียนที่ 4 จากทั้งหมด 4 บทเรียน คุณสามารถอ่านบทเรียนทั้งหมดด้านล่างฟรี — จากนั้นลองปฏิบัติด้วยตัวคุณเองในเบราว์เซอร์พร้อมตัวแก้ไขโค้ดในตัวและติวเตอร์ AI ตลอด 24/7 บทเรียนนี้เป็นส่วนหนึ่งของเส้นทางการเรียน AI Agents และความก้าวหน้าของคุณจะซิงค์ข้ามเว็บและแอป CoddyKit คอร์ส AI Agents มีบทเรียนทั้งหมด 4 บทเรียน

บางส่วนของบทเรียนนี้ยังไม่ได้รับการแปล และแสดงเป็นภาษาอังกฤษ

นอกเหนือจากการสนทนาภายในกระบวนการ

สำหรับระบบขนาดเล็ก เอเจนต์จะเรียกหากันภายในกระบวนการ สำหรับระบบขนาดใหญ่ ให้แยกการทำงานออกจากกันผ่านบัสข้อความ เช่น Redis, NATS หรือ Kafka

เหตุใดจึงต้องแยกการทำงานออกจากกัน

สามารถปรับขนาดเอเจนต์แต่ละตัวได้อย่างอิสระ
แยกผลกระทบจากข้อผิดพลาดของเอเจนต์แต่ละตัว
รองรับกระบวนการทำงานแบบอะซิงโครนัส โดยไม่ต้องรอเอเจนต์ที่ทำงานช้าอยู่ภายในกระบวนการ
เล่นข้อความซ้ำเพื่อแก้ไขจุดบกพร่อง
ทำงานข้ามภาษาได้: เอเจนต์ Python และเอเจนต์ Go สามารถสมัครรับข้อความได้ทั้งคู่

Simple Bus: Redis Pub/Sub

import redis
r = redis.Redis()

# Publisher
r.publish('agent.research.task', json.dumps({'task_id': 'abc', 'query': '...'}))

# Subscriber
p = r.pubsub()
p.subscribe('agent.research.task')
for msg in p.listen():
    if msg['type'] == 'message':
        handle_task(json.loads(msg['data']))

การเผยแพร่และสมัครรับข้อมูลเพื่อกระจายข้อความ

ใช้เมื่อเอเจนต์หลายตัวอาจต้องรับทราบเหตุการณ์เดียวกัน เช่น "ได้รับคำถามจากผู้ใช้"

คิวสำหรับการกระจายงาน

ใช้คิว เช่น Redis BLPOP หรือ RabbitMQ เมื่อควรมีผู้ปฏิบัติงานเพียงหนึ่งรายจัดการแต่ละงาน:

import json, queue

r = queue.Queue()

# Producer
r.put(json.dumps({'task_id': 'abc', 'kind': 'research'}))

def process(task):
    print('processing', task)

# Worker
while not r.empty():
    raw = r.get()
    task = json.loads(raw)
    process(task)

NATS เพื่อความเร็ว

NATS คือโบรกเกอร์ข้อความขนาดเล็กที่ออกแบบมาสำหรับไมโครเซอร์วิส มีเวลาแฝงต่ำกว่าหนึ่งมิลลิวินาที และมีการร้องขอ/ตอบกลับในตัว:

import nats
nc = await nats.connect('nats://localhost:4222')

await nc.publish('agent.research', json.dumps(task).encode())

# Request/reply
response = await nc.request('agent.research', payload, timeout=10)

Kafka เพื่อความคงทน

Kafka เพิ่มบันทึกที่คงทนและเล่นซ้ำได้ เหมาะอย่างยิ่งสำหรับการตรวจสอบย้อนหลังและการเล่นซ้ำ:

from confluent_kafka import Producer, Consumer

producer.produce('agent-events', key=task_id, value=json.dumps(task))
producer.flush()

โครงสร้างข้อความเหตุการณ์

กำหนดรูปแบบของข้อความทุกข้อความด้วย Pydantic หรือ Protobuf หากไม่มีโครงสร้าง บัสของคุณจะกลายเป็นระเบียบยุ่งเหยิง:

class ResearchTask(BaseModel):
    task_id: str
    user_id: str
    query: str
    deadline: datetime

class ResearchResult(BaseModel):
    task_id: str
    findings: list[str]
    duration_ms: int

รหัสความสัมพันธ์

ข้อความที่เกี่ยวข้องกันทุกข้อความจะมีรหัสความสัมพันธ์เดียวกัน เพื่อให้คุณติดตามงานข้ามบริการต่าง ๆ ได้:

record = {'correlation_id': 'abc', 'task_id': 'def', 'span': 'llm_call'}
print(record)

ตัวจัดการแบบทำซ้ำได้โดยไม่เกิดผลซ้ำ

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

คิวข้อความที่ส่งไม่สำเร็จ

เมื่อตัวจัดการล้มเหลวซ้ำหลายครั้ง ให้ส่งข้อความไปยัง DLQ เพื่อให้มนุษย์ตรวจสอบ แทนที่จะวนทำงานไม่รู้จบ:

if attempts > MAX_RETRIES:
    r.rpush('queue:research:dlq', raw)
    log.error('Sent to DLQ', extra={'task_id': task_id})

ความสามารถในการสังเกตการณ์

ติดตามข้อความทุกข้อความ:

รหัสการติดตามไหลผ่านบัส
บันทึกประกอบด้วยรหัสงานและรหัสความสัมพันธ์
ตัวชี้วัด: ปริมาณงาน เวลาแฝง และอัตราข้อผิดพลาดแยกตามหัวข้อ

ควรเริ่มใช้เมื่อใด

เริ่มจากวิธีง่าย ๆ ก่อน โดยใช้เอเจนต์ภายในกระบวนการ เปลี่ยนไปใช้บัสเมื่อคุณมีสิ่งต่อไปนี้:

เอเจนต์อย่างน้อย 5 ตัวในระบบใช้งานจริง
จำเป็นต้องปรับขนาดเอเจนต์แต่ละตัวอย่างอิสระ
กระบวนการทำงานที่ครอบคลุมหลายบริการ

การส่งอย่างน้อยหนึ่งครั้ง

การส่งแบบ "อย่างน้อยหนึ่งครั้ง" กำหนดให้ตัวจัดการของคุณต้องทำอะไร

ทบทวน

แยกการทำงานด้วย Redis สำหรับกรณีง่าย ๆ ใช้ NATS เมื่อให้ความสำคัญกับเวลาแฝง และใช้ Kafka เมื่อให้ความสำคัญกับความคงทน ใช้โครงสร้างข้อความ รหัสความสัมพันธ์ ตัวจัดการที่ทำงานซ้ำได้โดยไม่เกิดผลซ้ำ และ DLQ

คำถามที่พบบ่อย

บทเรียน “โพรโทคอลการสื่อสาร (บัสข้อความ)” ฟรีหรือไม่

ใช่ — ข้อความเต็มของ “โพรโทคอลการสื่อสาร (บัสข้อความ)” ฟรีให้อ่านที่นี่บนเว็บ เพื่อปฏิบัติแบบโต้ตอบ (ตัวแก้ไขโค้ดในตัวและติวเตอร์ AI ตลอด 24/7) และปลดล็อคส่วนที่เหลือของคอร์ส AI Agents ให้อัปเกรดเป็น CoddyKit PRO คอร์ส AI Agents มีบทเรียนทั้งหมด 4 บทเรียน

คุณจะเรียนรู้อะไรในบทเรียน “โพรโทคอลการสื่อสาร (บัสข้อความ)”

แยกเอเจนต์ออกจากกันด้วยบัสข้อความ เช่น Redis, NATS และ Kafka เพื่อให้ขยายระบบและล้มเหลวได้อย่างอิสระ คุณปฏิบัติ AI Agents ด้วยโค้ดที่ใช้งานได้จริงที่คุณเรียกใช้โดยตรงในเบราว์เซอร์ และติวเตอร์ AI ตลอด 24/7 ตอบคำถามของคุณขณะที่คุณไปผ่านบทเรียน

คุณต้องมีประสบการณ์ก่อนที่จะเริ่มเรียน AI Agents หรือไม่

ไม่จำเป็นต้องมีประสบการณ์มาก่อน AI Agents บน CoddyKit ออกแบบมาสำหรับผู้เริ่มต้นไปจนถึงผู้เรียนขั้นสูง คุณสามารถเริ่มต้นที่นี่หรือเริ่มจากตัวแรกและเรียนด้วยความเร็วของคุณเอง นี่คือบทเรียนที่ 4 จากทั้งหมด 4 บทเรียน

บทเรียน “โพรโทคอลการสื่อสาร (บัสข้อความ)” ใช้เวลานานแค่ไหน

บทเรียน CoddyKit ส่วนใหญ่ใช้เวลาประมาณ 5–10 นาที แต่ละบทเรียนจึงสั้นและเป็นแบบโต้ตอบ คุณสามารถก้าวหน้าอย่างต่อเนื่องและกลับมาเรียนต่อจากตรงที่เพิ่งหยุดบนเว็บและแอปได้เลย

ฉันเขียนและรันโค้ดในบทเรียน AI Agents นี้ได้ไหม

ได้ บทเรียน AI Agents ทุกบทมีตัวแก้ไขโค้ดในตัว คุณจึงเขียนและรันโค้ดจริงได้เลยในเบราว์เซอร์ และได้รับข้อเสนอแนะจาก AI ในทันที — ไม่ต้องติดตั้งในเครื่องของคุณ

บทเรียนทั้งหมดในหลักสูตรนี้

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

← กลับไปที่ AI Agents