ทำลายกำแพงระหว่างการอนุมานและการฝึก! Uni-Agent: กรอบงานแบบรวมที่ทำให้การสร้าง การทำงาน และการฝึกเรียนรู้แบบเสริมกำลังของเอเจนต์เป็นไปอย่างราบรื่น

ปัจจุบัน วงการ AI Agent กำลังเผชิญกับสถานการณ์ที่ค่อนข้างน่าอึดอัดใจ นั่นคือการแยกส่วนกันอย่างชัดเจนระหว่างขั้นตอนการอนุมาน (Inference) ที่ต้องพึ่งพาเฟรมเวิร์กหนึ่ง และขั้นตอนการฝึก (Training) ที่ต้องเปลี่ยนไปใช้อีกเฟรมเวิร์กหนึ่ง ท่อส่งข้อมูล (Data Pipeline), สิ่งที่เป็นนามธรรมของสภาพแวดล้อม (Environment Abstraction) และอินเทอร์เฟซเครื่องมือ (Tool Interface) ระหว่างทั้งสองนี้แทบจะไม่เข้ากันเลย

นั่นหมายความว่า หลังจากที่นักวิจัยได้ตรวจสอบความสามารถในการอนุมานของ Agent ตัวหนึ่งแล้ว หากต้องการปรับปรุงมันให้ดียิ่งขึ้นด้วยการเรียนรู้แบบเสริมกำลัง (Reinforcement Learning) พวกเขาก็จำเป็นต้องเขียนตรรกะการโต้ตอบทั้งหมดขึ้นมาใหม่ตั้งแต่ต้น กระบวนการนี้ไม่เพียงแต่ใช้เวลามหาศาล แต่ยังง่ายต่อการนำบั๊กที่เกิดจากความไม่สอดคล้องกันในรูปแบบต่างๆ ที่คาดไม่ถึงเข้ามาอีกด้วย

เพื่อแก้ไขความขัดแย้งพื้นฐานนี้ Uni-Agent จึงได้ถือกำเนิดขึ้น มันนำเสนอ Unified Interaction Stack เพื่อให้แน่ใจว่า Agent จะใช้เส้นทางโค้ดที่เหมือนกันทุกประการเมื่อทำการอนุมานและเมื่อทำการฝึกด้วยการเรียนรู้แบบเสริมกำลัง

ปรัชญาการออกแบบหลักของมันคือการแยกส่วนออกเป็นสามชั้น: model รับผิดชอบการคิด, tool รับผิดชอบการรับรู้และการกระทำ, env รับผิดชอบการคงสถานะไว้ ทั้งสามส่วนนี้สามารถถูกแทนที่ได้อย่างอิสระ และยังสามารถประกอบเข้าด้วยกันได้อย่างราบรื่น ด้วยกลไกการควบคุมการทำงานพร้อมกันแบบสัญญาณอะซิงโครนัส (Asynchronous Semaphore) Uni-Agent จึงทำงานได้อย่างเสถียรในระดับการทำงานพร้อมกันมากกว่า 1,000 งาน และ ทำคะแนนได้สูงถึง 67.7 บน SWE-Bench-Verified

รูปด้านล่างแสดงสถาปัตยกรรมโดยรวมของระบบ Agent Uni-Agent ซึ่งแบ่งออกเป็นสองโมดูลหลักอย่างชัดเจน: การโต้ตอบและการทำงาน (Interaction & Execution) และการฝึกโมเดล (Model Training) Agent Interaction System ในชั้นบนเป็นแกนหลักของวงจรการทำงาน: Agent Chat Model (ที่รวมเอาโมเดลขนาดใหญ่ชั้นนำหลายตัว) เรียกใช้ความสามารถต่างๆ จาก Tools Pool เช่น เครื่องมือพัฒนา เบราว์เซอร์ ฯลฯ เพื่อระบุหรือแก้ไข Environment (เช่น แผงตรวจสอบ บริการคลาวด์ สภาพแวดล้อมคอนเทนเนอร์) การเปลี่ยนแปลงสถานะของสภาพแวดล้อมจะถูกส่งกลับไปยังโมเดล เพื่อสนับสนุนการตัดสินใจในขั้นตอนถัดไป Training System ในชั้นล่าง (ใช้ verl framework เป็นตัวอย่าง) รับผิดชอบการปรับแต่งโมเดล: เมื่อภารกิจเสร็จสิ้น Reward System จะสร้างสัญญาณรางวัล ส่งผ่าน Message Queue ไปยัง Training Engine ซึ่งจะอัปเดตน้ำหนักของ Agent Model ตามข้อมูลเหล่านี้ ทำให้ Agent สามารถพัฒนาความสามารถในการตัดสินใจได้อย่างต่อเนื่องผ่านการโต้ตอบและการฝึกฝนอย่างต่อเนื่อง ก่อให้เกิดห่วงโซ่ที่สมบูรณ์ตั้งแต่การดำเนินการไปจนถึงการตอบรับและการปรับปรุงซ้ำ

