1. บทนำ
Codelab นี้จะแนะนำวิธีสร้างระบบแบบ Agent ยุคถัดไปด้วยเวิร์กโฟลว์และกราฟใน Agent Development Kit (ADK) คุณจะใช้รูปแบบสถาปัตยกรรมทั่วไป จัดการการโต้ตอบที่มีคนคอยตรวจสอบ (HITL) และจัดการการดำเนินการแบบอะซิงโครนัสที่ใช้เวลานาน นอกจากนี้ คุณยังจะผสานรวมฐานความรู้ขององค์กรและหน่วยความจำแบบถาวรเพื่อปรับแต่งและพัฒนาพฤติกรรมของเอเจนต์ได้ด้วย สุดท้าย คุณจะเชื่อมต่อความสามารถเหล่านี้เพื่อขับเคลื่อนไปป์ไลน์การสร้างวิดีโออัตโนมัติ
สถานการณ์
คุณมีช่องดิจิทัลบน VibeTube ที่มีผู้ชมที่ใช้งานอยู่และมีไอเดียสร้างสรรค์ที่รอการเผยแพร่เพิ่มขึ้นเรื่อยๆ การผลิตวิดีโอแต่ละรายการต้องดำเนินการอย่างต่อเนื่องในหลายขั้นตอน ได้แก่ การค้นคว้าหาเทรนด์รูปแบบต่างๆ การสังเคราะห์ความคิดเห็นของผู้ชม การพัฒนาสคริปต์ การตรวจสอบการปฏิบัติตามนโยบาย และการสร้างวิดีโอคลิป โมเดล Generative สามารถร่างชิ้นงานแต่ละรายการได้ แต่การเผยแพร่ผลงานอย่างสม่ำเสมอต้องใช้สถาปัตยกรรมเอเจนต์ที่ประสานงานกัน
หากต้องการทำให้วงจรนี้เป็นอัตโนมัติ คุณจะต้องสร้าง VibeStudio ไปป์ไลน์แบบเอเจนต์นี้จะดำเนินการวิจัยตามปกติแบบคู่ขนาน นำเสนอตัวเลือกที่คัดสรรมาแล้วสำหรับการอนุมัติจากมนุษย์ในลูป ใช้เกตนโยบายอัตโนมัติก่อนสร้างวิดีโอ และรักษาบริบทในการดำเนินการผลิต
สิ่งที่คุณจะได้เรียนรู้
- พื้นฐานด้านวิศวกรรมกราฟ: สถาปัตยกรรมของ Agent แบบหลายขั้นตอนต้องมีโฟลว์การควบคุมที่ชัดเจนและเส้นทางการดำเนินการที่มีโครงสร้าง คุณสร้าง ADK
Workflowโดยใช้ทูเพิลขอบSTARTจุดแรกเข้าJoinNodeสำหรับการรวบรวมแบบ Fan-Out แบบขนาน และโหนดเราเตอร์ที่กำหนดไว้เพื่อควบคุมการดำเนินการตามสถานะ - โหมด Agent และการเรียกกลับวงจร: งานเฉพาะทางต้องมีลักษณะการทำงานที่แตกต่างกันและมีขอบเขตที่กำหนดไว้ คุณกำหนดค่าอินสแตนซ์ ADK
Agentโดยใช้โหมดchat,single_turnและโหมดที่เปิดใช้เครื่องมือtaskเป็นโหนดเวิร์กโฟลว์ โดยใช้ตัวสกัดกั้นกับbefore_model_callbackและafter_agent_callback - การประสานงานแบบมีมนุษย์ร่วม: ไปป์ไลน์การผลิตจะหยุดชั่วคราวเพื่อให้เจ้าหน้าที่ตัดสินใจในจุดตรวจสอบครีเอทีฟโฆษณาที่สำคัญ คุณใช้
RequestInputเพื่อระงับการดำเนินการเวิร์กโฟลว์ บังคับใช้สคีมาการตอบกลับที่มีโครงสร้าง และดำเนินการต่อโดยไม่ต้องรักษากระบวนการรันไทม์ที่ไม่มีการใช้งานให้ทำงานอยู่ - หน่วยความจำของตัวแทนแบบลำดับชั้น: ระบบการผลิตจะแยกสถานะการดำเนินการชั่วคราวออกจากบริบทที่คงทน คุณจัดการสถานะเซสชันระยะสั้นโดยใช้
Event(state=...)และการเชื่อมโยงพารามิเตอร์ รวมถึงเชื่อมต่อ GEAP Memory Bank เพื่อดึงข้อมูล รวบรวม และบันทึกค่ากำหนดของครีเอเตอร์ในการเรียกใช้ต่างๆ - การอ้างอิงจากฐานความรู้ขององค์กร: เอเจนต์อัตโนมัติต้องมีบริบทของโดเมนแบบไดนามิกและความรู้สึกของกลุ่มเป้าหมาย คุณเชื่อมต่อคลังข้อมูลของ GEAP RAG Engine เป็นโหนดการดึงข้อมูลเฉพาะภายใน Fan-out แบบขนานเพื่อยึดเอาต์พุตของ Agent ตามความหมาย
- เวิร์กโฟลว์และการติดตั้งใช้งานที่ใช้เวลานาน: การเรนเดอร์วิดีโอแบบมัลติโมดัลจะทำงานแบบไม่พร้อมกันเป็นระยะเวลานาน คุณจะใช้
LongRunningFunctionToolกับใบเสร็จการโทรที่รอดำเนินการเพื่อระงับและกลับมาทำงานต่อในเวิร์กโฟลว์ตามรหัสการโทร และติดตั้งใช้งานไปป์ไลน์ที่เสร็จสมบูรณ์แล้วโดยใช้ ADKRunnerใน Cloud Run
การจัดระเบียบ Codelab นี้
Codelab นี้เป็นข้อมูลอ้างอิงเชิงแนวคิดและสถาปัตยกรรม แต่ละส่วนจะอธิบายโครงสร้าง ADK ที่ใช้ในขั้นตอนเวิร์กเบนช์ที่เกี่ยวข้อง ให้โค้ดอ้างอิง และกำหนดหลักการออกแบบหลัก โปรดอ่านแต่ละส่วนก่อนทำแบบฝึกหัดที่เกี่ยวข้องในเวิร์กเบนช์
การทำงานจริงจะเกิดขึ้นใน VibeStudio Workbench ซึ่งเป็นอินเทอร์เฟซเว็บที่มาพร้อมกับโปรแกรมแก้ไขโค้ดแบบอินเทอร์แอกทีฟ ตัวตรวจสอบรันไทม์ และเครื่องมือตรวจสอบ ADK แบบฝัง การกำหนดหมายเลขขั้นตอนในเวิร์กเบนช์จะสอดคล้องกับ Codelab นี้โดยตรงเพื่อให้ความคืบหน้าของคุณซิงค์กัน การแก้ไขกราฟพื้นฐานจะยังคงอยู่ตลอดขั้นตอนต่างๆ โดยเวิร์กเบนช์จะตรวจสอบข้อกำหนดเบื้องต้นโดยอัตโนมัติเมื่อคุณดำเนินการต่อ
เมื่อทำแบบฝึกหัดในเวิร์กเบนช์เสร็จแล้ว คุณจะประกอบไปป์ไลน์แบบเอเจนต์ตั้งแต่ต้นจนจบและนำแอปพลิเคชัน VibeStudio ที่ทำงานอยู่ไปใช้งานใน Cloud Run เพื่อสร้างเนื้อหาวิดีโอ
สภาพแวดล้อมประกอบด้วยคอมโพเนนต์หลัก 3 อย่าง ได้แก่ VibeStudio Workbench (อินเทอร์เฟซเว็บในเครื่องสำหรับการแก้ไขโค้ดและการยืนยันรันไทม์), แบ็กเอนด์ (ADK Workflow และแซนด์บ็อกซ์ของสเตจใน agent/) และ Google Cloud (โมเดล Gemini, GEAP Memory Bank, RAG Engine และการสร้างวิดีโอ Veo)
2. ตั้งค่า
รับเครดิตเวิร์กช็อป
หากเข้าร่วมแล็บที่มีผู้สอนเป็นผู้นำ ผู้สอนจะแจกจ่ายเครดิตสำหรับโปรเจ็กต์ Google Cloud ของคุณ ทำตามวิธีการของผู้สอนเพื่อแลกสิทธิ์เครดิตและตรวจสอบว่าการเรียกเก็บเงินในบัญชีของคุณใช้งานได้ก่อนดำเนินการต่อ
เปิด Cloud Shell
Cloud Shell เป็นสภาพแวดล้อมในการพัฒนาซอฟต์แวร์บนเบราว์เซอร์ที่มี gcloud, Python และ git ติดตั้งไว้ล่วงหน้า
วิธีเปิดใช้ Cloud Shell
- ไปที่ คอนโซล Google Cloud
- ในส่วนหัวของการนำทางด้านบน ให้คลิกเปิดใช้งาน Cloud Shell (ไอคอนหน้าต่างเทอร์มินัล)
เซสชันเทอร์มินัลจะเปิดขึ้นที่ด้านล่างของหน้าต่างเบราว์เซอร์
โคลนและเริ่มต้นที่เก็บ
เรียกใช้คำสั่งต่อไปนี้ในเทอร์มินัล Cloud Shell เพื่อโคลนโปรเจ็กต์
git clone https://github.com/gca-americas/vibetube-studio cd ~/vibetube-studio
ข้อความแจ้งการกำหนดค่า
ในระหว่างการตั้งค่า ระบบจะแจ้งให้คุณระบุรายละเอียดต่อไปนี้
- รหัสโปรเจ็กต์ที่อยู่ในระบบคลาวด์ของ Google: เมื่อ
setup_project.shแจ้ง ให้กด Enter เพื่อสร้างโปรเจ็กต์ใหม่โดยอัตโนมัติ หากต้องการใช้โปรเจ็กต์ที่มีอยู่ (เช่น โปรเจ็กต์ที่กำหนดไว้ล่วงหน้า) ให้ป้อนรหัสโปรเจ็กต์และตรวจสอบว่าสะกดถูกต้องและมีการเรียกเก็บเงินที่ใช้งานอยู่ - รหัสกิจกรรม: ป้อนรหัสห้องที่ได้รับจากผู้สอน หากไม่ได้รับ โปรดสอบถามผู้ช่วยสอนหรือเพื่อนบ้าน หากทำแล็บนี้ที่บ้าน ให้กด Enter เพื่อยอมรับห้อง
sandboxเริ่มต้น - ชื่อที่แสดงของช่อง: ป้อนชื่อหรือแฮนเดิลของช่องที่ต้องการเมื่อ
setup_codelab.shแจ้ง หรือกดป้อนเพื่อยอมรับค่าเริ่มต้นที่สร้างจากบัญชี Google
เรียกใช้สคริปต์การตั้งค่า 2 รายการตามลำดับต่อไปนี้
./setup_project.sh ./setup_codelab.sh
setup_project.sh: สร้างหรือนำโปรเจ็กต์ Google Cloud ที่มีการเรียกเก็บเงินที่ใช้งานอยู่กลับมาใช้ใหม่ บันทึกรหัสโปรเจ็กต์ไปยัง~/project_id.txtและกำหนดค่าบริบทgcloudที่ใช้งานอยู่setup_codelab.sh: ติดตั้งuvและการอ้างอิง Python ลงใน.venvเปิดใช้ Google Cloud APIs ที่จำเป็น กำหนดค่าการตั้งค่าแชแนลใน.envยืนยันการเข้าถึงโมเดลด้วย Gemini จัดสรรทรัพยากร Memory Bank และ RAG สร้างอินเทอร์เฟซ Workbench และเริ่ม VibeStudio Workbench
สคริปต์จะเรียกใช้การตรวจสอบก่อนส่งและเริ่ม VibeStudio Workbench ในเบื้องหลัง บรรทัดสุดท้ายจะแสดงลิงก์สำหรับเปิด
7 · Preflight
✓ python 3.12
✓ auth path A: Vertex via ADC (STUDIO_VERTEX=1)
✓ Google Cloud ADC (project <your-project>)
✓ stage0_prompt loads
...
✓ stage6_video loads (13 edges)
✓ aiplatform.googleapis.com enabled (Gemini, Veo, Memory Bank, RAG Engine)
✓ vectorsearch.googleapis.com enabled (the vector store a RAG corpus is built on)
✓ Memory Bank connected
✓ RAG corpus connected
✓ VibeStudio Workbench running on port 4600
PREFLIGHT GREEN
Setup finished. The VibeStudio Workbench is already running.
Open this and start at step 1
https://4600-<your cloud shell host>/step/story
It runs in the background. You do not need to start anything else.
log runs/lab.log
stop kill $(cat runs/lab.pid)
start scripts/start.sh
คลิกลิงก์นั้น คุณจะดูที่อยู่เดียวกันได้ในส่วนตัวอย่างเว็บ → เปลี่ยนพอร์ต → 4600
หากต้องการตรวจสอบสภาพแวดล้อมอีกครั้งเมื่อใดก็ตาม ให้เรียกใช้ python scripts/preflight.py หากต้องการรีสตาร์ทเวิร์กเบนช์ ให้เรียกใช้ scripts/restart.sh หากต้องการตั้งค่าอีกครั้ง ให้เรียกใช้ ./setup_codelab.sh ซึ่งจะเก็บการกำหนดค่าและความคืบหน้าของคุณไว้
เมื่อเปิดแล้ว ให้อ่านขั้นตอนที่ 1 เรื่องราวสำหรับสถานการณ์ และขั้นตอนที่ 2 สิ่งที่คุณสร้างสำหรับรูปร่างของกราฟที่เสร็จสมบูรณ์ ทั้งสองไม่มีการออกกำลังกาย จากนั้นกลับมาที่นี่เพื่อทำขั้นตอนที่ 3
ทุกส่วนที่ต้องลงมือปฏิบัติจริงของ VibeStudio Workbench จะสิ้นสุดด้วยแผงการยืนยันที่อ่านอาร์ติแฟกต์จริง ได้แก่ ไฟล์ในดิสก์และเซสชันที่เขียนโดยการเรียกใช้
เลย์เอาต์ของที่เก็บ
ที่เก็บมีโครงสร้างเป็นตรรกะเวิร์กโฟลว์หลัก แซนด์บ็อกซ์แบบทีละขั้นตอน สภาพแวดล้อมของเวิร์กเบนช์ และแอปพลิเคชันเวอร์ชันที่ใช้งานจริง
vibe-studio-lab/
├── agent/ # Core ADK workflow, graph definition, and platform services
│ ├── graph.py # Workflow graph definition, node functions, and routers
│ ├── desk.py # Video render desk using LongRunningFunctionTool
│ ├── schemas.py # Pydantic schemas for directions, gates, and scripts
│ ├── trends.py # Trend generation and sampling utilities
│ ├── backlog.txt # Creator video ideas backlog
│ ├── comments.md # Audience comments for RAG Engine corpus seeding
│ ├── policy_words.txt # Blocked subject words for deterministic policy checks
│ └── platform/ # Google Cloud service clients (Memory Bank, RAG, Veo)
│ ├── config.py # Environment variables, locations, and model configurations
│ ├── memory.py # GEAP Memory Bank callbacks and context injection
│ ├── rag.py # GEAP RAG Engine corpus creation and semantic retrieval
│ └── videogen.py # Veo video generation and operation polling
├── stage0_prompt/ # Step sandboxes: isolated agent.py files runnable in adk web
│ └── ... # stage1_fanout through stage6_video for incremental steps
├── server/ & web/ # VibeStudio Workbench (FastAPI backend and React frontend)
├── vibestudio/ # Complete production application deployed to Cloud Run
│ ├── server/ # FastAPI production server and event runner
│ ├── web/ # End-user React web application
agent/: มีกราฟเวิร์กโฟลว์หลัก คุณจะแก้ไขไฟล์ในไดเรกทอรีนี้เพื่อใช้โหนดแบบ Fan-Out แบบขนาน การกำหนดเส้นทางนโยบายที่แน่นอน การเรียกกลับของหน่วยความจำ และเครื่องมือสร้างวิดีโอagent/platform/: เชื่อมต่อกับบริการของ Google Cloud ซึ่งรวมถึงโมเดล Gemini, GEAP Memory Bank, GEAP RAG Engine และการสังเคราะห์วิดีโอ Veostage0_prompt/ถึงstage6_video/: สภาพแวดล้อมแซนด์บ็อกซ์แบบแยก แต่ละโฟลเดอร์จะส่งออกroot_agentแบบสแตนด์อโลนเพื่อให้คุณเรียกใช้และตรวจสอบแต่ละขั้นตอนแยกกันได้ผ่านอินเทอร์เฟซการพัฒนา ADK แบบฝังserver/และweb/: แอปพลิเคชัน VibeStudio Workbench ที่ทำงานในเครื่องบนพอร์ต 4600 ซึ่งเป็นที่ตั้งของเอกสารประกอบแบบทีละขั้นตอน ตัวแก้ไขโค้ดในหน้าเว็บ เครื่องมือยืนยันหลักฐานรันไทม์ และการแสดงภาพกราฟvibestudio/: แอปพลิเคชันเวอร์ชันที่ใช้งานจริงที่สมบูรณ์ซึ่งแพ็กเกจและติดตั้งใช้งานใน Cloud Run ในขั้นตอนสุดท้าย โดยจะมีสำเนาแบบสแตนด์อโลนของกราฟเวิร์กโฟลว์ที่เสร็จสมบูรณ์แล้ว
3. เอเจนต์แบบโมโนลิธ
ก่อนสร้างกราฟเวิร์กโฟลว์แบบหลายโหนด คุณต้องสร้างพื้นฐานสถาปัตยกรรมด้วยเอเจนต์เดียวใน stage0_prompt/agent.py เอเจนต์นี้ใช้พรอมต์ของระบบแบบ Monolithic ที่อธิบายไปป์ไลน์การผลิตในรูปแบบร้อยแก้ว ซึ่งได้รับการสนับสนุนจากเครื่องมือฟังก์ชัน Python 2 รายการ
การประเมินพื้นฐานนี้แสดงให้เห็นขอบเขตการดำเนินงานของการประสานงานที่ขับเคลื่อนด้วยพรอมต์ และอธิบายว่าเหตุใดระบบเวอร์ชันที่ใช้งานจริงจึงต้องมีการจัดการเป็นกลุ่มกราฟ
สถาปัตยกรรมของ ADK Agent (3A)
ใน VibeStudio Workbench ให้ไปที่ขั้นตอนที่ 3 · Monolithic agent แล้วเปิดสถาปัตยกรรมของ Agent ADK (3A) มุมมองนี้แสดงเลเยอร์สถาปัตยกรรมหลักของเอเจนต์ ADK (LlmAgent)
from google.adk.agents import LlmAgent
from google.adk.tools import mcp_toolset
root_agent = LlmAgent(
model="gemini-3.5-flash", # model
instruction=BRAND_INSTRUCTION, # instruction
skills=[load_skill("brand-audit")], # skills
tools=[mcp_toolset("mcp_brand_style")], # tools
output_schema=BrandStyleReport, # structured output
before_agent_callback=setup_ctx, # interceptor
before_model_callback=require_image, # interceptor
after_model_callback=schema_guard, # interceptor
)
แผนภาพแบบอินเทอร์แอกทีฟจะจัดกลุ่มคอมโพเนนต์ของเอเจนต์เป็น 5 โดเมนการทำงาน ดังนี้
- เลเยอร์การให้เหตุผล (โมเดล): โมเดลภาษาหลัก (เช่น Gemini 3 Flash) ที่ดำเนินการงานด้านการรับรู้ การให้เหตุผลตามพรอมต์ และการเลือกเครื่องมือ ส่วนอื่นๆ ทั้งหมดในสถาปัตยกรรมจะให้ข้อมูลหรือจำกัดโมเดลนี้
- เลเยอร์บริบท (คำสั่งและทักษะ): คำสั่งที่กำหนดการให้เหตุผลของโมเดล
instructionสร้างพรอมต์ของระบบ บุคลิก และกฎการปฏิบัติงานถาวรskillsให้คำแนะนำเกี่ยวกับขั้นตอนการทำงานแบบเวอร์ชัน (SKILL.md) สำหรับเวิร์กโฟลว์ที่ทำซ้ำได้ - เลเยอร์การทำงานร่วมกันและการดำเนินการ (เครื่องมือ, Subagent, เวิร์กโฟลว์, สคีมาเอาต์พุต): อินเทอร์เฟซที่ช่วยให้เอเจนต์ดำเนินการในระบบภายนอกและส่งข้อมูลที่พิมพ์
toolsระบุฟังก์ชัน Python ที่เรียกใช้ได้หรือปลายทาง Model Context Protocol (MCP)subagentsดำเนินการงานที่ได้รับมอบหมายจากผู้ใต้บังคับบัญชาworkflowประสานงานกราฟแบบหลายเอเจนต์output_schemaใช้โมเดล Pydantic เพื่อรับประกันว่าผู้ใช้ปลายทางจะได้รับ JSON ที่ผ่านการตรวจสอบแล้วแทนที่จะเป็นข้อความที่ไม่มีโครงสร้าง - เลเยอร์ Interceptor (การเรียกกลับของวงจร): การ์ดเรลแบบดีเทอร์มินิสติกที่เรียกใช้โค้ดที่กำหนดเองก่อนและหลังการดำเนินการของเอเจนต์ (
before_agent/after_agent) เทิร์นของโมเดลแต่ละรายการ (before_model/after_model) และการเรียกใช้เครื่องมือ (before_tool/after_tool) Interceptor จะบังคับใช้กฎนโยบายโดยไม่ต้องอาศัยการปฏิบัติตามข้อกำหนดของโมเดล - สถานะภายนอก (เซสชันและหน่วยความจำ): การคงสถานะแบบมีสถานะที่แยกออกจากตรรกะของเอเจนต์
Sessionจะเก็บหน่วยความจำที่ใช้ชั่วคราวและร่องรอยเหตุการณ์สำหรับเธรดการดำเนินการปัจจุบันMemoryจะรักษาข้อเท็จจริงและค่ากำหนดข้ามเซสชันที่คงทนโดยใช้บริการที่มีการจัดการ เช่น GEAP Memory Bank
ในขั้นตอนนี้ เอเจนต์แบบ Monolithic จะใช้เฉพาะ Primitive 3 รายการ ได้แก่ model, instruction และ tools ขั้นตอนถัดไปจะแนะนำเวิร์กโฟลว์กราฟ สคีมาที่มีโครงสร้าง ตัวสกัดกั้น และบริการหน่วยความจำแบบถาวร
ข้อกำหนดของเอเจนต์แบบ Monolithic (3B)
ในเวิร์กเบนช์ ให้ไปที่ข้อกำหนดของเอเจนต์แบบ Monolithic (3B) เปิด stage0_prompt/agent.py เพื่อตรวจสอบคำจำกัดความของ Agent พื้นฐาน
- คำสั่งพรอมต์เดียว: พรอมต์ระบบจะสรุปงานการผลิตที่แตกต่างกัน 5 อย่างให้เป็นร้อยแก้วที่ต่อเนื่อง ได้แก่ การค้นหาเทรนด์ของแพลตฟอร์ม การตรวจสอบไอเดียที่ค้างไว้ การเสนอแนวคิดครีเอทีฟโฆษณา การบังคับใช้นโยบายเรื่องที่ไม่อนุญาต และการร่างรายการช็อต
- แหล่งข้อมูลพื้นฐาน: เอเจนต์อ้างอิงแหล่งข้อมูล 2 แหล่งที่กำหนดไว้ข้างกราฟ ได้แก่
agent/trends.py: สุ่มตัวอย่างเทรนด์รูปแบบและสไตล์ที่กำลังมาแรง 10 รายการจากกลุ่มตัวอย่าง 250 รายการที่มีคะแนนความนิยมแบบไดนามิกagent/backlog.txt: อ่านบันทึกแนวคิดฉบับร่างของครีเอเตอร์ทีละบรรทัด
เครื่องมือใน Agent (3C)
ในเวิร์กเบนช์ ให้ไปที่เครื่องมือใน Agent (3C)
เครื่องมือสำหรับตัวแทนคืออะไร
โมเดลภาษาเป็นเครื่องมือให้เหตุผลแบบโลกปิดโดยธรรมชาติ ซึ่งทำงานเฉพาะกับน้ำหนักที่ผ่านการฝึกมาก่อนและโทเค็นที่อยู่ในหน้าต่างบริบทโดยตรง โดยจะค้นหาฐานข้อมูล เข้าถึง API แบบเรียลไทม์ หรือเรียกใช้โค้ดโดยตรงไม่ได้
เครื่องมือจะช่วยเชื่อมต่อขอบเขตนี้ โดยจะให้สิทธิ์เอเจนซีภายนอกแก่โมเดล ซึ่งช่วยให้โมเดลดึงข้อมูลความจริงพื้นฐานและดำเนินการที่กำหนดไว้ล่วงหน้าในระบบภายนอกได้
การเรียกใช้เครื่องมือเป็นไปตามโปรโตคอล 5 ขั้นตอนที่ชัดเจนระหว่างโมเดลกับรันไทม์ ADK ดังนี้
- การประกาศสคีมา: นักพัฒนาแอปจะให้ฟังก์ชัน Python แก่เอเจนต์ ADK จะตรวจสอบชื่อของแต่ละฟังก์ชัน คำอธิบายประกอบประเภท และสตริงเอกสารเพื่อสร้างการประกาศสคีมา JSON ที่เข้ากันได้กับ OpenAPI ซึ่งอธิบายพารามิเตอร์และวัตถุประสงค์ของฟังก์ชัน
- การให้เหตุผลของโมเดล: ในระหว่างการอนุมาน โมเดลจะประเมินว่าพรอมต์ของผู้ใช้ต้องใช้ข้อมูลภายนอกหรือไม่ หากจำเป็น โมเดลจะปล่อยเหตุการณ์ที่มีโครงสร้าง
function_callซึ่งมีชื่อฟังก์ชันเป้าหมายและพจนานุกรมอาร์กิวเมนต์ที่ตรงกับสคีมา - การดำเนินการรันไทม์: โมเดลเองไม่ได้เรียกใช้โค้ด รันไทม์ของ ADK จะสกัดกั้น
function_callเรียกใช้ฟังก์ชัน Python ในเครื่องจริงโดยใช้อาร์กิวเมนต์ที่ระบุ และบันทึกค่าที่ส่งคืน - การแทรกบริบทอีกครั้ง: รันไทม์ ADK จะแพ็กเกจค่าที่ฟังก์ชันส่งคืนเป็นเหตุการณ์
function_responseและต่อท้ายค่าดังกล่าวในประวัติเซสชันที่ใช้งานอยู่ - การสังเคราะห์ขั้นสุดท้าย: โมเดลจะประมวลผลเอาต์พุตของเครื่องมือที่อยู่ในหน้าต่างบริบทและสร้างคำตอบให้เสร็จสมบูรณ์
ใน stage0_prompt/agent.py เครื่องมือวิจัย 2 อย่างจะกำหนดเป็นฟังก์ชัน Python มาตรฐาน ดังนี้
def check_trends() -> dict:
"""Ten formats trending on the platform right now, with a heat score each."""
from agent.trends import sample_trends
return {"trends": sample_trends()}
def read_backlog() -> dict:
"""The creator's backlog: ideas they noted down to make someday."""
from agent.graph import backlog_notes
return {"backlog": backlog_notes()}
การแก้ไขและการดำเนินการด้วยตนเอง
ในตัวแก้ไขโค้ดของเวิร์กเบนช์ ให้เพิ่มการอ้างอิงฟังก์ชัน 2 รายการลงในรายการ tools ของเอเจนต์
tools=[check_trends, read_backlog],
บันทึกการเปลี่ยนแปลง ไฟล์จะอัปเดตในดิสก์ และแถวการยืนยันจะยืนยันว่าเครื่องมือทั้ง 2 เชื่อมต่อกัน
คลิกเปิดเว็บ ADK เพื่อเปิดอินเทอร์เฟซการพัฒนา ADK แบบฝัง ส่งพรอมต์ไอเดียที่แนะนำ
tonight's idea: a tiny robot doing laundry at midnight
สิ่งที่จะเกิดขึ้นและเหตุผล
เมื่อส่งพรอมต์นี้ ให้สังเกตลำดับการดำเนินการต่อไปนี้ในร่องรอยเซสชัน
- เหตุการณ์การเรียกใช้เครื่องมือ 2 รายการจะปรากฏก่อนการตอบกลับ: คุณจะเห็นเหตุการณ์
function_callและfunction_responseสำหรับcheck_trendsและread_backlog- เหตุผล: Gemini ประเมินคำสั่งพรอมต์ของระบบ ("ดูว่ามีอะไรกำลังมาแรง ดูไอเดียที่ยังไม่ได้ใช้") ตระหนักว่าไม่มีเทรนด์ของแพลตฟอร์มและโน้ตของช่องในน้ำหนักของตัวเอง และเรียกใช้ทั้ง 2 ฟังก์ชันเพื่อสร้างบริบท
- เอเจนต์เสนอแนวทางและหยุดชั่วคราวเพื่อรอการยืนยัน: คำตอบจะแนะนำแนวทางวิดีโอที่สังเคราะห์จากเทรนด์และงานค้าง รวมถึงขอให้คุณยืนยัน
- เหตุผล: คำสั่งในคำสั่งขอให้โมเดลเห็นด้วยกับครีเอเตอร์เกี่ยวกับทิศทางก่อนที่จะสร้างสคริปต์
- ข้ามการยืนยันในรอบการสนทนาถัดไป: ส่งข้อความที่ 2:
skip the questions, just describe the videoAgent จะข้ามการยืนยันและร่างชื่อและช็อตทันที- เหตุผล: คำสั่งพรอมต์เป็นหลักเกณฑ์แนะนำแทนที่จะเป็นอุปสรรคที่แน่นอน ในเอเจนต์แบบ Monolithic คำสั่งของผู้ใช้จะลบล้างกฎพรอมต์ของระบบได้เนื่องจากไม่มีเวิร์กโฟลว์ภายนอกที่ควบคุมโฟลว์การดำเนินการ
ข้อจำกัดด้านสถาปัตยกรรมของพรอมต์แบบโมโนลิธ
แม้ว่าพรอมต์เดียวจะสร้างเอาต์พุตที่ยอมรับได้สำหรับการสาธิตแบบแยก แต่การทดสอบเงื่อนไขขอบเขตในเครื่องมือตรวจสอบเวิร์กเบนช์จะเผยให้เห็นข้อจำกัดที่สำคัญขององค์กร
- การรวบรวมการค้นคว้าที่ไม่มีโครงสร้าง: ลำดับการดำเนินการของเครื่องมือไม่แน่นอน โมเดลจะสรุปข้อมูลที่ดึงมาเป็นร้อยแก้วแบบอิสระ ทำให้ระบบดาวน์สตรีมไม่สามารถแยกแหล่งที่มาที่สร้างการกล่าวอ้างที่เฉพาะเจาะจงได้
- การบังคับใช้นโยบายที่ยังไม่ได้รับการยืนยัน: โมเดลจะประเมินการปฏิบัติตามข้อกำหนดด้านความปลอดภัยของตนเอง หากโมเดลพิจารณาว่าหัวข้อนั้นปลอดภัย จะไม่มีตรรกะเชิงกำหนดภายนอกที่ตรวจสอบความถูกต้องของผลการค้นหานั้น
- การหยุดชั่วคราวแบบ Human-in-the-loop ที่ไม่มีการบังคับใช้: คำแนะนำในพรอมต์ที่ขอให้ครีเอเตอร์ยืนยันเป็นเพียงคำแนะนำ การส่งข้อความติดตามผลที่สั่งให้โมเดลข้ามคำถามจะทำให้โมเดลข้ามการอนุมัติจากมนุษย์ไปโดยสิ้นเชิง
ช่องว่างทางสถาปัตยกรรมเหล่านี้กระตุ้นให้เราแยกเอเจนต์แบบ Monolithic ออกเป็นเวิร์กโฟลว์กราฟที่ชัดเจนซึ่งสร้างขึ้นในขั้นตอนถัดไป
4. พื้นฐานของเวิร์กโฟลว์ Agentic AI
ใน VibeStudio Workbench ให้ไปที่ขั้นตอนที่ 4 · พื้นฐานของเวิร์กโฟลว์แบบ Agentic ส่วนที่ 4A ถึง 4D
ขั้นตอนนี้จะเปลี่ยนจากพื้นฐานแบบเอเจนต์เดียวเป็นการจัดระเบียบกราฟเชิงกำหนดโดยใช้ ADK Workflow คุณจะสร้างการแยกย่อยการวิจัยแบบขนาน ซิงค์กิ่งก้านด้วยโหนด Join สร้างครีเอทีฟโฆษณาที่ผ่านการตรวจสอบสคีมา และเปิดตัวเกตการอนุมัติแบบ Human-in-the-Loop ที่กำหนดได้
สถาปัตยกรรมกราฟและห่วงโซ่การดำเนินการ (4A)
ในเวิร์กเบนช์ ให้เปิดสถาปัตยกรรมกราฟและห่วงโซ่การดำเนินการ (4A)
ADK Workflow จัดโครงสร้างการดำเนินการของ Agent เป็นกราฟแบบมีทิศทางที่กำหนดโดยรายการขอบ
- เชน: ทูเพิลตามลําดับจะกําหนดการดำเนินการโหนดเชิงเส้น (
(node_a, node_b, node_c)) - กิ่งก้านแบบขนาน: เชนอิสระที่แชร์โหนดต้นทางจะดำเนินการพร้อมกัน
- การซิงโครไนซ์: เชนที่บรรจบกันใน
JoinNodeจะรอจนกว่ากิ่งก้านทั้งหมดที่เข้ามาจะรายงานก่อนจึงจะเผยแพร่ - การควบคุมแบบดีเทอร์มินิสติก: โครงสร้างโค้ดที่ประกาศจะควบคุมโฟลว์การดำเนินการแทนที่จะอนุมานจากข้อความพรอมต์
รูปแบบโหนดใน ADK
เวิร์กโฟลว์ ADK ประกอบด้วยโหนดเฉพาะทางหลายประเภท อาร์คีไทป์แต่ละรายการมีบทบาทด้านการดำเนินงานที่เฉพาะเจาะจงในกราฟ โดยแยกการเรียกใช้โค้ดที่แน่นอนจากการให้เหตุผลของโมเดล Generative ดังนี้
อาร์คีไทป์ของโหนด | การใช้งาน | บทบาทในไปป์ไลน์ |
โหนดฟังก์ชัน | ฟังก์ชัน Python ที่แสดงผล | เรียกใช้ตรรกะที่แน่นอน การดึงข้อมูล และการเปลี่ยนแปลงสถานะ |
โหนดเข้าร่วม | อินสแตนซ์ | ซิงค์กิ่งก้านที่เกิดขึ้นพร้อมกันเป็นพจนานุกรมแบบรวม |
โหนด Agent |
| ประเมินวิธีการเทียบกับอินพุตต้นทางและส่งออกข้อมูลที่ตรวจสอบแล้ว |
โหนดเราเตอร์ | ฟังก์ชันที่แสดงผล | ประเมินตรรกะแบบมีเงื่อนไขเพื่อเลือกกิ่งก้านการดำเนินการที่อยู่ปลายน้ำ |
โหนดอินพุตจากมนุษย์ | ฟังก์ชันที่ให้ผลลัพธ์ | ระงับสถานะการดำเนินการจนกว่าจะได้รับการตอบกลับจากผู้ใช้ภายนอก |
root_agent = Workflow(
name="stage1_fanout",
description="2 real readers -> join -> one research dict",
edges=[...])
ในการกำหนดค่านี้ root_agent เป็นอินสแตนซ์ของ Workflow แทนที่จะเป็น Agent แบบสแตนด์อโลน ADK ถือว่าเวิร์กโฟลว์เป็นเอเจนต์ระดับเฟิร์สคลาส ซึ่งช่วยให้โหลด แสดง และตรวจสอบทั้งกราฟเป็นแอปพลิเคชันแบบรวมได้ name ลงทะเบียนแอปพลิเคชันใน ADK Web ขณะที่รายการ edges จะกำหนดโทโพโลยีการดำเนินการ
การวิจัยแบบขนาน แฟนเอาต์ (Fan-Out) (4B)
ในเวิร์กเบนช์ ให้ไปที่แฟนเอาต์การวิจัยแบบขนาน (4B) เปิด stage1_fanout/agent.py
โหนดฟังก์ชันและอุปสรรคในการซิงโครไนซ์
ระยะการวิจัยใช้โหนดฟังก์ชัน 2 รายการที่นำเข้าจาก agent/graph.py ดังนี้
scan_trends: แสดงผลEvent(output={"trends": [...]})ที่มีเทรนด์ของแพลตฟอร์ม 10 รายการที่ได้คะแนนread_backlog: การตอบกลับEvent(output={"backlog": [...], "idea": "..."})ที่มีไอเดียสำหรับเนื้อหาที่ค้างไว้ในช่อง 15 รายการพร้อมกับพรอมต์การรันครั้งแรก
แต่ละฟังก์ชันจะรับ node_input (เอาต์พุตของโหนดก่อนหน้า) และแสดงผล Event
JoinNode ทำหน้าที่เป็นอุปสรรคในการซิงโครไนซ์ โดยจะหยุดชั่วคราวจนกว่าทุกเชนขาเข้าจะส่งเหตุการณ์ จากนั้นจะรวบรวมผลลัพธ์ของทุกสาขาเป็นพจนานุกรมที่มีคีย์เป็นชื่อโหนด ({"scan_trends": {...}, "read_backlog": {...}})
การแก้ไขภาคปฏิบัติ: การกำหนดขอบที่เชื่อมต่อและขอบคู่ขนาน
ใน stage1_fanout/agent.py ให้สร้างอินสแตนซ์ของ JoinNode และเชื่อมต่อเชน 2 รายการแบบขนานโดยเริ่มจาก START
join_research = JoinNode(name="join_research")
edges=[(START, scan_trends, join_research),
(START, read_backlog, join_research)])
บันทึกการเปลี่ยนแปลง เครื่องมือตรวจสอบเวิร์กเบนช์จะยืนยันว่าการเชื่อมต่อและขอบมีการเชื่อมต่อ เรียกใช้สเตจโดยใช้เรียกใช้สเตจ 1 หรือผ่านอินเทอร์เฟซเว็บ ADK ที่ฝังไว้
สิ่งที่จะเกิดขึ้นและเหตุผล
- การดำเนินการของผู้อ่านพร้อมกัน: ในกราฟการดำเนินการ
scan_trendsและread_backlogจะดำเนินการพร้อมกัน- เหตุผล: ทั้ง 2 เครือข่ายมีต้นทางอยู่ที่
STARTเครื่องมือ ADK จะกำหนดเวลาให้สาขาอิสระทำงานพร้อมกัน
- เหตุผล: ทั้ง 2 เครือข่ายมีต้นทางอยู่ที่
- เอาต์พุตพจนานุกรมแบบรวม: เวิร์กโฟลว์จะเสร็จสมบูรณ์ที่
join_researchโดยจะแสดงเอาต์พุตเป็นพจนานุกรมที่มีรายการสำหรับทั้งผู้อ่าน- เหตุผล:
JoinNodeช่วยให้มั่นใจได้ว่าระบบจะบันทึกข้อมูลทั้งหมดก่อนที่จะอนุญาตให้โหนดที่ตามมาดำเนินการ
- เหตุผล:
โหนด Agent (4C)
ในเวิร์กเบนช์ ให้ไปที่โหนด Agent (4C) เปิด stage2_direction/agent.py
โหมดการทำงานและสคีมาที่มีโครงสร้าง
เมื่อฝังไว้ภายใน Workflow Agent จะทํางานในโหมด single_turn โดยค่าเริ่มต้น
- โดยจะรับเอาต์พุตของโหนดก่อนหน้าเป็นอินพุตบริบท
- โดยจะดำเนินการเรียกใช้การอนุมานครั้งเดียวโดยไม่มีการสนทนาไปมา
- โดยจะส่งออก Structured Data ไปยังโหนดถัดไป
การกำหนด output_schema=Directions จะทำให้เอเจนต์บังคับใช้การตรวจสอบ Pydantic กับเอาต์พุตโมเดล กราฟดาวน์สตรีมจะได้รับออบเจ็กต์ที่มีการพิมพ์แทนร้อยแก้วที่ไม่มีโครงสร้าง
class Direction(BaseModel):
title: str # <=60 chars, filmable, characterful
angle: str # the twist, one line
hook: str = "" # 2-4 words, the video's sticker line
evidence: list[Evidence]
class Directions(BaseModel):
candidates: list[Direction] # exactly 4
PROPOSE_INSTRUCTION จะสั่งให้โมเดลเสนอผู้สมัคร 4 คนโดยอ้างอิงหลักฐานจากทั้งเทรนด์และแบ็คล็อก ผู้สมัคร 1-3 เสนอแนวคิดช่องที่ใช้ได้ ผู้สมัครคนที่ 4 จงใจนำเสนอแนวคิดที่ละเมิดนโยบายเพื่อทดสอบเกตความปลอดภัยในขั้นตอนถัดไป
การแก้ไขภาคปฏิบัติ: การกำหนดโหนด Agent และการเชื่อมโยงการเข้าร่วม
ใน stage2_direction/agent.py ให้กำหนดค่า propose_directions และขยายขอบเวิร์กโฟลว์
propose_directions = Agent(
name="propose_directions",
model=config.MODEL,
instruction=PROPOSE_INSTRUCTION,
output_schema=Directions)
edges=[(START, scan_trends, join_research),
(START, read_backlog, join_research),
(join_research, propose_directions, direction_gate)])
สิ่งที่จะเกิดขึ้นและเหตุผล
- การใช้พจนานุกรมโดยตรง:
propose_directionsใช้เพย์โหลด JSON ที่join_researchปล่อยออกมาโดยไม่ต้องจัดรูปแบบด้วยตนเอง - เอาต์พุตของผู้สมัครที่พิมพ์: เอเจนต์จะปล่อยออบเจ็กต์
Directionsที่ตรวจสอบแล้วซึ่งมีผู้สมัครที่แตกต่างกัน 4 ราย โหนดปลายทางจะอ่านฟิลด์ตามชื่อแอตทริบิวต์ (candidate.title) โดยไม่ต้องแยกวิเคราะห์สตริง
โมเดลแบบมีมนุษย์ร่วม (4D)
ในเวิร์กเบนช์ ให้ไปที่โมเดลแบบมีมนุษย์ร่วม (4D) เปิด agent/graph.py
คำสั่งพรอมต์กับการระงับที่กำหนด
เวิร์กโฟลว์การผลิตที่ทำให้เกิดต้นทุนทางการเงินหรือเผยแพร่เนื้อหาต้องมีการกำกับดูแลจากเจ้าหน้าที่ในจุดตัดสินใจที่สำคัญ ในพรอมต์เดียว คำขอการยืนยันเป็นคำแนะนำที่ผู้ใช้สามารถพรอมต์โมเดลให้ข้ามได้อย่างง่ายดาย ในเวิร์กโฟลว์ ADK เครื่องมือดำเนินการจะบังคับใช้การอนุมัติจากเจ้าหน้าที่ โดยกราฟจะหยุดที่โหนดที่กำหนดและจะดำเนินการต่อไม่ได้จนกว่าจะได้รับอินพุตภายนอกที่ผ่านการตรวจสอบสคีมา
- การส่งคืนค่าจะ
RequestInputระงับการดำเนินการเวิร์กโฟลว์ทันที - ADK จะบันทึกการเรียกใช้การขัดจังหวะที่เปิดอยู่ไว้ในที่เก็บเซสชันและออก
interrupt_idที่ไม่ซ้ำกัน - กระบวนการดำเนินการจะหยุดโดยไม่ใช้โทเค็นหรือเธรดของเซิร์ฟเวอร์
- Graph Execution จะกลับมาทำงานอีกครั้งเมื่อมีการส่ง
function_responseที่ถูกต้องซึ่งตรงกับสคีมาและรหัสการขัดจังหวะ
การแก้ไขด้วยตนเอง: การระงับการดำเนินการด้วย RequestInput
ใน agent/graph.py ให้ใช้การเรียกการระงับภายใน direction_gate ดังนี้
yield RequestInput(
message="Pick tonight's direction: 1, 2, 3 or 4.",
response_schema={
"type": "object",
"properties": {
"pick": {"type": "string", "enum": ["1", "2", "3", "4"]}}},
payload={"candidates": cands})
RequestInput กำหนดค่าแอตทริบิวต์ 3 รายการดังนี้
message: พรอมต์รีวิวที่แสดงต่อผู้ใช้response_schema: สคีมา JSON ที่ส่วนหน้าแสดงเป็นแบบฟอร์มอินพุต ซึ่ง ADK จะตรวจสอบเมื่อส่งpayload: ข้อมูลเมตาที่รวมอยู่ในคำขอ (ผู้สมัคร 4 คน) ช่วยให้อินเทอร์เฟซไคลเอ็นต์แสดงการ์ดรีวิวได้โดยไม่ต้องค้นหาสถานะเซสชัน
สิ่งที่จะเกิดขึ้นและเหตุผล
- เวิร์กโฟลว์หยุดที่ direction_gate: ใน ADK Web หรืออินเทอร์เฟซ Workbench การเรียกใช้จะหยุดชั่วคราวและแสดงแบบฟอร์มการเลือกผู้สมัครแบบโต้ตอบ
- เหตุผล: เครื่องมือพบ
RequestInputและคงสถานะการดำเนินการไว้ในruns/sessions.db
- เหตุผล: เครื่องมือพบ
- การกลับมาทำงานต่อต้องใช้ข้อมูลที่มีโครงสร้าง: การส่งข้อความแชทโดยพลการจะไม่ทำให้กราฟคืบหน้า การเลือกตัวเลือก (1, 2, 3 หรือ 4) จะส่ง
function_responseที่พิมพ์ซึ่งตรงตามresponse_schemaและดำเนินการต่อ
5. State และ Router
ใน VibeStudio Workbench ให้ไปที่ขั้นตอนที่ 5 · สถานะและเราเตอร์ ส่วน (5A) ถึง (5C)
คุณจะคงการเลือกของผู้ใช้ไว้ในสถานะเซสชัน บังคับใช้นโยบายความปลอดภัยของช่องโดยใช้โหนดเราเตอร์ที่กำหนด และประกอบเอเจนต์งานแบบวนซ้ำเพื่อแก้ไขการละเมิดนโยบายโดยอัตโนมัติก่อนที่จะสร้างสคริปต์วิดีโอ
สถานะเวิร์กโฟลว์ (5A)
ใน Workbench ให้ไปที่สถานะเวิร์กโฟลว์ (5A)
สถานะเซสชันเทียบกับเอาต์พุตของโหนด
ในเวิร์กโฟลว์ ADK ข้อมูลจะเคลื่อนที่ผ่านกราฟผ่านกลไก 2 อย่างที่แตกต่างกัน ดังนี้
- เอาต์พุตของโหนด (
Event(output=...)): ข้อมูลที่ส่งไปยังผู้บริโภคปลายทางที่อยู่ถัดไปโดยตรงตามที่กำหนดไว้ในรายการเอดจ์ - สถานะเซสชัน (
Event(state=...)): พจนานุกรมคู่คีย์-ค่าที่แชร์ซึ่งเข้าถึงได้โดยโหนดใดๆ ที่ตามมาในวงจรการดำเนินการ
เมื่อผู้ใช้เลือกตัวเลือกที่ direction_gate ตัวเลือกจะมาถึงเป็นดัชนีตัวเลข ({"pick": "2"}) โหนดปลายทางต้องมีออบเจ็กต์ทิศทางที่สมบูรณ์ ได้แก่ ชื่อ มุมมองเรื่อง และประโยคเปิด persist_direction จะเขียนผู้สมัครที่ได้รับการแก้ไขไปยังสถานะเซสชันที่แชร์แทนที่จะส่งข้อมูลเมตาแบบละเอียดผ่านเพย์โหลดของโหนดกลางทุกโหนด
โหนดไม่จำเป็นต้องส่งพจนานุกรมสถานะเซสชันทั้งหมด เมื่อโหนดให้ผลลัพธ์ Event(state=...) โหนดจะให้เฉพาะคู่คีย์-ค่าใหม่หรือที่อัปเดตแล้ว ADK จะผสานรวมการอัปเดตเหล่านี้เข้ากับที่เก็บเซสชันโดยอัตโนมัติ
yield Event(state={"direction": chosen["title"], "angle": chosen.get("angle", ""),
"hook": hook, "user:prefs": {"last_direction": chosen["title"]}})
การส่งคืนค่านี้จะEventส่งการควบคุมไปยังรันไทม์ของ Workflow ซึ่งจะบันทึกค่าใหม่ลงในบันทึกเซสชันใน runs/sessions.db
การเชื่อมโยงพารามิเตอร์
โหนดฟังก์ชัน ADK จะอ่านสถานะเซสชันโดยอัตโนมัติผ่านการตรวจสอบพารามิเตอร์ หากลายเซ็นฟังก์ชันประกาศชื่อพารามิเตอร์ที่ตรงกับคีย์สถานะที่มีอยู่ ADK จะดึงคีย์ดังกล่าวจากสถานะและส่งต่อโดยตรง
def persist_direction(node_input, candidates: list = []):
ni = node_input if isinstance(node_input, dict) else {}
raw = ni.get("pick")
pick = str(raw).strip() if raw is not None else ""
if candidates:
i = int(pick) - 1 if pick.isdigit() else 0
chosen = candidates[max(0, min(len(candidates) - 1, i))]
else:
chosen = {"title": "untitled", "angle": "", "evidence": []}
hook = chosen.get("hook") or " ".join(chosen["title"].split()[:4])
ในที่นี้ candidates เขียนลงในสถานะเซสชันโดย direction_gate ADK จะเชื่อมโยงโดยตรงกับ persist_direction(node_input, candidates: list = []) โดยไม่ต้องค้นหาในพจนานุกรมอย่างชัดเจน
คีย์ที่ขึ้นต้นด้วย user: จะคงอยู่ตลอดเซสชันในพื้นที่เก็บข้อมูลระดับผู้ใช้ ซึ่งช่วยให้การเรียกใช้เวิร์กโฟลว์ในภายหลังเข้าถึงค่ากำหนดของครีเอเตอร์ได้
การแก้ไขภาคปฏิบัติ: การคงสถานะและการเชื่อมต่อโหนด
- ใน
agent/graph.pyภายในpersist_directionให้แทนที่บรรทัดTODO: PERSIST_STATEด้วยผลลัพธ์ของเหตุการณ์สถานะ
yield Event(state={"direction": chosen["title"], "angle": chosen.get("angle", ""),
"hook": hook, "user:prefs": {"last_direction": chosen["title"]}})
- ใน
stage3_router/agent.pyให้ต่อท้ายpersist_directionในเชนที่ 3 ในรายการedgesดังนี้
(join_research, propose_directions, direction_gate,
persist_direction)
บันทึกไฟล์ ในเวิร์กเบนช์ ให้ตรวจสอบว่า state write in place และ persist_direction in the chain มีเครื่องหมายถูกสีเขียวทั้งคู่
โหนดเราเตอร์ (5B)
ในเวิร์กเบนช์ ให้ไปที่โหนดเราเตอร์ (5B)
การกำหนดเส้นทางตามนโยบายแบบดีเทอร์มินิสติก
เราเตอร์คือโหนดฟังก์ชันเฉพาะที่ประเมินเอาต์พุตต้นทางและกํากับการดำเนินการตามกิ่งก้านของกราฟแบบมีเงื่อนไข เราเตอร์จะใช้ตรรกะที่กำหนดไว้โดยไม่ต้องเรียก LLM ซึ่งต่างจากเอเจนต์แบบ Generative
เราเตอร์จะแสดง Event โดยระบุแท็ก route ดังนี้
def length_check(node_input):
too_long = len(node_input.get("title", "")) > 60
return Event(output=node_input, route="TRIM" if too_long else "PASS")
ในการกำหนดเวิร์กโฟลว์ เป้าหมาย Edge ที่กำหนดเป็นพจนานุกรมจะแมปชื่อเส้นทางกับโหนดปลายทาง ดังนี้
(length_check, {"TRIM": shorten, "PASS": scripter}),
เราเตอร์เวิร์กโฟลว์ policy_check จะอ่านวลีที่ไม่อนุญาตจาก agent/policy_words.txt และทำการจับคู่คำแบบทั้งคำกับชื่อและมุมของทิศทางที่เลือก
return Event(output=node_input, route="BLOCK" if bad else "OK")
การจัดเก็บนโยบายเป็นข้อมูลแทนที่จะเป็นวิธีการที่ฮาร์ดโค้ดช่วยให้สามารถอัปเดตได้โดยไม่ต้องแก้ไขกราฟเวิร์กโฟลว์ การอัปเดตไฟล์ข้อความจะมีผลกับการเรียกใช้ครั้งต่อๆ ไปทันที เนื่องจากการประเมินเป็นการจับคู่ regex ที่กำหนดไว้ จึงดำเนินการในหน่วยมิลลิวินาทีโดยไม่มีค่าใช้จ่ายโทเค็นก่อนที่การเขียนสคริปต์แบบ Generative จะเริ่มขึ้น
ปลายทาง: Scripter และกักเก็บ
เราเตอร์จะกำหนดเส้นทางการรับส่งข้อมูลไปยังโหนดปลายทาง 2 โหนด
scripter: โหนดตัวแทนsingle_turnที่แปลงทิศทางที่ได้รับอนุมัติเป็นสคริปต์การผลิตที่มีโครงสร้างตามScriptสคีมา Pydantic
scripter = Agent(
name="scripter",
model=config.MODEL,
instruction=SCRIPT_INSTRUCTION,
output_schema=Script)
quarantine: เดิมเป็นฟังก์ชันตัวยึดตำแหน่งที่หยุดเส้นทางที่ถูกแจ้ง ต่อมาจะแทนที่ด้วยเอเจนต์การแก้ไขอัตโนมัติในส่วนถัดไป
การแก้ไขภาคปฏิบัติ: การกำหนดเส้นทางการตรวจสอบนโยบาย
- ใน
agent/graph.pyภายในpolicy_checkให้กรอกคำสั่ง "return" ดังนี้
return Event(output=node_input, route="BLOCK" if bad else "OK")
- ใน
stage3_router/agent.pyให้อัปเดตedgesเพื่อกำหนดเส้นทางpolicy_checkและรวมสาขาที่แยกกักตัวเข้ากับscripterอีกครั้ง
(join_research, propose_directions, direction_gate,
persist_direction, policy_check),
(policy_check, {"OK": scripter, "BLOCK": quarantine}),
(quarantine, scripter)])
บันทึกไฟล์ ในเวิร์กเบนช์ ให้ตรวจสอบว่าการแมป Edge ของเราเตอร์ได้รับการยืนยันแล้ว
โหมด Agent และโหนดงาน (5C)
ในเวิร์กเบนช์ ให้ไปที่โหมด Agent และโหนดงาน (5C)
โหมดการดำเนินการของ Agent
อินสแตนซ์ ADK Agent รองรับโหมดการดำเนินการ 3 โหมดที่ปรับให้เหมาะกับข้อกำหนดของไปป์ไลน์ที่เฉพาะเจาะจง ดังนี้
โหมด | วงจรการดำเนินการ | บทบาทในไปป์ไลน์ |
| ลูปการสนทนาไปมาแบบหลายรอบ โมเดลจะกำหนดเวลาที่จะเรียกใช้เครื่องมือ ขอข้อมูล หรือสิ้นสุดเทิร์น | เอเจนต์รูทที่โต้ตอบกับผู้ใช้ที่เป็นมนุษย์ |
| การเรียกใช้การอนุมานโมเดลเดียว ยอมรับอินพุตของโหนดก่อนหน้าและส่งออบเจ็กต์สคีมาที่มีโครงสร้าง | การแปลงกราฟตามลำดับ ( |
| ลูปอัตโนมัติที่มีการดำเนินการเครื่องมือ เอเจนต์จะทำซ้ำจนกว่าจะเรียกใช้เครื่องมือ | การแก้ไขและการตรวจสอบแบบหลายขั้นตอน ( |
การแก้ไขนโยบายแบบอัตโนมัติ
การเขียนเส้นทางที่ถูกแจ้งว่าไม่เหมาะสมใหม่ต้องใช้taskเนื่องจากจำนวนการทำซ้ำในการแก้ไขนั้นแตกต่างกันไป เอเจนต์จะได้รับคำสั่งที่ถูกแจ้งว่าไม่เหมาะสม เรียกใช้ find_policy_hits เพื่อตรวจหาการละเมิด ขอตัวเลือกที่ได้รับอนุมัติผ่าน suggest_replacement เขียนคำสั่งใหม่ และตรวจสอบความเหมาะสมก่อนดำเนินการต่อ
เครื่องมือทั้ง 2 อย่างได้รับการกําหนดไว้ใน agent/cleanup_tools.py พร้อมลายเซ็นที่พิมพ์และสตริงเอกสาร
def find_policy_hits(text: str) -> dict:
"""Which refused words appear in `text`. Matches whole words and phrases
from agent/policy_words.txt, case-insensitive.
Returns {"hits": [...], "clean": bool}. clean is true when hits is empty.
"""
def suggest_replacement(word: str) -> dict:
"""The channel's approved stand-in for a refused word, read from
agent/policy_replacements.txt.
Returns {"word", "replacement", "listed"}. When the word has no entry,
listed is false and replacement is a hint to pick a gentle synonym.
"""
การแก้ไขภาคปฏิบัติ: การประกอบเอเจนต์งานกักเก็บ
ใน stage3_router/agent.py ให้แทนที่ฟังก์ชันตัวยึดตำแหน่ง quarantine ด้วยคำจำกัดความของเอเจนต์งาน ดังนี้
quarantine = Agent(
name="quarantine",
model=config.MODEL,
instruction=QUARANTINE_INSTRUCTION,
mode="task",
tools=[find_policy_hits, suggest_replacement],
output_schema=CleanedDirection,
)
โหมดงานจะติดตั้งเครื่องมือให้กับ Agent และสิ้นสุดการดำเนินการโดยการเรียกใช้ finish_task เมื่อกำหนดค่า mode="task" แล้ว ADK จะให้ finish_task โดยอัตโนมัติและรับพารามิเตอร์จาก output_schema เพื่อให้มั่นใจว่าโหนดจะสร้างออบเจ็กต์ CleanedDirection ที่พิมพ์แล้วซึ่งตรงกับสคีมาอินพุตของโหนดสคริปต์
สิ่งที่จะเกิดขึ้นและเหตุผล
ทดสอบเส้นทางการดำเนินการทั้ง 2 เส้นทางใน ADK Web หรือ VibeStudio Workbench
- เส้นทางที่ได้รับอนุมัติ (ผู้สมัคร 1, 2 หรือ 3):
- การเลือกเส้นทางที่แนะนำซึ่งได้รับการอนุมัติจาก
policy_checkไปยังscripterโดยตรง (route="OK") - ผู้เขียนสคริปต์จะสร้างสคริปต์การผลิตแบบ 3 ช็อตตามรูปแบบ
Script
- การเลือกเส้นทางที่แนะนำซึ่งได้รับการอนุมัติจาก
- เส้นทางการแก้ไขการกักเก็บ (ผู้สมัคร 4):
- ตัวเลือกที่ 4 มีคำศัพท์ที่ถูกแจ้ง ("คลิกเบต" "เคล็ดลับไวรัล")
policy_checkเส้นทางไปยังquarantine(route="BLOCK")- ในร่องรอยเซสชัน ให้สังเกตการเรียก
quarantine, การเรียกfind_policy_hits, การเรียกsuggest_replacementสำหรับการละเมิดแต่ละครั้ง การเขียนชื่อใหม่ และการเรียกfinish_task - การดำเนินการจะกลับมาเข้าร่วม
scripterอีกครั้งเพื่อสร้างสคริปต์จากทิศทางที่ผ่านการล้างข้อมูลแล้ว
6. คลังความทรงจำ
ใน VibeStudio Workbench ให้ไปที่ขั้นตอนที่ 6 · Memory Bank ส่วน (6A) และ (6B)
ปัจจุบันเวิร์กโฟลว์ทำงานโดยไม่มีหน่วยความจำในเซสชันต่างๆ โดยแต่ละครั้งจะเริ่มจากศูนย์ ไม่ทราบว่าครีเอเตอร์เลือกอะไรไว้ก่อนหน้านี้หรือชอบแนวเพลงใด ในขั้นตอนนี้ คุณจะเชื่อมต่อ Vertex AI Agent Engine Memory Bank เพื่อจัดเก็บและเรียกค่ากำหนดของครีเอเตอร์ในแต่ละการเรียกใช้
ที่สำคัญคือ ระบบจะผสานรวมหน่วยความจำผ่านการเรียกกลับของวงจรตัวแทนแทนที่จะเป็นโหนดไปป์ไลน์ เนื่องจากการแยกและการดึงข้อมูลหน่วยความจำจะให้บริการแก่เอเจนต์แต่ละรายแทนที่จะเป็นขั้นตอนข้อมูลระดับกลาง การแนบการเรียกกลับจึงช่วยรักษาโทโพโลยีของกราฟที่สะอาดและแยกออกจากกัน
Memory Bank (6A)
ในเวิร์กเบนช์ ให้ไปที่ธนาคารความทรงจำ (6A)
ความทรงจำระดับผู้ใช้ที่มีการจัดการ
Memory Bank เป็นบริการที่มีการจัดการสำหรับหน่วยความจำของผู้ใช้ในระยะยาว โดยจะจัดระเบียบข้อเท็จจริงเกี่ยวกับบุคคลภายใต้ขอบเขตที่กำหนด ซึ่งระบุได้จากชื่อแอปพลิเคชันและรหัสผู้ใช้
SCOPE = {"app_name": config.APP, "user_id": config.USER}
TOPICS = {
"CREATOR_TASTE": "Which video directions this creator picks and passes on, "
"and how that preference changes over time.",
"CHANNEL_RULES": "Standing instructions the creator states for every video "
"(style, subjects to avoid, format rules).",
}
หัวข้อหน่วยความจำที่กำหนดเองจะกำหนดขอบเขตของสิ่งที่ธนาคารบันทึกไว้ ดังนี้
- การแยกหัวข้อ: เมื่อมีการส่งข้อความการสนทนาใหม่ผ่าน
memories.generateบริการจะใช้โมเดลการแยกกับคำอธิบายหัวข้อแต่ละรายการ ข้อความที่ไม่ตรงกับหัวข้อจะไม่สร้างความทรงจำ - การรวมและการขจัดข้อมูลที่ซ้ำกัน: บริการจะแปลงข้อเท็จจริงที่แยกออกมาใหม่เป็น Embedding และเปรียบเทียบกับความทรงจำที่มีอยู่ภายในขอบเขต เมื่อการสังเกตสอดคล้องกับความทรงจำที่มีอยู่ บริการจะอัปเดตความทรงจำนั้น เมื่อแสดงข้อมูลใหม่ บริการจะสร้างรายการใหม่ กระบวนการรวมนี้ช่วยให้มั่นใจได้ว่าเซสชันหลายๆ เซสชันเกี่ยวกับหัวข้อหนึ่งๆ จะรวมกันเป็นข้อมูลสรุปที่สอดคล้องกันแทนที่จะสร้างรายการที่ซ้ำซ้อน
- การดึงข้อมูล: การเรียกใช้
memories.retrieveที่มีขอบเขตผู้ใช้จะแสดงข้อเท็จจริงที่จัดเก็บไว้ โดยเรียงจากเก่าสุดก่อน
การดำเนินการทั้ง 2 อย่างนี้จะดำเนินการใน agent/platform/memory.py ชื่อทรัพยากรของธนาคารที่จัดสรรจะแคชไว้ในเครื่องใน runs/memorybank.json
การตั้งค่าธนาคารความทรงจำ
ใช้ตัวควบคุมเวิร์กเบนช์หรือเรียกใช้คำสั่ง CLI ในเทอร์มินัล
- เชื่อมต่อและจัดสรรธนาคาร:
python -m agent.platform.bankCREATOR_TASTEและCHANNEL_RULES - เริ่มต้นเซสชันย้อนหลัง
python -m agent.platform.bank load - ตรวจสอบข้อเท็จจริงที่รวบรวม
python -m agent.platform.bank list
การเรียกกลับ (6B)
ในเวิร์กเบนช์ ให้ไปที่การเรียกกลับ (6B) เปิด stage4_memory/agent.py
การเรียกกลับวงจรของ ADK Agent
Callback คือฟังก์ชันที่ส่งผ่านเป็นอาร์กิวเมนต์ไปยัง Agent ADK จะเรียกใช้การเรียกกลับในช่วงวงจรที่กำหนดไว้ล่วงหน้า โดยส่งบริบทที่ใช้งานอยู่ การส่งคืน None จะดำเนินการตามปกติต่อไป ส่วนการส่งคืนออบเจ็กต์แทนที่จะลบล้างหรือสกัดกั้นการดำเนินการ
ADK มีการเรียกกลับ 3 คู่ ดังนี้
คู่การเรียกกลับ | จุดเรียกใช้ | พารามิเตอร์ที่ได้รับ | ลักษณะการทำงานของค่าที่แสดงผล |
| รอบการเปลี่ยนตัวแทนทั้งหมด |
| การกด |
| รอบๆ การเรียกใช้การอนุมาน LLM แต่ละครั้ง |
| การส่งคืนจะ |
| การดำเนินการแต่ละเครื่องมือ | คำจำกัดความของเครื่องมือ อาร์กิวเมนต์ ผลลัพธ์ | การส่งคืนพจนานุกรมจะลบล้างเอาต์พุตของเครื่องมือ |
การเรียกกลับเป็นตำแหน่งที่ชัดเจนสำหรับการแทรกบริบท การป้องกัน เทเลเมทรี และการค้นหาแคชโดยไม่ต้องเพิ่มโหนดที่ไม่เกี่ยวข้องลงในกราฟเวิร์กโฟลว์
การแก้ไขด้วยตนเอง: การเรียกคืนการเชื่อมต่อและการจดจำการเรียกกลับ
- ใน
stage4_memory/agent.pyให้อัปเดตpropose_directionsเพื่อแนบbefore_model_callback=recall_tasteโดยทำดังนี้
output_schema=Directions,
before_model_callback=recall_taste)
recall_taste จะทำงานทันทีก่อนที่ Gemini จะสร้างเส้นทางที่เป็นไปได้ โดยจะดึงประวัติของครีเอเตอร์จากธนาคารความทรงจำ จัดรูปแบบความทรงจำโดยเรียงจากเก่าที่สุดก่อน แล้วต่อท้ายใน LlmRequest ที่ส่งออก พรอมต์จะสั่งให้โมเดลแนะนำผู้สมัครรับเลือกอันดับ 1-3 ตามรสนิยมปัจจุบันของครีเอเตอร์ ในขณะที่ถือว่ากฎของช่องเป็นข้อจำกัดที่เข้มงวด
- ใน
stage4_memory/agent.pyให้อัปเดตscripterเพื่อแนบafter_agent_callback=remember_pickโดยทำดังนี้
output_schema=Script,
after_agent_callback=remember_pick)
remember_pick จะทำงานหลังจาก scripter เล่นจบ โดยจะอ่านทิศทางที่เลือกจากสถานะเซสชัน สังเคราะห์ข้อความที่กระชับซึ่งสรุปการตัดสินใจของครีเอเตอร์ และเรียกใช้ memories.generate เพื่ออัปเดตธนาคารความทรงจำ
สิ่งที่จะเกิดขึ้นและเหตุผล
ทดสอบเวิร์กโฟลว์ที่เพิ่มด้วยการเรียกกลับในเวิร์กเบนช์หรือ ADK Web โดยทำดังนี้
- เรียกใช้การทดสอบด้วยพรอมต์ที่ว่างเปล่า:
- ในร่องรอยของเซสชัน ให้ตรวจสอบ
LlmRequestสำหรับpropose_directionsโปรดสังเกตบริบทความทรงจำที่ต่อท้ายซึ่งให้รายละเอียดเกี่ยวกับความชอบของครีเอเตอร์ในเรื่องธีมแฟนตาซีและจังหวะการเล่าเรื่องที่กระชับ - สังเกตทิศทางที่แนะนำ: ผู้สมัคร 1-3 สอดคล้องกับค่ากำหนดในอดีตของครีเอเตอร์แม้ว่าเทรนด์จะเน้นหัวข้ออื่นๆ ก็ตาม
- ในร่องรอยของเซสชัน ให้ตรวจสอบ
- เลือกผู้สมัครที่
direction_gate - หลังจาก
scripterเสร็จสมบูรณ์แล้ว ให้ตรวจสอบระเบียน Memory Bank โดยทำดังนี้python -m agent.platform.bank list
7. เครื่องมือ RAG
ใน VibeStudio Workbench ให้ไปที่ขั้นตอนที่ 7 · เครื่องมือ RAG ส่วน (7A) และ (7B)
วิดีโอที่เผยแพร่จะสะสมความคิดเห็นของผู้ชมอย่างต่อเนื่อง เราจะรวบรวมความคิดเห็นที่เป็นตัวแทน 30 รายการในagent/comments.md ซึ่งจะบันทึกคำชมของผู้ชม คำวิจารณ์เกี่ยวกับจังหวะการแสดงโฆษณา และความชอบด้านเสียง ในขั้นตอนนี้ คุณจะจัดทำดัชนีความคิดเห็นเหล่านี้โดยใช้ Vertex AI RAG Engine และเชื่อมต่อการดึงข้อมูลเชิงความหมายเข้ากับแฟนเอาต์ (Fan-Out) การค้นคว้า
การดึงข้อมูลจากเอกสาร (7A)
ใน Workbench ให้ไปที่ RAG Engine (7A)
Memory Bank กับเครื่องมือ RAG
เครื่องมือทั้ง 2 รายการนี้ใช้เวิร์กโฟลว์ตามข้อมูลภายนอก แต่มีจุดประสงค์ด้านสถาปัตยกรรมที่แตกต่างกัน ดังนี้
มิติข้อมูล | คลังความทรงจำ | เครื่องมือ RAG |
กรณีการใช้งานหลัก | ค่ากำหนดของผู้ใช้และกฎการปฏิบัติงานในระยะยาว | การดึงข้อมูลเชิงความหมายในคอลเล็กชันเอกสารขนาดใหญ่ |
ขอบเขต | กำหนดขอบเขตเป็นรหัสผู้ใช้และชื่อแอปพลิเคชันแต่ละรายการ | กำหนดขอบเขตไว้ที่ทรัพยากรคลังข้อความที่แชร์ในผู้ใช้ทั้งหมด |
การประมวลผลข้อมูล | การแยก การฝัง และการรวมความหมายแบบเรียลไทม์ | การแบ่งเอกสารออกเป็นส่วนๆ การฝังเวกเตอร์ และการค้นหาเพื่อนบ้านที่ใกล้ที่สุด |
การผสานรวมกราฟ | Agent Lifecycle Callback ( | โหนดฟังก์ชันเฉพาะใน Fan-Out ของการวิจัย ( |
การแบ่งเอกสารออกเป็นส่วนๆ และการฝัง
เครื่องมือ RAG จะจัดทำดัชนีเอกสารโดยการแบ่งข้อความเป็นข้อความเชิงความหมายและจัดเก็บเวกเตอร์ไว้ในฐานข้อมูลที่มีการจัดการ
corpus = rag.create_corpus(
display_name="vibestudio-feedback",
description="Vibe Studio: what the audience wrote under the channel's past videos.",
backend_config=rag.RagVectorDbConfig(
rag_embedding_model_config=rag.RagEmbeddingModelConfig(
vertex_prediction_endpoint=rag.VertexPredictionEndpoint(
publisher_model="publishers/google/models/text-embedding-005"))))
rag.upload_file(
corpus_name=corpus.name, path="agent/comments.md", display_name="comments.md",
transformation_config=rag.TransformationConfig(
chunking_config=rag.ChunkingConfig(chunk_size=120, chunk_overlap=20)))
- ขนาดก้อน: กำหนดค่าเป็น 120 โทเค็นโดยมีโทเค็นที่ทับซ้อนกัน 20 โทเค็น ซึ่งจะบันทึกความคิดเห็น 2-3 รายการต่อข้อความ เพื่อให้มั่นใจว่าเวกเตอร์แต่ละรายการแสดงถึงความรู้สึกที่สอดคล้องกันโดยไม่ลดทอนความหมายของความคิดเห็นที่ไม่เกี่ยวข้อง
- โมเดลการฝัง:
text-embedding-005แปลงข้อความเป็นเวกเตอร์ที่มีมิติสูง เมื่อส่งคำค้นหา โมเดลจะแปลงคำค้นหาเป็นเวกเตอร์และค้นหารายการที่ตรงกันมากที่สุดตามระยะห่างเชิงความหมาย ความคิดเห็นเกี่ยวกับมังกรตัวเล็กๆ ที่เฝ้าถุงเท้าตรงกับพรอมต์เกี่ยวกับสัตว์วิเศษโดยไม่ต้องมีคีย์เวิร์ดที่ทับซ้อนกันอย่างแม่นยำ
การตั้งค่าคลัง RAG
เริ่มต้น Corpus โดยใช้ปุ่มเวิร์กเบนช์หรือคำสั่งเทอร์มินัล
- สร้างคลังข้อความ
python -m agent.platform.ragruns/ragcorpus.json - อัปโหลดและจัดทำดัชนีความคิดเห็น: อัปโหลด
agent/comments.mdด้วยการกำหนดค่าการแบ่งกลุ่มและรอให้การจัดทำดัชนีเสร็จสมบูรณ์ - ค้นหาในคลังข้อความ: ทดสอบการดึงข้อมูลความคล้ายคลึงด้วยการค้นหาที่ไม่มีคำตรงกับความคิดเห็น (เช่น ค้นหา "สัตว์วิเศษตัวเล็กๆ" เพื่อดึงความคิดเห็นเกี่ยวกับมังกร)
โหนดการดึงข้อมูล (7B)
ในเวิร์กเบนช์ ให้ไปที่ผู้อ่านคนที่ 3 (7B) เปิด stage5_rag/agent.py
การดึงข้อมูลเป็นโหนดกราฟ
ความคิดเห็นของผู้ชมแสดงถึงข้อมูลการวิจัยที่แชร์ในเวิร์กโฟลว์ ความรู้สึกของผู้ชมจะส่งไปยัง join_research โดยตรงพร้อมกับข้อมูลเทรนด์และข้อมูลที่ค้างอยู่ ซึ่งแตกต่างจากหน่วยความจำส่วนตัวของผู้สร้าง ดังนั้นจึงมีการติดตั้งใช้งานเป็นโหนดฟังก์ชันดังนี้
def read_feedback(node_input):
"""The third reader (step 7): what the audience wrote under past videos,
the passages nearest to tonight's idea. Retrieval, not a model call."""
from .platform import rag
idea = idea_text(node_input)
query = idea or "what viewers liked and what they complained about"
try:
hits = rag.retrieve(query)
except Exception as e:
print(f" [rag] feedback unavailable ({str(e)[:80]})")
return Event(output={"query": query, "feedback": [],
"note": "no corpus connected - run: python -m agent.platform.rag"})
return Event(output={"query": query, "feedback": [h["text"] for h in hits]})
read_feedback จะดึงแนวคิดเริ่มต้นของผู้ใช้และเรียกใช้การค้นหาเวกเตอร์กับคลังข้อมูลของเครื่องมือ RAG โดยจะส่งความคิดเห็นที่ดึงมาในEvent(output=...)เพย์โหลด
การแก้ไขภาคปฏิบัติ: การต่อสายรีดเดอร์ตัวที่ 3 เข้ากับ Fan-Out
ใน stage5_rag/agent.py ให้อัปเดต edges เพื่อเพิ่ม read_feedback เป็นกิ่งก้านแบบขนานที่ 3 ซึ่งเข้าสู่ join_research
(START, read_backlog, join_research),
(START, read_feedback, join_research),
เนื่องจาก join_research เป็น JoinNode จึงซิงโครไนซ์กิ่งทั้งหมดที่เข้ามา โดยจะรอจนกว่า scan_trends, read_backlog และ read_feedback จะปล่อยเหตุการณ์ทั้งหมดก่อนที่จะส่งต่อแพ็กเกจที่รวบรวมแล้วไปยังสตรีมปลายทาง
สิ่งที่จะเกิดขึ้นและเหตุผล
เรียกใช้เวิร์กโฟลว์ในเวิร์กเบนช์
- ส่งพรอมต์ไอเดีย (เช่น "มังกรตัวเล็กๆ เฝ้าเคาน์เตอร์ครัว")
- ในร่องรอยการดำเนินการ ให้ตรวจสอบว่าโหนด Reader ทั้ง 3 โหนดดำเนินการพร้อมกัน
- สังเกต
join_research: พจนานุกรมเอาต์พุตมีtrends,backlogและfeedbackแล้ว - ตรวจสอบผู้สมัครที่สร้างขึ้นจาก
propose_directions: โมเดลจะรวมความคิดเห็นของผู้ชมไว้ในข้อเสนอและอ้างอิงความรู้สึกของผู้ชมในช่องหลักฐาน - โปรดทราบว่าการดึงข้อมูล RAG เป็นแบบดีเทอร์มินิสติก (คำค้นหาที่เหมือนกันจะแสดงข้อความความคิดเห็นที่เหมือนกัน) ในขณะที่โหนดข้อเสนอแบบ Generative จะสร้างรูปแบบที่สร้างสรรค์
8. การสร้างวิดีโอแบบอะซิงโครนัสด้วย Veo
ใน VibeStudio Workbench ให้ไปที่ขั้นตอนที่ 8 · วิดีโอ ส่วน (8A) และ (8B)
การสร้างวิดีโอความละเอียดสูงด้วย Google Veo ต้องใช้เวลาหลายนาทีต่อการเรนเดอร์ การบล็อก Graph Execution ในช่วงนี้จะทำให้สิ้นเปลืองทรัพยากรการประมวลผล ล็อกพูลเธรด และทำให้การเรียกใช้เกิดการหลุดของการเชื่อมต่อ HTTP ในขั้นตอนนี้ คุณจะทำให้การแสดงวิดีโอเป็นแบบอะซิงโครนัสโดยใช้ LongRunningFunctionTool ของ ADK
เครื่องมือที่ใช้เวลานาน (8A)
ในเวิร์กเบนช์ ให้ไปที่เครื่องมือที่ทำงานเป็นเวลานาน (8A) เปิด stage6_video/agent.py และ agent/deliver.py
เครื่องมือแบบซิงโครนัสเทียบกับเครื่องมือที่ทำงานเป็นเวลานาน
เครื่องมือฟังก์ชัน ADK มาตรฐานจะทำงานแบบซิงโครนัสภายในเทิร์นของเอเจนต์ โดยโมเดลจะเรียกใช้เครื่องมือ รอเพย์โหลดที่ส่งคืน และรวมผลลัพธ์เข้ากับเทิร์นที่กำลังดำเนินการ
การเรนเดอร์วิดีโอไม่สามารถดำเนินการให้เสร็จสมบูรณ์ได้ในรอบเดียว แต่ render_submit จะเริ่มงานสร้างและแสดงใบเสร็จการดำเนินการที่มีสถานะ "pending" ทันที
def render_submit(prompt: str) -> dict:
"""Submit one Veo render of `prompt`. Returns at once with a pending
receipt; the clip is delivered later, to this call, by id."""
receipt = videogen.start(f"{prompt} {videogen.NO_TEXT}")
return {"status": "pending", "operation": receipt["operation"], "prompt": receipt["prompt"]}
เมื่อห่อหุ้มด้วย LongRunningFunctionTool ADK จะสกัดกั้นสถานะ "pending" เทิร์นของตัวแทนจะสิ้นสุดลง เวิร์กโฟลว์จะหยุดชั่วคราวที่โหนด และระบบจะบันทึกข้อมูลเมตาการโทรที่รอดำเนินการ (รวมถึงรหัสการโทรและใบเสร็จ) ใน runs/sessions.db กระบวนการดำเนินการจะสิ้นสุดอย่างราบรื่นโดยไม่ต้องรักษาการเชื่อมต่อเครือข่ายที่ใช้งานอยู่หรือเธรดของ Worker
การแก้ไขภาคปฏิบัติ: การรวมเครื่องมือการแสดงผล
ใน stage6_video/agent.py ให้อัปเดต render_desk เพื่อรวม render_submit ไว้ใน LongRunningFunctionTool ดังนี้
tools=[LongRunningFunctionTool(render_submit)])
การกลับมาทำงานต่อตามรหัสการโทร
รูปแบบการกลับมาเล่นอีกครั้งแบบสากล
ADK ใช้กลไกเดียวกันในการระงับและกลับมาทำงานต่อสำหรับทั้งมนุษย์และเครื่องมือภายนอก ดังนี้
ทริกเกอร์การระงับ | การเริ่มต้น Construct | สถานะการระงับที่จัดเก็บ | เหตุการณ์การกลับมาทำงาน |
การตัดสินใจของเจ้าหน้าที่ |
| เปิดพรอมต์อินพุตในที่เก็บเซสชัน |
|
เครื่องมือที่ใช้เวลานาน |
| เปิดการเรียกใช้เครื่องมือในที่เก็บเซสชัน |
|
ในทั้ง 2 กรณี เวิร์กโฟลว์จะหยุดโดยสมบูรณ์และจะกลับมาทำงานอีกครั้งเมื่อมีเหตุการณ์ที่มี FunctionResponse ที่ตรงกันมาจากแหล่งที่มาภายนอก ได้แก่ อินเทอร์เฟซผู้ใช้ เว็บฮุก หรือ Worker ที่ทำงานเบื้องหลัง
การแก้ไขภาคปฏิบัติ: การตอบกลับการนำส่ง
ใน agent/deliver.py ให้สร้างส่วนFunctionResponseการกลับมาทำงานต่อ
part = Part(function_response=FunctionResponse(
id=row["call_id"], name=row["name"], response=response))
Daemon การนำส่งจะสำรวจ Veo จนกว่าจะสร้างไฟล์วิดีโอ จากนั้นจะส่ง FunctionResponse นี้ไปยังเซสชัน ADK จะจับคู่รหัสการโทรและดำเนินการเวิร์กโฟลว์ต่อที่โหนดถัดไปโดยตรง โหนดที่เสร็จสมบูรณ์แล้วจะไม่ดำเนินการอีก และเอเจนต์จะไม่ดำเนินการสร้างอีก
การตั้งค่า STUDIO_REAL_VIDEO=0 ใน .env จะเปิดใช้การแสดงผลจำลอง: start จะแสดงใบเสร็จการทดสอบทันที และ check จะจำลองการดำเนินการให้เสร็จสมบูรณ์ใน 5 วินาทีโดยไม่ต้องทำการเรียกใช้ Veo API ที่เรียกเก็บเงินได้
การผสานรวมไปป์ไลน์ (8B)
ในเวิร์กเบนช์ ให้ไปที่ render_desk ในกราฟ (8B) เปิด stage6_video/agent.py
โหนดเทอร์มินัลในไปป์ไลน์คือ store_video โดยจะอ่านข้อมูลการเรนเดอร์ที่เสร็จสมบูรณ์จาก runs/state.json (ที่บันทึกกระบวนการนำส่ง) และส่ง URL ของวิดีโอและสถานะการสร้างไปยังสถานะเซสชันที่แชร์
การแก้ไขภาคปฏิบัติ: การเชื่อมต่อท่อส่งวิดีโอที่สมบูรณ์
ใน stage6_video/agent.py ให้อัปเดต edges เพื่อต่อท้าย render_desk และ store_video โดยทำดังนี้
(quarantine, scripter),
(scripter, render_desk, store_video)])
สิ่งที่จะเกิดขึ้นและเหตุผล
ทดสอบโฟลว์การสร้างแบบไม่พร้อมกันในเวิร์กเบนช์
- ดำเนินการเวิร์กโฟลว์ผ่านการเลือกผู้สมัครรับเลือกและการสร้างสคริปต์
- ที่
render_deskให้สังเกต Agent เรียกใช้render_submit - เวิร์กโฟลว์จะถูกระงับทันที ในเวิร์กเบนช์หรือ ADK Web ให้สังเกตสถานะ "รอดำเนินการ" โดยเซสชันจะมีรหัสการเรียกที่เปิดอยู่ และไม่มีกระบวนการเบื้องหลังใดที่ใช้ทรัพยากร
- เรียกใช้ Daemon การนำส่งโดยใช้คอนโซล Workbench หรือในเทอร์มินัล
python -m agent.deliver - ใน ADK Web ให้รีเฟรชเซสชัน: การดำเนินการจะกลับมาทำงานต่อที่
store_video, บันทึก URL ของวิดีโอไปยังสถานะเซสชัน และเวิร์กโฟลว์จะเสร็จสมบูรณ์
9. ทำให้ใช้งานได้กับ Cloud Run
ใน VibeStudio Workbench ให้ไปที่ขั้นตอนที่ 9 · การติดตั้งใช้งาน
คุณได้พัฒนาและยืนยันแต่ละคอมโพเนนต์ของไปป์ไลน์ในแซนด์บ็อกซ์เฉพาะ ในขั้นตอนนี้ คุณจะได้ประกอบไปป์ไลน์การผลิตที่สมบูรณ์และนำไปใช้งานใน Google Cloud Run
ADK Runner
adk web เป็นผู้ประสานงานกราฟในระหว่างการพัฒนา ในเวอร์ชันที่ใช้งานจริง แอปพลิเคชันจะโฮสต์เวิร์กโฟลว์โดยใช้คลาส Runner ของ ADK ดังนี้
self._svc = DatabaseSessionService(db_url=config.DB_URL)
self._runner = Runner(app_name=config.APP, agent=wf, session_service=self._svc)
async for ev in self._runner.run_async(user_id=config.USER, session_id=run_id, new_message=message):
self._absorb(ev) # fold the ADK event into the run state, publish one app event
# the gate's answer and the render's delivery are the same call, with a function_response part
part = Part(function_response=FunctionResponse(id=call_id, name=name, response=response))
run_async: ขับเคลื่อนการดำเนินการเวิร์กโฟลว์ ซึ่งจะสร้างเหตุการณ์ตามลําดับเมื่อโหนดทํางานและคงการอัปเดตไว้ในบริการเซสชัน- การกลับมาทำงานต่อแบบรวม: ทั้งการตัดสินใจของผู้ใช้ที่
direction_gateและการนำส่งวิดีโอที่เสร็จสมบูรณ์จาก Veo จะกลับมาดำเนินการต่อผ่านออบเจ็กต์FunctionResponseที่เหมือนกันซึ่งส่งไปยังrun_async
สถาปัตยกรรมแอปพลิเคชันเวอร์ชันที่ใช้งานจริง
แอปพลิเคชันการผลิตใน vibestudio/ ผสานรวมไปป์ไลน์ทั้งหมด
vibestudio/
server/
main.py FastAPI: application server, REST routes, static assets
api.py REST API endpoints: run, pick, publish, backlog, profile, history
runner.py Runner orchestration over the workflow, background render poller
platform/ Event bus (SSE stream), file storage, publishing, telemetry
agent/ Production agent package, verified by checks/verify_app.py
graph.py The complete workflow graph and node definitions
desk.py render_desk and render_submit wrapped with LongRunningFunctionTool
schemas.py Pydantic schemas: Directions, CleanedDirection, Script
cleanup_tools.py Deterministic policy tools: find_policy_hits, suggest_replacement
platform/ Memory Bank, RAG Engine, and Veo integrations
web/ Production React user interface
Dockerfile · deploy.py · run.sh
- สตรีมเหตุการณ์เดียว: Backend ของ FastAPI จะเผยแพร่เหตุการณ์ในสตรีมเหตุการณ์ที่เซิร์ฟเวอร์ส่ง (SSE) รายการเดียว ส่วนหน้าของ React จะแสดงภาพความคืบหน้าของกราฟแบบเรียลไทม์และจัดการการเชื่อมต่อที่ล่าช้าโดยไม่สูญเสียสถานะ
- การดำเนินการที่แยกออกจากกัน: แอปพลิเคชันจัดการวนรอบเหตุการณ์ กราฟเวิร์กโฟลว์จะมุ่งเน้นที่ตรรกะการดำเนินการโดยสมบูรณ์ โดยไม่ทราบอินเทอร์เฟซส่วนหน้า
รายการ Edge ของเวิร์กโฟลว์ที่สมบูรณ์ใน agent/graph.py จะรวมรูปแบบสถาปัตยกรรมทั้งหมดที่สร้างขึ้นตลอดทั้งโค้ดแล็บนี้
(START, scan_trends, join_research),
(START, read_backlog, join_research),
(START, read_feedback, join_research),
(join_research, propose_directions, direction_gate,
persist_direction, policy_check),
(policy_check, {"OK": scripter, "BLOCK": quarantine}),
(quarantine, scripter),
(scripter, render_desk, store_video),
การติดตั้งใช้งานกับ Cloud Run
Google Cloud Run มีการโฮสติ้งแบบไร้เซิร์ฟเวอร์ที่มีการปรับขนาดอัตโนมัติ การกำหนดเส้นทางคำขอ และบิลด์คอนเทนเนอร์แบบผสานรวม
gcloud run deploy vibestudio --source vibestudio \
--project $GOOGLE_CLOUD_PROJECT --region us-central1 \
--labels dev-tutorial-codelab=vibetube --allow-unauthenticated \
--memory 2Gi --cpu 2 --timeout 3600 --concurrency 40 \
--max-instances 1 --min-instances 1 --session-affinity \
--set-env-vars GOOGLE_CLOUD_PROJECT=...,STUDIO_VERTEX=1,STUDIO_MEMORY_BANK=...,STUDIO_RAG_CORPUS=...,VIBETUBE_URL=...,VIBETUBE_EVENT=...,VIBETUBE_NAME=...,VIBETUBE_PROJECT=...
- บิลด์คอนเทนเนอร์:
gcloud run deploy --sourceแพ็กเกจไดเรกทอรีvibestudio/สร้างอิมเมจคอนเทนเนอร์โดยใช้ Cloud Build และทำให้บริการใช้งานได้ในการดำเนินการเดียว - เซสชันแอฟฟินิตี้: นำคำขอจากผู้ใช้รายเดียวกันไปยังอินสแตนซ์คอนเทนเนอร์เดียวกัน ซึ่งจะรักษาสถานะเซสชันในเครื่องไว้ตลอดขั้นตอนการทำซ้ำ
- ความสามารถในการสังเกต: การผสานรวม Cloud Trace จะบันทึก Span แบบกระจายสำหรับทุกโหนด การเรียก LLM และการดำเนินการเครื่องมือ ซึ่งเข้าถึงได้ใน คอนโซล Google Cloud ภายใต้ Trace Explorer
คลิกปุ่มติดตั้งใช้งานในเวิร์กเบนช์เพื่อเรียกใช้สคริปต์การติดตั้งใช้งาน เมื่อการสร้างเสร็จสมบูรณ์ เทอร์มินัลจะแสดง URL ของบริการที่ใช้งานจริง
10. สรุป
ใน VibeStudio Workbench ให้ไปที่ขั้นตอนที่ 10 · สรุปเพื่อตรวจสอบสถาปัตยกรรมที่เสร็จสมบูรณ์
ขั้นตอน | สถาปัตยกรรมและแนวคิด | รูปแบบการใช้งาน |
พรอมต์เดียว | พรอมต์เดียว เครื่องมือฟังก์ชัน ลูปแชทตามลำดับ |
|
พื้นฐานของเวิร์กโฟลว์ Agentic AI | เวิร์กโฟลว์กราฟ การวิจัยแบบคู่ขนาน เอาต์พุตสคีมา เกตของมนุษย์ |
|
State และ Router | สถานะเซสชันที่แชร์ การเชื่อมโยงพารามิเตอร์ การกำหนดเส้นทางที่แน่นอน ตัวแทนงาน |
|
คลังความทรงจำ | ความทรงจำระยะยาวระดับผู้ใช้ การรวมความหมาย ฮุกวงจร |
|
เครื่องมือ RAG | การดึงข้อมูลเอกสารจากความคิดเห็นของผู้ชม การฝังเชิงความหมาย | โหนด |
การสร้างวิดีโอแบบอะซิงโครนัสด้วย Veo | เครื่องมือที่ทำงานเป็นเวลานาน ใบเสร็จที่รอดำเนินการ Daemon การนำส่งภายนอก |
|
ทำให้ใช้งานได้กับ Cloud Run | การจัดการเป็นกลุ่มแบบเป็นโปรแกรม, Server-Sent Events, คอนเทนเนอร์แบบ Serverless |
|
หลักการสำคัญด้านสถาปัตยกรรม
- ระงับแทนการรอ: เวิร์กโฟลว์จะหยุดชั่วคราวอย่างราบรื่นเพื่อรออินพุตจากผู้ใช้ (
RequestInput) หรือการดำเนินการที่ใช้เวลานาน (LongRunningFunctionTool) กระบวนการจะไม่รออย่างไม่มีการใช้งานในเทรดหรือซ็อกเก็ตเครือข่าย - การกลับมาทำงานอีกครั้งแบบสากล: การระงับทุกครั้งจะกลับมาทำงานอีกครั้งผ่านกลไกเดียวกัน นั่นคือ
function_responseเดียวที่มีรหัสการเรียกของโหนดที่ถูกระงับ - การจัดการสถานะที่แยกออกจากกัน: โหนดแชร์ข้อมูลผ่านคีย์สถานะเซสชันที่มีชื่อและการเชื่อมโยงพารามิเตอร์แทนที่จะใช้เพย์โหลดระดับกลางที่เชื่อมโยงอย่างแน่นหนาและมีรายละเอียดมาก
- การกำหนดเส้นทางที่แน่นอนก่อนต้นทุนแบบ Generative: เราเตอร์ตามกฎและตัวกรองนิพจน์ทั่วไปจะประเมินนโยบายโดยไม่มีค่าโทเค็นก่อนที่โมเดล Generative จะทำงาน
- การแยกความกังวล: บริบทที่เฉพาะเจาะจงสำหรับตัวแทนแต่ละรายจะอยู่ในการเรียกกลับของวงจร ส่วนการอ้างอิงข้อมูลที่แชร์จะอยู่ใน
