ยินดีต้อนรับสู่โลกของ Multi-Agent System! วันนี้เราจะมาสอนคุณวิธีตั้งค่า AutoGen Agents แบบกระจาย (Distributed) โดยใช้ OpenAI Compatible Gateway และ MCP (Model Context Protocol) สำหรับการเรียกใช้ Tools ต่างๆ แม้คุณจะไม่มีประสบการณ์ API มาก่อนเลยก็ตาม
AutoGen คืออะไร?
AutoGen เป็น Framework จาก Microsoft ที่ช่วยให้คุณสร้างระบบ Agent หลายตัวที่สามารถคุยกันและทำงานร่วมกันได้ ลองนึกภาพว่าคุณมีพนักงาน AI หลายคน ทำหน้าที่แตกต่างกัน เช่น คนหนึ่งค้นหาข้อมูล อีกคนเขียนรายงาน และอีกคนตรวจสอบความถูกต้อง พวกเขาสามารถส่งข้อความถามความคิดเห็น และทำงานร่วมกันแบบอัตโนมัติ
MCP (Model Context Protocol) คืออะไร?
MCP เป็นมาตรฐานการสื่อสารที่ช่วยให้ AI Agent สามารถเรียกใช้ Tools หรือ Functions ต่างๆ ได้อย่างเป็นมาตรฐาน คล้ายกับการที่คุณมีรีโมทคอนโทรล一台ที่ควบคุมอุปกรณ์หลายเครื่องได้ แทนที่จะต้องมีรีโมทแยกกัน
เริ่มต้นติดตั้ง Environment
ก่อนอื่นเราต้องเตรียมเครื่องมือให้พร้อมกันก่อน
# สร้าง Virtual Environment (สภาพแวดล้อมแยกสำหรับโปรเจกต์นี้)
python -m venv autogen_env
เปิดใช้งาน Virtual Environment
สำหรับ Windows
autogen_env\Scripts\activate
สำหรับ macOS/Linux
source autogen_env/bin/activate
ติดตั้ง AutoGen Studio และ Dependencies
pip install autogen-agentchat autogen-ext[mcp]
ติดตั้ง MCP SDK (มาตรฐานใหม่)
pip install "mcp[cli]" anthropicตั้งค่า HolySheep AI เป็น Gateway
HolySheep AI เป็น OpenAI Compatible Gateway ที่รองรับโมเดลหลากหลาย เช่น GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash และ DeepSeek V3.2 ในราคาที่ประหยัดมาก อัตราแลกเปลี่ยน ¥1=$1 คิดเป็นการประหยัดมากกว่า 85% เมื่อเทียบกับการใช้งานโดยตรง รองรับการชำระเงินผ่าน WeChat และ Alipay มีความหน่วงต่ำกว่า 50ms และให้เครดิตฟรีเมื่อลงทะเบียน คุณสามารถ สมัครที่นี่ ได้เลย
ดาวน์โหลด MCP Servers ที่จำเป็น
# ติดตั้ง MCP Server สำหรับ Filesystem
pip install mcp-server-filesystem
ติดตั้ง MCP Server สำหรับ Web Fetch
pip install mcp-server-fetchสร้างไฟล์ Configuration หลัก
ต่อไปเราจะสร้างไฟล์ config.py ที่เป็นหัวใจหลักของระบบ ไฟล์นี้จะกำหนดว่า Agent แต่ละตัวจะใช้โมเดลอะไร และเรียกใช้ Tools อะไรได้บ้าง
"""
config.py - ไฟล์ตั้งค่าหลักสำหรับ AutoGen Distributed Agents
"""
import os
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import SelectorGroupChat
from autogen_ext.tools.mcp import McpWorkforce
============================================
1. ตั้งค่า HolySheep API Configuration
============================================
กำหนด Base URL สำหรับ OpenAI Compatible Gateway
BASE_URL = "https://api.holysheep.ai/v1"
กำหนด API Key ของคุณ (นำมาจาก HolySheep Dashboard)
อย่าลืมเปลี่ยน YOUR_HOLYSHEEP_API_KEY เป็น Key จริงของคุณ!
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
============================================
2. กำหนด Model Configuration
============================================
def get_model_config(model_name: str):
"""Function สำหรับสร้าง Model Configuration"""
return {
"model": model_name,
"api_key": HOLYSHEEP_API_KEY,
"base_url": BASE_URL,
"temperature": 0.7,
"max_tokens": 4096
}
กำหนดโมเดลที่จะใช้งาน (ราคาต่อล้าน Tokens)
MODEL_CATALOG = {
"gpt-4.1": {
"provider": "OpenAI Compatible",
"price_per_mtok": 8.00, # $8/MTok
"description": "โมเดลทั่วไป เหมาะกับงานหลากหลาย"
},
"claude-sonnet-4.5": {
"provider": "Anthropic Compatible",
"price_per_mtok": 15.00, # $15/MTok
"description": "โมเดลสำหรับงานที่ต้องการความแม่นยำสูง"
},
"gemini-2.5-flash": {
"provider": "Google Compatible",
"price_per_mtok": 2.50, # $2.50/MTok
"description": "โมเดลเร็ว ประหยัด เหมาะกับงานทั่วไป"
},
"deepseek-v3.2": {
"provider": "DeepSeek Compatible",
"price_per_mtok": 0.42, # $0.42/MTok - ประหยัดที่สุด!
"description": "โมเดลคุ้มค่าที่สุด ราคาเพียง $0.42/MTok"
}
}
============================================
3. สร้าง MCP Workload (เครื่องมือสำหรับ Agent)
============================================
async def create_mcp_workforce():
"""สร้าง MCP Workforce ที่รวม Tools ต่างๆ เข้าด้วยกัน"""
workforce = McpWorkforce(
workers=[
# Filesystem Worker - สำหรับอ่าน/เขียนไฟล์
{
"type": "filesystem",
"config": {
"allowed_directories": ["./workspace"]
}
},
# Fetch Worker - สำหรับดึงข้อมูลจากเว็บ
{
"type": "fetch",
"config": {
"allowed_domains": ["*"]
}
}
]
)
return workforce
============================================
4. สร้าง Agent Definitions
============================================
def create_researcher_agent():
"""สร้าง Researcher Agent - ทำหน้าที่ค้นหาข้อมูล"""
return AssistantAgent(
name="researcher",
model_client={
"provider": "openai",
"config": get_model_config("deepseek-v3.2") # ใช้โมเดลประหยัดสำหรับค้นหา
},
system_message="""
คุณคือ Researcher Agent - ผู้เชี่ยวชาญด้านการค้นหาข้อมูล
หน้าที่ของคุณ:
1. ค้นหาข้อมูลที่เกี่ยวข้องจากแหล่งต่างๆ
2. สรุปข้อมูลสำคัญให้กระชับ
3. ใช้ MCP Tools เพื่อดึงข้อมูลจากเว็บไซต์
เมื่อทำเสร็จแล้ว ส่งข้อมูลให้ Writer Agent เพื่อเขียนรายงาน
""",
tools=["fetch"] # อนุญาตให้ใช้ fetch tool
)
def create_writer_agent():
"""สร้าง Writer Agent - ทำหน้าที่เขียนรายงาน"""
return AssistantAgent(
name="writer",
model_client={
"provider": "openai",
"config": get_model_config("gemini-2.5-flash") # ใช้โมเดลเร็วสำหรับเขียน
},
system_message="""
คุณคือ Writer Agent - ผู้เชี่ยวชาญด้านการเขียน
หน้าที่ของคุณ:
1. เขียนรายงานหรือบทความจากข้อมูลที่ได้รับ
2. จัดรูปแบบให้อ่านง่าย
3. บันทึกไฟล์ลงใน workspace
เมื่อเขียนเสร็จแล้ว ส่งให้ Editor Agent ตรวจสอบ
""",
tools=["filesystem"] # อนุญาตให้ใช้ filesystem tool
)
def create_editor_agent():
"""สร้าง Editor Agent - ทำหน้าที่ตรวจสอบคุณภาพ"""
return AssistantAgent(
name="editor",
model_client={
"provider": "openai",
"config": get_model_config("claude-sonnet-4.5") # ใช้โมเดลคุณภาพสูงสำหรับตรวจ
},
system_message="""
คุณคือ Editor Agent - ผู้เชี่ยวชาญด้านการตรวจสอบคุณภาพ
หน้าที่ของคุณ:
1. ตรวจสอบความถูกต้องของเนื้อหา
2. แก้ไขข้อผิดพลาดทางไวยากรณ์
3. ให้คำแนะนำปรับปรุง
เมื่อตรวจสอบเสร็จแล้ว รายงานผลลัพธ์สุดท้าย
""",
tools=["filesystem", "fetch"]
)
print("Configuration สร้างเรียบร้อยแล้ว!")
print(f"ราคาโมเดลต่อล้าน Tokens:")
for model, info in MODEL_CATALOG.items():
print(f" - {model}: ${info['price_per_mtok']}")สร้าง Distributed Team และ Orchestration
ต่อไปเราจะสร้างไฟล์ team.py ที่กำหนดวิธีการทำงานร่วมกันของ Agents ทั้งหมด และการส่งต่องานกัน
"""
team.py - กำหนดโครงสร้าง Distributed Agent Team
"""
import asyncio
from autogen_agentchat.teams import SelectorGroupChat, RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination, MaxMessageTermination
from config import (
create_researcher_agent,
create_writer_agent,
create_editor_agent,
get_model_config,
HOLYSHEEP_API_KEY,
BASE_URL
)
class DistributedAgentTeam:
"""
คลาสสำหรับจัดการ Distributed Agent Team
มีการกระจายงานและประสานงานระหว่าง Agent หลายตัว
"""
def __init__(self):
self.agents = []
self.team = None
async def setup(self):
"""ตั้งค่า Team และ Agent ทั้งหมด"""
# สร้าง Agents ทั้ง 3 ตัว
researcher = create_researcher_agent()
writer = create_writer_agent()
editor = create_editor_agent()
self.agents = [researcher, writer, editor]
# กำหนดเงื่อนไขการหยุดทำงาน
termination_conditions = [
TextMentionTermination("งานเสร็จสมบูรณ์"), # หยุดเมื่อ Editor พูดว่างานเสร็จ
MaxMessageTermination(max_messages=20) # หยุดเมื่อส่งข้อความครบ 20 ครั้ง
]
# สร้าง Selector Group Chat - เลือก Agent ที่เหมาะสมที่สุดในแต่ละขั้น
self.team = SelectorGroupChat(
participants=self.agents,
termination_condition=termination_conditions[0],
selector_prompt="""
เลือก Agent ที่เหมาะสมที่สุดสำหรับขั้นตอนถัดไป:
ถ้ายังไม่ได้เริ่มทำงาน → เลือก 'researcher' เพื่อค้นหาข้อมูล
ถ้าได้ข้อมูลแล้ว → เลือก 'writer' เพื่อเขียนรายงาน
ถ้าเขียนรายงานเสร็จแล้ว → เลือก 'editor' เพื่อตรวจสอบ
ถ้าตรวจสอบเสร็จแล้ว → พูดว่า 'งานเสร็จสมบูรณ์'
"""
)
print("✅ Team Setup เสร็จสมบูรณ์")
print(f" มี Agents: {[a.name for a in self.agents]}")
async def run_task(self, task: str):
"""รันงานผ่าน Team"""
if self.team is None:
await self.setup()
print(f"📋 เริ่มงาน: {task}")
# รัน Task Stream เพื่อดูขั้นตอนการทำงาน
task_result = None
async for message in self.team.run_stream(task=task):
# แสดงข้อความจาก Agent แต่ละตัว
if hasattr(message, 'source'):
print(f" [{message.source}] {message.content[:100]}...")
task_result = message
print("✅ งานเสร็จสมบูรณ์")
return task_result
async def cleanup(self):
"""ทำความสะอาด Resources"""
for agent in self.agents:
if hasattr(agent, 'stop'):
agent.stop()
print("🧹 Cleanup เสร็จสมบูรณ์")
async def main():
"""Entry Point หลัก"""
print("=" * 50)
print("AutoGen Distributed Agent Team")
print("Using HolySheep AI Gateway")
print(f"Base URL: {BASE_URL}")
print("=" * 50)
# สร้าง Team
team = DistributedAgentTeam()
try:
# ตั้งค่า Team
await team.setup()
# รันงานตัวอย่าง
sample_task = """
ค้นหาข้อมูลเกี่ยวกับประโยชน์ของ AI Agents ในธุรกิจ
และเขียนบทความสั้น 200 คำ
"""
result = await team.run_task(sample_task)
print("\n📊 ผลลัพธ์:")
print(result)
finally:
# ทำความสะอาดเสมอ
await team.cleanup()
รันเมื่อ execute ไฟล์นี้โดยตรง
if __name__ == "__main__":
asyncio.run(main())สร้าง MCP Tool Server ของตัวเอง
บางครั้งคุณอาจต้องการสร้าง MCP Tool Server ของตัวเองเพื่อให้ Agent สามารถเรียกใช้งานเฉพาะทางได้ ตัวอย่างนี้จะสร้าง Tool Server สำหรับจัดการข้อมูลลูกค้า
"""
mcp_server.py - สร้าง Custom MCP Tool Server
"""
from mcp.server import Server
from mcp.server.stdio import stdio_server
from pydantic import AnyUrl
import asyncio
import json
สร้าง Server Instance
server = Server("customer-service-tools")
ฐานข้อมูลลูกค้าตัวอย่าง (ในทางปฏิบัติควรใช้ Database จริง)
CUSTOMER_DB = {
"C001": {"name": "สมชาย ใจดี", "plan": "premium", "credit": 5000},
"C002": {"name": "สมหญิง รักสบาย", "plan": "basic", "credit": 1000},
"C003": {"name": "วิชัย มั่นคง", "plan": "enterprise", "credit": 50000}
}
@server.list_tools()
async def list_tools():
"""ประกาศ Tools ที่ Server นี้มีให้ใช้งาน"""
return [
{
"name": "get_customer",
"description": "ดึงข้อมูลลูกค้าจากรหัส",
"inputSchema": {
"type": "object",
"properties": {
"customer_id": {
"type": "string",
"description": "รหัสลูกค้า เช่น C001"
}
},
"required": ["customer_id"]
}
},
{
"name": "check_credit",
"description": "ตรวจสอบเครดิตคงเหลือของลูกค้า",
"inputSchema": {
"type": "object",
"properties": {
"customer_id": {
"type": "string",
"description": "รหัสลูกค้า"
}
},
"required": ["customer_id"]
}
},
{
"name": "update_credit",
"description": "เพิ่มเครดิตให้ลูกค้า",
"inputSchema": {
"type": "object",
"properties": {
"customer_id": {"type": "string"},
"amount": {"type": "number", "description": "จำนวนที่จะเพิ่ม"}
},
"required": ["customer_id", "amount"]
}
}
]
@server.call_tool()
async def call_tool(name: str, arguments: dict):
"""Implement การทำงานจริงของแต่ละ Tool"""
customer_id = arguments.get("customer_id")
if name == "get_customer":
if customer_id in CUSTOMER_DB:
return {"status": "success", "data": CUSTOMER_DB[customer_id]}
else:
return {"status": "error", "message": f"ไม่พบลูกค้า {customer_id}"}
elif name == "check_credit":
if customer_id in CUSTOMER_DB:
customer = CUSTOMER_DB[customer_id]
return {
"status": "success",
"customer_id": customer_id,
"credit": customer["credit"],
"plan": customer["plan"]
}
else:
return {"status": "error", "message": f"ไม่พบลูกค้า {customer_id}"}
elif name == "update_credit":
amount = arguments.get("amount", 0)
if customer_id in CUSTOMER_DB:
CUSTOMER_DB[customer_id]["credit"] += amount
return {
"status": "success",
"message": f"เพิ่มเครดิต {amount} ให้ {customer_id} เรียบร้อย",
"new_credit": CUSTOMER_DB[customer_id]["credit"]
}
else:
return {"status": "error", "message": f"ไม่พบลูกค้า {customer_id}"}
return {"status": "error", "message": "Unknown tool"}
async def main():
"""รัน MCP Server"""
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
server.create_initialization_options()
)
if __name__ == "__main__":
print("🛠️ MCP Customer Service Server พร้อมใช้งาน")
asyncio.run(main())รันระบบ Distributed Agents
ตอนนี้ทุกอย่างพร้อมแล้ว มาดูวิธีการรันระบบกัน
"""
main.py - Entry Point หลักสำหรับรันระบบ Distributed Agents
"""
import asyncio
import os
from team import DistributedAgentTeam
============================================
ตั้งค่า Environment Variables
============================================
วิธีที่ 1: ตั้งค่าผ่าน Environment Variable (แนะนำ)
os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
หรือวิธีที่ 2: ใส่ Key ตรงใน config.py (ไม่แนะนำสำหรับ Production)
async def run_demo():
"""
Demo การทำงานของ Distributed Agent System
"""
print("=" * 60)
print("🚀 AutoGen Distributed Agent with MCP Tools")
print("=" * 60)
# สร้าง Team Instance
team = DistributedAgentTeam()
try:
# ตั้งค่า Team
await team.setup()
# ==========================================
# งานที่ 1: ค้นหาและเขียนรายงาน
# ==========================================
task1 = """
ค้นหาข้อมูลเกี่ยวกับเทคโนโลยี AI ในปี 2026
และเขียนบทความสรุปพร้อมบันทึกลงไฟล์ ai_trends_2026.md
"""
print("\n📝 งานที่ 1: ค้นหาและเขียนบทความ")
await team.run_task(task1)
# ==========================================
# งานที่ 2: วิเคราะห์ข้อมูล
# ==========================================
task2 = """
วิเคราะห์แนวโน้ม AI ในไฟล์ ai_trends_2026.md
และสร้างรายงานสรุป executive summary
"""
print("\n📝 งานที่ 2: วิเคราะห์ข้อมูล")
await team.run_task(task2)
except Exception as e:
print(f"❌ เกิดข้อผิดพลาด: {e}")
finally:
# ทำความสะอาด Resources เสมอ
await team.cleanup()
async def run_with_custom_mcp_tools():
"""
Demo การใช้ Custom MCP Tools กับ Agents
"""
from autogen_ext.tools.mcp import McpWorkforce
from autogen_agentchat.agents import AssistantAgent
print("\n" + "=" * 60)
print("🔧 ทดสอบ Custom MCP Tools")
print("=" * 60)
# สร้าง MCP Workforce ที่เชื่อมต่อกับ Custom Server ของเรา
# หรือจาก MCP Registry ที่มีอยู่แล้ว
# ตัวอย่าง: ใช้ Math Tool จาก MCP Registry
mcp_workforce = McpWorkforce(
workers=[
{"type": "math"} # MCP Tool สำหรับคำนวณทางคณิตศาสตร์
]
)
# สร้าง Agent ที่ใช้ Math Tool
math_agent = AssistantAgent(
name="math_expert",
model_client={
"provider": "openai",
"config": {
"model": "deepseek-v3.2",
"api_key": os.environ.get("HOLYSHEEP_API_KEY"),
"base_url": "https://api.holysheep.ai/v1"
}
},
tools=mcp_workforce.tools
)
# ทดสอบถามคำถามคณิตศาสตร์
result = await math_agent.run(task="คำนวณ 12345 * 67890 = ?")
print("📊 ผลลัพธ์:")
print(result)
async def cost_estimation_demo():
"""
Demo การประมาณการค่าใช้จ่าย
"""
from config import MODEL_CATALOG
print("\n" + "=" * 60)
print("💰 การประมาณการค่าใช้จ่าย")
print("=" * 60)
# สมมติว่าใช้งา