แผงตรวจสอบแบบน้ำหนักเบาของ Uni-Agent ส่วนบนแสดงสถานะโดยรวมของงานอย่างชัดเจน: มีทั้งหมด 500 งาน กำลังทำงาน 64 งาน รอคิว 339 งาน เสร็จสิ้น 97 งาน สะท้อนถึงคิวงานและความคืบหน้าของการดำเนินการได้อย่างชัดเจน แผงด้านซ้ายแสดง ID ของงานที่กำลังทำงานอยู่ทั้งหมดและขั้นตอนการดำเนินการปัจจุบัน แผงบันทึกสี่ช่องตรงกลางและด้านขวาจะแสดงห่วงโซ่การโต้ตอบที่สมบูรณ์ของ Agent แต่ละตัว ห่วงโซ่เหล่านี้ประกอบด้วย “ความคิด (Thought)”, “การกระทำ (Action เช่น การรันสคริปต์ Python, การรันคำสั่งทดสอบ pytest)”, “การสังเกต (Observation เช่น ข้อมูลข้อผิดพลาดของโค้ด, คำเตือนการเลิกใช้งาน, ข้อมูลการใช้ Token)” ซึ่งสร้างกระบวนการตัดสินใจ การดำเนินการตามคำสั่ง และการตอบรับจากสภาพแวดล้อมของ Agent ขึ้นมาใหม่ทั้งหมด ช่วยให้นักพัฒนาสามารถติดตามสถานะการทำงานของ Agent ในงานพัฒนาและดีบักได้แบบเรียลไทม์ และระบุจุดที่มีปัญหาได้อย่างรวดเร็ว ข้อมูลเพิ่มเติมสามารถดูได้ที่: แผงตรวจสอบแบบน้ำหนักเบาของ Uni-Agent[1]

unsetunsetสารบัญunsetunset

• เริ่มต้นใช้งานอย่างรวดเร็ว
• หนึ่ง ภาพรวมโครงสร้างโครงการ
• สอง ภาพรวมสถาปัตยกรรมและกระแสข้อมูลหลัก
• 2.1 ปรัชญาการออกแบบแบบแยกส่วนสามชั้น
• 2.2 กระแสข้อมูลของการโต้ตอบที่สมบูรณ์หนึ่งครั้ง
• สาม จุดเริ่มต้นแบบรวม: ตรรกะการจัดเรียงของ UniAgentLoop
• 3.1 การควบคุมการทำงานพร้อมกันและการแยกทรัพยากร
• 3.2 ขั้นตอนการดำเนินการหลัก
• 3.3 การลดระดับอย่างสง่างามสำหรับเส้นทางที่ผิดปกติ
• สี่ เอนจินการโต้ตอบ: วงจรหลายรอบของ AgentInteraction
• 4.1 โมเดลข้อมูล StepOutput
• 4.2 สายการผลิตห้าขั้นตอนของการโต้ตอบทีละขั้นตอน
• 4.3 กลไกงบประมาณหมดเวลา
• ห้า Rollout Cache: การสะสมข้อมูลการฝึกแบบเรียลไทม์
• 5.1 โครงสร้างข้อมูลหลัก
• 5.2 การสะสมระหว่างการสร้างโมเดล
• 5.3 การสะสมระหว่างการตอบรับจากสภาพแวดล้อม
• 5.4 การคำนวณที่แม่นยำของ Message Boundary Tokens
• หก สิ่งที่เป็นนามธรรมของสภาพแวดล้อม: AgentEnv และการปรับใช้หลายแบ็กเอนด์
• 6.1 อินเทอร์เฟซสภาพแวดล้อมแบบรวม
• 6.2 ยูเนียนจำแนกประเภทของแบ็กเอนด์การปรับใช้
• 6.3 RemoteRuntime: ชั้นการสื่อสาร HTTP ที่เชื่อถือได้
• 6.4 การจัดการหมดเวลาและการขัดจังหวะที่แข็งแกร่ง
• เจ็ด ระบบเครื่องมือ: ตั้งแต่การลงทะเบียนไปจนถึงการดำเนินการ
• 7.1 รูปแบบรีจิสทรี
• 7.2 ขั้นตอนการติดตั้งเครื่องมือ
• 7.3 การแปล Tool Call เป็น Bash
• 7.4 Tool Parser หลายรูปแบบ
• แปด ระบบรางวัล: การตรวจสอบภายในสภาพแวดล้อม
• 8.1 รีจิสทรีและคลาสพื้นฐานนามธรรม
• 8.2 รางวัล SWE-Bench: ขับเคลื่อนโดยกรณีทดสอบจริง
• เก้า auto_await: การเชื่อมต่อแบบซิงโครนัสและอะซิงโครนัสอย่างราบรื่น
• สิบ สิ่งที่เป็นนามธรรมของโมเดลคู่: ความเป็นหนึ่งเดียวของการฝึกและการอนุมานบริสุทธิ์
• 10.1 AgentChatModel vs OpenAICompatibleChatModel
• 10.2 การจัดการ Tool Call สำหรับโมเดลที่เข้ากันได้กับ OpenAI
• สิบเอ็ด โมเดลการทำงานพร้อมกันและวิศวกรรมประสิทธิภาพ
• 11.1 สถาปัตยกรรมอะซิงโครนัสแบบครบวงจร
• 11.2 ข้อมูลประสิทธิภาพที่สำคัญ
• สิบสอง การบันทึกผลลัพธ์และการสังเกตการณ์
• 12.1 การคงอยู่ของผลลัพธ์การโต้ตอบ
• 12.2 การอนุมานสถานะของ Dashboard
• สรุปและแนวโน้มในอนาคต

unsetunsetเริ่มต้นใช้งานอย่างรวดเร็วunsetunset

# โคลน repository และเริ่มต้นโมดูลย่อย
git clone https://github.com/verl-project/uni-agent.git
cd uni-agent
git submodule update --init --recursive

# ติดตั้ง verl (เป็น Python package)
pip install --no-deps -e ./verl

# ติดตั้ง dependencies หลัก
pip install swe-rex loguru pydantic pydantic_settings aiohttp

หลังจากติดตั้งเสร็จ คุณสามารถเริ่มสภาพแวดล้อม Agent และรันสคริปต์ตัวอย่างได้โดยตรง สำหรับการกำหนดค่าโดยละเอียดเพิ่มเติมเกี่ยวกับสถานการณ์อื่นๆ (เช่น การสร้าง Search Agent, การอนุมานแบบขนาน, การฝึก RL) โปรดดูเอกสารทางการ[2]

เริ่มแผงตรวจสอบแบบเรียลไทม์ด้วยคำสั่งต่อไปนี้:

python -m dashboard.server --log-dir /tmp/swebench_qwen3_coder --port 8765

สำหรับรายละเอียดการกำหนดค่าการส่งต่อพอร์ตระยะไกลและการผูกของ Dashboard โปรดดูเอกสาร dashboard/README.md[3]

หนึ่ง วิเคราะห์โครงสร้างโครงการแบบพาโนรามา

ก่อนที่จะลงลึกในรายละเอียดโค้ด เรามาทำความเข้าใจตรรกะการจัดระเบียบโดยรวมของโครงการผ่านแผนผังไดเรกทอรีกันก่อน:

uni-agent/  
├── uni_agent/                  # แพ็คเกจ Python หลัก  
│   ├── agent_loop.py           # จุดเริ่มต้นแบบรวม: UniAgentLoop (เชื่อมต่อการฝึก verl)  
│   ├── async_logging.py        # การบันทึกแบบอะซิงโครนัส (แยกตาม run_id)  
│   ├── utils.py                # ยูทิลิตี้เช่น auto_await  
│   ├── interaction/            # ชั้นเอนจินการโต้ตอบ  
│   │   ├── interaction.py      #   วงจรการโต้ตอบหลายรอบ AgentInteraction  
│   │   ├── model.py            #   สิ่งที่เป็นนามธรรมของโมเดล (AgentChatModel / OpenAI Compatible)  
│   │   ├── env.py              #   สิ่งที่เป็นนามธรรมของสภาพแวดล้อม AgentEnv (การดำเนินการ)  
│   │   ├── tools_manager.py    #   การจัดการเครื่องมือและการแปลการเรียก  
│   │   └── tool_parser.py      #   ตัวแยกวิเคราะห์รูปแบบการเรียกเครื่องมือ (XML/Hermes)  
│   ├── tools/                  # การใช้งานเครื่องมือเฉพาะ  
│   │   ├── base.py             #   คลาสพื้นฐาน AbstractTool  
│   │   ├── registry.py         #   รีจิสทรีเครื่องมือ  
│   │   ├── execute_bash/       #   เครื่องมือดำเนินการ Bash  
│   │   ├── str_replace_editor/ #   เครื่องมือแก้ไขไฟล์  
│   │   ├── search/             #   เครื่องมือค้นหาโค้ด  
│   │   ├── search_arxiv/       #   ค้นหาบทความ arXiv  
│   │   ├── finish/             #   เครื่องหมายเสร็จสิ้น  
│   │   └── submit/             #   เครื่องหมายส่ง  
│   ├── deployment/             # แบ็กเอนด์การปรับใช้สภาพแวดล้อม  
│   │   ├── config.py           #   การกำหนดค่าการปรับใช้แบบรวม (ยูเนียนจำแนกประเภท)  
│   │   ├── remote_runtime.py   #   ไคลเอนต์ HTTP รันไทม์ระยะไกล  
│   │   ├── host/               #   โหมดดำเนินการโดยตรงบนโฮสต์  
│   │   ├── local/              #   แซนด์บ็อกซ์คอนเทนเนอร์ท้องถิ่น (Docker/Apptainer)  
│   │   ├── modal/              #   แพลตฟอร์มคลาวด์ Modal  
│   │   └── vefaas/             #   Volcengine FaaS  
│   └── reward/                 # โมดูลคำนวณรางวัล  
│       ├── base.py             #   คลาสพื้นฐาน AbstractRewardSpec  
│       ├── registry.py         #   รีจิสทรีฟังก์ชันรางวัล  
│       ├── swe_bench.py        #   รางวัลประเมิน SWE-Bench  
│       ├── swe_bench_live.py   #   รูปแบบ SWE-Bench Live  
│       ├── swe_rebench.py      #   รูปแบบ SWE-ReBench  
│       ├── r2e_gym.py          #   รางวัลฝึก R2E-Gym  
│       └── search.py           #   รางวัล Search Agent  
├── examples/                   # สคริปต์ตัวอย่าง  
│   ├── agent_env/              #   ตัวอย่างเริ่มต้นสภาพแวดล้อม  
│   ├── agent_interaction/      #   ตัวอย่างการโต้ตอบแบบขนาน  
│   ├── agent_train/            #   สคริปต์ฝึก RL  
│   ├── search_agent/           #   การฝึก Search Agent  
│   ├── search_arxiv/           #   ตัวอย่าง arXiv Agent  
│   └── data_preprocess/        #   การประมวลผลข้อมูลล่วงหน้า  
├── dashboard/                  # แผงตรวจสอบแบบเรียลไทม์ (JS/CSS/Python server)  
├── verl/                       # Git submodule → volcengine/verl  
├── docs/                       # ซอร์สโค้ดเอกสาร Sphinx  
└── pyproject.toml              # ข้อมูลเมตาของแพ็คเกจและการพึ่งพา  

สอง ภาพรวมสถาปัตยกรรมและกระแสข้อมูลหลัก

2.1 แนวคิดการออกแบบแบบแยกส่วนสามชั้น

การออกแบบโดยรวมของ Uni-Agent สามารถแสดงให้เห็นได้ผ่านแผนภาพสถาปัตยกรรมต่อไปนี้:

แผนภาพแสดงการแบ่งหน้าที่ของสามโมดูลหลักอย่างชัดเจน:

• Model (เอนจินการอนุมาน): โมดูลนี้ห่อหุ้มการโต้ตอบการอนุมานกับ Large Language Model (LLM) มีหน้าที่ดูแล Rollout Cache (ประกอบด้วยลำดับ Token, Mask และ Logprobs) ในโหมดฝึก มันสามารถเชื่อมต่อกับเซิร์ฟเวอร์ AsyncLLM ของ verl framework ได้อย่างราบรื่น ในขณะที่ในสถานการณ์การอนุมานบริสุทธิ์ มันเข้ากันได้กับ API ใดๆ ที่เป็นไปตามมาตรฐานอินเทอร์เฟซของ OpenAI
• Tool (ชั้นการรับรู้และการดำเนินการ): หน้าที่หลักของโมดูลนี้คือการแปลคำสั่งเรียกเครื่องมือ (tool call) ที่มีโครงสร้างซึ่งสร้างโดยโมเดลให้เป็นคำสั่ง bash ที่เฉพาะเจาะจง เครื่องมือเหล่านี้ถูกติดตั้งในคอนเทนเนอร์ที่รันในรูปแบบของสคริปต์ที่สามารถดำเนินการได้
• Env (แซนด์บ็อกซ์รันไทม์): โมดูลนี้สื่อสารกับคอนเทนเนอร์หรือแซนด์บ็อกซ์ระยะไกลผ่านโปรโตคอล SWE-ReX คุณสมบัติที่สำคัญคือความสามารถในการรองรับแบ็กเอนด์การปรับใช้สี่แบบและสลับไปมาระหว่างกันได้อย่างราบรื่น

2.2 การวิเคราะห์กระแสข้อมูลการโต้ตอบที่สมบูรณ์หนึ่งครั้ง

ภายใต้กรอบงานของ Uni-Agent การโต้ตอบของ Agent ที่สมบูรณ์หนึ่งครั้งจะเป็นไปตามกระแสข้อมูลที่ชัดเจนและสามารถวนซ้ำได้ กระบวนการทั้งหมดเริ่มต้นจากจุดเข้า UniAgentLoop ผ่านการวนซ้ำหลายขั้นตอนภายในโมดูล AgentInteraction และในที่สุดก็สร้างเอาต์พุตที่มีโครงสร้างซึ่งสามารถใช้สำหรับการฝึกเรียนรู้แบบเสริมกำลัง

ขั้นตอนเฉพาะมีดังนี้:
1. สอบถามโมเดล (Query Model): ระบบจะส่งประวัติการสนทนาปัจจุบันพร้อมกับคำแนะนำงานเป็นอินพุตไปยังโมดูล AgentChat Model
2. แยกวิเคราะห์การเรียกเครื่องมือ (Parse Tool Call): หากการตอบสนองที่ส่งกลับโดยโมเดลมีคำสั่งเรียกเครื่องมือ โมดูล AgentInteraction จะแยกวิเคราะห์เป็นคำขอ Tool Call ที่มีโครงสร้าง
3. ดำเนินการในสภาพแวดล้อม (Execute in Env): การเรียกเครื่องมือที่แยกวิเคราะห์แล้วจะถูกส่งต่อไปยัง AgentEnv เพื่อดำเนินการตามคำสั่งที่เกี่ยวข้อง (เช่น รันคำสั่ง bash) ในคอนเทนเนอร์ โฮสต์ระยะไกล หรือสภาพแวดล้อมท้องถิ่นจริง
4. สังเกตและผนวก (Observe & Append): หลังจากสภาพแวดล้อมดำเนินการเสร็จสิ้น ผลลัพธ์การดำเนินการ (เช่น เอาต์พุตมาตรฐาน ข้อความแสดงข้อผิดพลาด) จะถูกส่งกลับเป็นผลการสังเกต (Observation) ผลการสังเกตนี้จะถูกผนวกเข้ากับประวัติการสนทนาเพื่อใช้เป็นอินพุตสำหรับการ Query Model ในครั้งถัดไป

ขั้นตอนข้างต้นจะ形成一个วงจร วนซ้ำอย่างต่อเนื่องจนกว่างานจะเสร็จสมบูรณ์ (done) หรือถึงจำนวนรอบสูงสุดที่กำหนดไว้ เมื่อวงจรสิ้นสุดลง ระบบจะเข้าสู่ขั้นตอนต่อไปนี้:

• การประเมินรางวัล (RewardSpec): โมดูล RewardSpec จะให้คะแนนและประเมินกระบวนการโต้ตอบทั้งหมดของ Agent ตามเกณฑ์การประเมินที่กำหนดไว้ล่วงหน้าภายในสภาพแวดล้อม
• การจัดรูปแบบเอาต์พุต (AgentLoopOutput): สุดท้าย โมดูล AgentLoopOutput จะ打包ข้อมูลสำคัญต่างๆ เช่น prompt, response, mask และ logprobs จากกระบวนการโต้ตอบทั้งหมดให้เป็นเอาต์พุตมาตรฐาน เอาต์พุตนี้สามารถป้อนเข้าสู่กรอบงานฝึกเรียนรู้แบบเสริมกำลัง verl ได้โดยตรงเพื่อใช้ในการปรับแต่งโมเดลให้ดียิ่งขึ้น

มาดูเส้นทางการทำงานที่สมบูรณ์ของ Agent ตั้งแต่รับงานไปจนถึงสร้างข้อมูลการฝึก:

ผู้ใช้ Prompt (messages)
│
▼
┌─ prepare_rollout_cache ─┐ เข้ารหัส messages เป็นลำดับ token
│ prompt_ids = tokenize │ เริ่มต้น response_mask/logprobs ที่ว่างเปล่า
└─────────┬───────────────┘
│
┌─────▼─────┐
│ STEP 1 │◀───────────────────────────────────────┐
├───────────┤ │
│ model.query() │
│ → response_ids (mask=1, บันทึก logprobs) │
│ │
│ tools_manager.parse_action() │
│ → ดึงชื่อฟังก์ชัน + อาร์กิวเมนต์ │
│ │
│ tools_manager.get_tool_bash_command() │
│ → แปลงเป็นสตริงคำสั่ง bash │
│ │
│ env.run_action() │
│ → สตริง observation │
│ │
│ model.append_messages_to_rollout_cache() │
│ → ผนวกการเข้ารหัส observation (mask=0) │
│ │
│ done? ──No──────────────────────────────────────────┘
│ │
│ Yes
└───┬───┘
│
▼
┌─ reward_spec.compute_reward() ─┐ รันการทดสอบในคอนเทนเนอร์เดียวกัน
│ เช่น: ดำเนินการสคริปต์ทดสอบ SWE-Bench │ และแยกวิเคราะห์รายงานการทดสอบ
└────────────┬───────────────────┘
│
▼
┌─ convert_to_agent_output() ──┐ แยก prompt/response
│ ประกอบ AgentLoopOutput │ ใช้ mask_abnormal_exit
│ → ส่งออกให้ verl ทำ PPO/GSPO │ ตัดลำดับที่ยาวเกินไป
└──────────────────────────────┘

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

สาม จุดเริ่มต้นแบบรวม: ตรรกะการจัดเรียงของ UniAgentLoop

3.1 การควบคุมการทำงานพร้อมกันและการแยกทรัพยากร

UniAgentLoop ทำหน้าที่เป็นสะพานเชื่อมระหว่าง Uni-Agent และกรอบงานฝึก verl มันสืบทอดมาจาก AgentLoopBase ของ verl และถูกเรียกใช้ในช่วง rollout ของการฝึกเรียนรู้แบบเสริมกำลัง:

# ที่มา: uni_agent/agent_loop.py  
class UniAgentLoop(AgentLoopBase):  
_semaphore: asyncio.Semaphore | None = None  # แชร์ในระดับคลาส  

async def run(self, sampling_params: dict[str, Any], **kwargs) -> AgentLoopOutput:  
config_dict = self._init_config(sampling_params, **kwargs)  
# การทำงานพร้อมกันทั่วโลกเฉลี่ยตามจำนวน worker  
global_concurrent = config_dict.get("concurrency", 512)  
num_workers = self.config.actor_rollout_ref.rollout.agent.num_workers  
worker_concurrent = max(global_concurrent // num_workers, 1)  
if UniAgentLoop._semaphore is None:  
UniAgentLoop._semaphore = asyncio.Semaphore(worker_concurrent)  

ที่นี่ _semaphore ถูกกำหนดให้เป็นตัวแปรคลาส อินสแตนซ์ UniAgentLoop ทั้งหมดใช้สัญญาณเดียวกัน สมมติว่าขีดจำกัดการทำงานพร้อมกันทั่วโลกตั้งไว้ที่ 512 และมี 4 worker ในระบบ ดังนั้นแต่ละ worker สามารถรันสภาพแวดล้อมได้สูงสุด 128 ตัวพร้อมกัน กลไกนี้ป้องกันข้อผิดพลาด OOM หรือความล้มเหลวในการจัดตารางคอนเทนเนอร์ที่เกิดจากการโอเวอร์โหลดของทรัพยากรได้อย่างมีประสิทธิภาพ

3.2 ขั้นตอนการดำเนินการหลัก

# ที่มา: uni_agent/agent_loop.py  
async with self._semaphore:  
try:  
await self.env.start()                           # เริ่มแซนด์บ็อกซ์  
self.chat_model.set_tools_schemas(...)           # ใส่ tool schema  
await self.env.install_tools(self.tools_manager.tools)  # ติดตั้งเครื่องมือในคอนเทนเนอร์  

interaction_result = await self.interaction.run() # วงจรการโต้ตอบหลัก  

# คำนวณรางวัลภายในสภาพแวดล้อม (สภาพแวดล้อมยังไม่ปิด)  
if self.reward_spec is not None:  
reward_score, _ = await self.reward_spec.compute_reward(  
interaction_result=interaction_result,  
)  
interaction_result["reward_score"] = reward_score  

self._save_interaction_result(interaction_result)  
output = await self.convert_to_agent_output(interaction_result)  
except Exception as e:  
output = await self._build_empty_agent_output(exit_reason="agent_loop_failed")  
finally:  
await self.env.close()  # ปิดสภาพแวดล้อมไม่ว่าจะสำเร็จหรือล้มเหลว  
return output  

บล็อก finally ที่นี่รับประกันว่าสภาพแวดล้อมจะถูกทำความสะอาดอย่างแน่นอน — ในสถานการณ์การทำงานพร้อมกันระดับพัน คอนเทนเนอร์ที่ไม่ได้ปล่อยจะทำให้ทรัพยากรคลัสเตอร์หมดลงอย่างรวดเร็ว

3.3 การลดระดับอย่างสง่างามสำหรับเส้นทางที่ผิดปกติ

เมื่อการโต้ตอบของ Agent ออกผิดปกติ ระบบจะไม่ทิ้งข้อมูลนี้ แต่จะสร้าง “เส้นทางว่าง” และใส่เครื่องหมายปิดบัง:

# ที่มา: uni_agent/agent_loop.py  
async def _build_empty_agent_output(self, exit_reason: str) -> AgentLoopOutput:  
# เติมด้วย pad_token หรือ eos_token  
dummy_token_id = getattr(self.tokenizer, "pad_token_id", None)  
if dummy_token_id is None:  
dummy_token_id = getattr(self.tokenizer, "eos_token_id", None)  

extra_fields["traj_masked"] = 1          # ข้ามเส้นทางนี้ระหว่างการฝึก  
extra_fields["traj_exit_reason"] = exit_reason  

return AgentLoopOutput(  
response_ids=[dummy_token_id] * dummy_response_length,  
response_mask=[0] * dummy_response_length,  # ศูนย์ทั้งหมด → ไม่เข้าร่วม gradient  
reward_score=0,  
...  
)  

ด้วยวิธีนี้ ความสม่ำเสมอของรูปร่าง batch การฝึกจึงได้รับการรับประกัน — ไม่จำเป็นต้องมีการ padding แบบไดนามิกหรือความผันผวนของขนาด batch

สี่ เอนจินการโต้ตอบ: วงจรหลายรอบของ AgentInteraction

4.1 โมเดลข้อมูล StepOutput

แต่ละขั้นตอนการโต้ตอบจะสร้าง StepOutput ที่มีโครงสร้าง:

# ที่มา: uni_agent/interaction/interaction.py  
class StepOutput(BaseModel):  
step_idx: int  
response: str = ""        # เอาต์พุตดิบของโมเดล  
thought: str = ""         # เนื้อหาความคิด (ข้อความก่อน tool_call)  
action: str = ""          # คำสั่ง bash ที่แปลงแล้ว  
observation: str = ""     # สิ่งที่สภาพแวดล้อมส่งกลับ  
execution_time: float | None = None  
done: bool = False  
exit_reason: str = ""     # finished/token_limit/timeout_error/...

### 4.2 สายการผลิตห้าขั้นตอนของการโต้ตอบทีละขั้นตอน

```python
# ที่มา: uni_agent/interaction/interaction.py  
async def step(self, step_idx: int):  
step_output = StepOutput(step_idx=step_idx)  

# ── ขั้นตอนที่ 1: การอนุมานโมเดล ──  
model_output, rollout_cache, generation_info = await self.model.query(  
messages=self.messages, rollout_cache=self.rollout_cache,  
)  

# ── ขั้นตอนที่ 2: แยกวิเคราะห์ Tool Call ──  
self.messages.append({"role": "assistant", "content": model_output})  
structured_tool_calls = self.rollout_cache.get("extra_fields", {}).get("last_tool_calls", [])  
if structured_tool_calls:  
# รูปแบบ tool_calls ดั้งเดิมของ OpenAI (สถานการณ์การอนุมาน)  
content, tool_calls = await self.tools_manager.parse_structured_action(  
content=model_output, tool_calls_data=structured_tool_calls,  
)  
else:  
# แยกวิเคราะห์รูปแบบ XML/Hermes จากข้อความ (สถานการณ์การฝึก)  
content, tool_calls = await self.tools_manager.parse_action(model_output=model_output)  

# ── ขั้นตอนที่ 3: แปลงเป็นคำสั่ง Bash ──  
tool_call = tool_calls[0]  
action_cmd = self.tools_manager.get_tool_bash_command(tool_call)  

# ── ขั้นตอนที่ 4: ดำเนินการในสภาพแวดล้อม ──  
observation = await self.env.run_action(action_cmd, action_timeout=self.action_timeout)  
tool_message = {"role": "tool", "content": observation}  
self.messages.append(tool_message)  
self.rollout_cache = await self.model.append_messages_to_rollout_cache(  
[tool_message], self.rollout_cache  
)  

# ── ขั้นตอนที่ 5: ตัดสินเงื่อนไขการสิ้นสุด ──  
if tool_call.function.name in ["finish", "submit"]:  
step_output.done = True  
step_output.exit_reason = "finished"  

กระบวนการทั้งหมดสามารถแสดงได้ดังนี้:

┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐
│  Model   │───▶│  Parse   │───▶│ Translate│───▶│ Execute  │───▶│  Judge   │
│  Query   │    │ ToolCall │    │ to Bash  │    │  in Env  │    │   Done?  │
└──────────┘    └──────────┘    └──────────┘    └──────────┘    └──────────┘
▲                                                               │
└───────────────────── loop if not done ────────────────────────┘

4.3 กลไกงบประมาณหมดเวลา

การออกแบบที่ชาญฉลาดสำหรับการทนทานต่อข้อผิดพลาดปรากฏอยู่ใน timeout_budget:

# ที่มา: uni_agent/interaction/interaction.py  
except ActionTimeoutError as e:  
step_output.exit_reason = "timeout_error"  
if self.timeout_budget > 0:  
self.timeout_budget -= 1  
return step_output  # ดำเนินการขั้นตอนต่อไป (ให้โอกาสโมเดลแก้ไข)  
else:  
step_output.done = True  # หมดเวลาหมดแล้ว บังคับยุติ  
return step_output  

ระบบอนุญาตให้เกิดการหมดเวลาได้ 3 ครั้งโดยค่าเริ่มต้น — Agent จะได้รับการตอบกลับข้อผิดพลาดหลังจากหมดเวลาสองสามครั้งแรก ทำให้มีโอกาสเลือกคำสั่งที่เร็วกว่า เฉพาะเมื่อการหมดเวลาเกิดขึ้นติดต่อกันหลายครั้ง ระบบจะตัดสินว่า “ไม่สามารถกู้คืนได้” และยุติการโต้ตอบ กลไกนี้ช่วยเพิ่มอัตราความสำเร็จของงานระยะยาวได้อย่างมีนัยสำคัญ

unsetunsetห้า Rollout Cache: การสะสมข้อมูลการฝึกแบบเรียลไทม์unsetunset

5.1 โครงสร้างข้อมูลหลัก

Rollout Cache เป็นนวัตกรรมหลักที่ทำให้ Uni-Agent แตกต่างจากเฟรมเวิร์ก Agent อื่นๆ ทั้งหมด มันเติบโตแบบเรียลไทม์ระหว่างการโต้ตอบ โครงสร้างข้อมูลมีดังนี้:

rollout_cache = {

"request_id":       "uuid-...",

"prompt_ids":       [101, 2035, 8841, ...],   # ลำดับ token ที่ต่อกันอย่างต่อเนื่อง

"response_mask":    [0, 0, 0, ..., 1, 1, 1, ..., 0, 0, ...],  # อันไหนที่โมเดลสร้าง

"response_logprobs": [0.0, 0.0, ..., -0.23, -1.1, ..., 0.0, ...],

"routed_experts":   [...] | None,   # ข้อมูลเส้นทาง MoE (ไม่บังคับ)

"metrics":          {"generate_sequences": 1.23, ...},

"extra_fields":     {"traj_masked": 0, ...},

}

ใช้ไทม์ไลน์เพื่อทำความเข้าใจบทบาทของ mask:
ลำดับ Token:  [system_prompt] [user_msg] [asst_response_1] [tool_obs_1] [asst_response_2] ...
Mask:         0 0 0 0 0 0 0   0 0 0 0    1 1 1 1 1 1 1 1   0 0 0 0 0    1 1 1 1 1 1 1 ...
╰─── prompt ──╯            ╰── การตัดสินใจของโมเดล ──╯  ╰─ สภาพแวดล้อมส่งกลับ

⚠️ หมายเหตุ: เนื้อหาได้รับการแปลโดย AI และตรวจสอบโดยมนุษย์ หากมีข้อผิดพลาดโปรดแจ้ง

    

        
☕ สนับสนุนค่ากาแฟทีมงาน
        
หากคุณชอบบทความนี้ สามารถสนับสนุนเราได้ผ่าน PromptPay
        

            
        
        

            SCAN TO PAY WITH ANY BANK
        
    
                                                        
本文来自网络搜集，不代表คลื่นสร้างอนาคต立场，如有侵权，联系删除。转载请注明出处：https://www.itsolotime.com/th/archives/35466