Pyspur: สร้าง AI Agent โอเพนซอร์ส

Pyspur คืออะไร?

Pyspur คือแพลตฟอร์มโอเพนซอร์สที่ออกแบบมาเพื่อเร่งการพัฒนา AI agent โดยการจัดเตรียมสภาพแวดล้อมแบบภาพที่ใช้โหนด ช่วยให้นักพัฒนาสามารถสร้าง ดีบัก และปรับใช้เวิร์กโฟลว์ AI ที่ซับซ้อนได้โดยการเชื่อมต่อส่วนประกอบแบบโมดูลาร์บนพื้นที่ทำงานแบบลากและวาง

Drag & Drop to build AI Agents.

And deploy to API in one click with Nocode.

100% Opensource. pic.twitter.com/iPEJEzIHVI

— Shubham Saboo (@Saboo_Shubham_) December 19, 2024

ปัญหาหลักที่ Pyspur แก้ไขคือการขาดความโปร่งใสและรอบการทำงานที่ช้าซึ่งพบได้ทั่วไปในการพัฒนา AI ช่วยจัดการกับ "prompt hell" และ "workflow blindspots" โดยให้นักพัฒนาสามารถตรวจสอบอินพุตและเอาต์พุตของทุกขั้นตอนในตรรกะของ agent ได้แบบเรียลไทม์ แพลตฟอร์มนี้มีการรองรับรูปแบบขั้นสูงในตัว เช่น Retrieval-Augmented Generation (RAG) อนุญาตให้มีการหยุดชะงักแบบ human-in-the-loop และสามารถปรับใช้เวิร์กโฟลว์ใดๆ ให้เป็น API ที่พร้อมสำหรับการใช้งานจริงได้ด้วยการคลิกเพียงครั้งเดียว ท้ายที่สุด Pyspur ช่วยให้นักพัฒนาสร้างระบบ AI ที่น่าเชื่อถือและดีบักได้ง่ายขึ้นและเร็วขึ้น

We’re thrilled to share that MCP support is coming soon to Apidog! 🚀

Apidog MCP Server lets you feed API docs directly to Agentic AI, supercharging your vibe coding experience! Whether you're using Cursor, Cline, or Windsurf - it'll make your dev process faster and smoother.… pic.twitter.com/ew8U38mU0K

— Apidog (@ApidogHQ) March 19, 2025

มาเริ่มกันเลย!

1. การตั้งค่าสภาพแวดล้อม

เลือกตัวเลือกการตั้งค่าที่เหมาะสมกับวัตถุประสงค์ของคุณมากที่สุด สำหรับการพัฒนาและการทดลองในเครื่อง การติดตั้ง pip ก็เพียงพอแล้ว สำหรับระบบที่ปรับขนาดได้หรือระบบการผลิต ขอแนะนำให้ใช้การตั้งค่าแบบ Docker เนื่องจากให้สภาพแวดล้อมแบบคอนเทนเนอร์ที่สามารถสร้างซ้ำได้ พร้อมด้วยอินสแตนซ์ PostgreSQL โดยเฉพาะ

ตัวเลือก A: การติดตั้ง pip ในเครื่อง

ข้อกำหนดเบื้องต้น: Python 3.11+

ติดตั้งจาก PyPI:

pip install pyspur

เริ่มต้นไดเรกทอรีโครงการ: คำสั่งนี้จะสร้างโครงร่างโครงการ รวมถึงไฟล์ .env สำหรับการกำหนดค่า

pyspur init my-pyspur-project && cd my-pyspur-project

เปิดเซิร์ฟเวอร์: แฟล็ก --sqlite จะสั่งให้ Pyspur ใช้ฐานข้อมูล SQLite ในเครื่อง ซึ่งช่วยลดการพึ่งพาเซิร์ฟเวอร์ฐานข้อมูลแยกต่างหาก

pyspur serve --sqlite

เข้าถึง UI: ไปที่ http://localhost:6080 ในเบราว์เซอร์ของคุณ

ตัวเลือก B: การตั้งค่า Docker

ข้อกำหนดเบื้องต้น: Docker Engine

เรียกใช้สคริปต์การตั้งค่า: คำสั่งนี้จะดาวน์โหลดและรันสคริปต์เชลล์ที่โคลน repository ของ Pyspur กำหนดค่าไฟล์ docker-compose.dev.yml และเปิดใช้งานสแต็กแอปพลิเคชัน (ส่วนหน้า, ส่วนหลัง, ฐานข้อมูล)

curl -fsSL https://raw.githubusercontent.com/PySpur-com/pyspur/main/start_pyspur_docker.sh | bash -s pyspur-project

เข้าถึง UI: ไปที่ http://localhost:6080

2. มาสร้างเวิร์กโฟลว์ด้วย Pyspur กันเถอะ

แทนที่จะสร้างตั้งแต่เริ่มต้น เราจะโหลดและวิเคราะห์เทมเพลต Pyspur ที่มีอยู่ แนวทางนี้ช่วยให้เห็นภาพเวิร์กโฟลว์ที่ไม่ซับซ้อนได้จริง

โหลดเทมเพลต:

• บนแดชบอร์ด Pyspur คลิก "New Spur"
• ในหน้าต่างโมดัลที่ปรากฏขึ้น ให้เลือกแท็บ "Templates"
• เลือกเทมเพลต "Joke Generator" พื้นที่ทำงานจะถูกเติมด้วยเวิร์กโฟลว์ที่สร้างไว้ล่วงหน้า

การวิเคราะห์เวิร์กโฟลว์:
เวิร์กโฟลว์นี้ออกแบบมาเพื่อสร้างเรื่องตลกแล้วปรับปรุงให้ดีขึ้น ใช้ BestOfNNode ซึ่งเป็นส่วนประกอบขั้นสูงที่รัน LLM prompt N ครั้ง ใช้การเรียก LLM อีกครั้งเพื่อให้คะแนนเอาต์พุต N รายการ และเลือกเอาต์พุตที่ดีที่สุด

มาเจาะลึกโหนดหลักตามที่กำหนดไว้ใน joke_generator.json:

input_node (InputNode): โหนดนี้กำหนดจุดเริ่มต้นของเวิร์กโฟลว์

• config.output_schema: { "topic": "string", "audience": "string" }
• สิ่งนี้ระบุว่าเวิร์กโฟลว์ต้องการอินพุตสตริงสองรายการ: topic ของเรื่องตลก และ audience ที่ตั้งใจไว้

JokeDrafter (BestOfNNode): นี่คือขั้นตอนแรกของการสร้างเรื่องตลก

• config.system_message: "You are a stand-up comedian who uses dark humor like Ricky Gervais or Jimmy Carr..."
• config.user_message: "Your audience is: {{input_node.audience}}\nThe topic should be about {{input_node.topic}}"
• เทมเพลต Jinja2 นี้จะแทรกข้อมูลจาก input_node แบบไดนามิก
• config.samples: 10
• สิ่งนี้สั่งให้โหนดสร้างร่างเรื่องตลก 10 ร่าง
• config.rating_prompt: "Rate the following joke on a scale from 0 to 10..."
• หลังจากสร้างเรื่องตลก 10 เรื่องแล้ว พรอมต์นี้จะถูกใช้เพื่อให้ LLM ให้คะแนนแต่ละเรื่อง จากนั้นโหนดจะเลือกเรื่องตลกที่ได้คะแนนสูงสุด
• config.output_schema: { "initial_joke": "string" }
• เอาต์พุตของโหนดคือสตริงเดียว: เรื่องตลกที่ดีที่สุดจาก 10 ตัวอย่าง

JokeRefiner (BestOfNNode): โหนดนี้จะนำเรื่องตลกที่ร่างไว้มาปรับปรุงให้ดีขึ้น

• config.system_message: "Your goal is to refine a joke to make it more vulgar and concise..."
• config.user_message: {{JokeDrafter.initial_joke}}
• ที่สำคัญคือ อินพุตของโหนดนี้คือ เอาต์พุต ของโหนด JokeDrafter
• config.samples: 3
• มันสร้างเรื่องตลกที่ปรับปรุงแล้ว 3 เวอร์ชันจากเรื่องตลกเริ่มต้น
• นอกจากนี้ยังใช้ rating prompt เพื่อเลือกสิ่งที่ดีที่สุดจากการปรับปรุงทั้งสาม
• config.output_schema: { "final_joke": "string" }

ลิงก์: อาร์เรย์ links ใน JSON กำหนดการไหลของข้อมูล:
input_node -> JokeDrafter -> JokeRefiner

การดำเนินการและการตรวจสอบ:

• ในแผงทดสอบทางด้านขวา ใช้ข้อมูลทดสอบที่เตรียมไว้ล่วงหน้า หรือสร้างข้อมูลของคุณเอง (เช่น หัวข้อ: "ผู้ช่วย AI", กลุ่มเป้าหมาย: "นักพัฒนา")
• คลิก Run
• ขณะที่เวิร์กโฟลว์กำลังทำงาน ให้คลิกที่โหนด JokeDrafter คุณสามารถตรวจสอบ run_data ซึ่งจะแสดงตัวอย่างที่สร้างขึ้นทั้งหมด 10 ตัวอย่างและการให้คะแนนที่เกี่ยวข้อง ทำให้คุณเห็นกระบวนการ "คิด" ของ agent ได้อย่างโปร่งใส
• เรื่องตลกที่ปรับปรุงแล้วขั้นสุดท้ายจะอยู่ในเอาต์พุตของโหนด JokeRefiner

3. การนำไปใช้ RAG Pipeline

แม้ว่าจะไม่ได้เป็นส่วนหนึ่งของตัวสร้างเรื่องตลก แต่ Retrieval-Augmented Generation (RAG) เป็นความสามารถที่สำคัญของ Pyspur นี่คือกระบวนการทางเทคนิคสำหรับการเพิ่มความรู้ให้กับ agent:

1. การนำเข้าเอกสาร (Collection): ไปที่ส่วน RAG เมื่อคุณสร้าง "Document Collection" และอัปโหลดไฟล์ (เช่น PDF) Pyspur จะเริ่มกระบวนการแบ็กเอนด์ที่แยกวิเคราะห์เอกสารเป็นข้อความ แบ่งส่วนเป็นส่วนย่อยที่กำหนดค่าได้ตามความยาวของโทเค็น และจัดเก็บส่วนย่อยเหล่านี้พร้อมข้อมูลเมตาของแหล่งที่มาในฐานข้อมูล
2. การสร้างเวกเตอร์ (Index): การสร้าง "Vector Index" จากคอลเลกชันจะเริ่มกระบวนการอื่น Pyspur จะวนซ้ำผ่านแต่ละส่วนของข้อความ ทำการเรียก API ไปยังโมเดลการฝังที่ระบุ (เช่น text-embedding-ada-002 ของ OpenAI) เพื่อรับการแสดงเวกเตอร์ และ upsert เวกเตอร์เหล่านี้ลงในฐานข้อมูลเวกเตอร์ที่กำหนดค่าไว้ (เช่น ChromaDB, PGVector)
3. การเรียกคืน (Workflow Node): ในเวิร์กโฟลว์ Retriever Node จะถูกกำหนดค่าให้ชี้ไปยัง Vector Index ที่เฉพาะเจาะจง ในขณะรันไทม์ อินพุต query จะถูกฝังโดยใช้โมเดลเดียวกัน และจะมีการค้นหาเชิงความหมาย (approximate nearest neighbor) กับฐานข้อมูลเวกเตอร์เพื่อดึงส่วนของข้อความที่เกี่ยวข้องมากที่สุด ส่วนย่อยเหล่านี้จะถูกส่งต่อไปยัง LLM ปลายทางเป็นบริบท

4. การปรับใช้เป็น Production API

เมื่อเวิร์กโฟลว์ของคุณเสร็จสมบูรณ์ คุณสามารถเปิดเผยเป็นปลายทาง HTTP ที่ปลอดภัยได้

เริ่มการปรับใช้: คลิกปุ่ม "Deploy" ที่แถบนำทางด้านบน

เลือกประเภทการเรียก API:

• Blocking (Synchronous): สำหรับเวิร์กโฟลว์ที่ทำงานเร็ว ไคลเอนต์จะได้รับผลลัพธ์ในการตอบสนอง HTTP เดียวกัน
• ปลายทาง: POST /api/wf/{workflow_id}/run/?run_type=blocking
• Non-Blocking (Asynchronous): สำหรับเวิร์กโฟลว์ที่ใช้เวลานาน (เช่น ตัวสร้างเรื่องตลกของเราที่มีการเรียก LLM มากกว่า 10 ครั้ง) ไคลเอนต์จะได้รับ run_id ทันทีและต้องทำการ poll ปลายทางแยกต่างหากเพื่อรับผลลัพธ์ ซึ่งช่วยป้องกันการหมดเวลาของไคลเอนต์
• ปลายทางเริ่มต้น: POST /api/wf/{workflow_id}/start_run/?run_type=non_blocking
• ปลายทางสถานะ: GET /api/runs/{run_id}/status/

รวมเข้ากับแอปพลิเคชันของคุณ:
หน้าต่างโมดัลการปรับใช้จะสร้างโค้ดไคลเอนต์ เนื้อหาของการร้องขอ POST ต้องเป็นอ็อบเจกต์ JSON โดยที่คีย์ initial_inputs มีอ็อบเจกต์ซึ่งคีย์ตรงกับ title ของโหนดอินพุตของคุณ

ตัวอย่างไคลเอนต์ Python (สำหรับตัวสร้างเรื่องตลก):

import requests
import json
import time

PYSUR_HOST = "http://localhost:6080"
WORKFLOW_ID = "your_workflow_id_here" # Get this from the deploy modal

# The keys in this dict must match the 'output_schema' of the input_node
payload = {
    "initial_inputs": {
        "input_node": {
            "topic": "Python decorators",
            "audience": "Senior Software Engineers"
        }
    }
}

# 1. Start the non-blocking run
start_url = f"{PYSUR_HOST}/api/wf/{WORKFLOW_ID}/start_run/?run_type=non_blocking"
start_resp = requests.post(start_url, json=payload)
run_id = start_resp.json()['id']
print(f"Workflow started with run_id: {run_id}")

# 2. Poll for the result
status_url = f"{PYSUR_HOST}/api/runs/{run_id}/status/"
while True:
    status_resp = requests.get(status_url)
    data = status_resp.json()
    status = data.get("status")
    print(f"Current status: {status}")
    if status in ["COMPLETED", "FAILED"]:
        print("Final Output:")
        print(json.dumps(data.get("outputs"), indent=2))
        break
    time.sleep(2)

สรุป

Pyspur มอบสภาพแวดล้อมที่แข็งแกร่ง โปร่งใส และมีพื้นฐานทางเทคนิคสำหรับการพัฒนา AI agent ด้วยการแยกการทำงานแต่ละรายการออกเป็นโหนดแบบโมดูลาร์และจัดเตรียมพื้นที่ทำงานแบบภาพ ช่วยให้นักพัฒนาสามารถมุ่งเน้นไปที่ตรรกะระดับสูงของ agent ได้ พลังที่แท้จริงของแพลตฟอร์มนี้อยู่ที่ความสามารถในการตรวจสอบอย่างละเอียด ซึ่งช่วยให้คุณสามารถติดตามการไหลของข้อมูลและดีบัก I/O ของแต่ละส่วนประกอบ และการเปลี่ยนผ่านจากต้นแบบแบบภาพไปสู่ ​​API ที่พร้อมสำหรับการใช้งานจริงได้อย่างราบรื่น การเปลี่ยนจาก agent แบบเรียกครั้งเดียวที่เรียบง่ายไปสู่เวิร์กโฟลว์ที่ซับซ้อนและหลายขั้นตอนโดยใช้รูปแบบเช่น Best-of-N หรือ RAG ไม่เพียงแต่เป็นไปได้ แต่ยังเป็นไปตามสัญชาตญาณ ด้วยการสร้างด้วย Pyspur คุณไม่ได้เพียงแค่เชื่อมต่อกล่องต่างๆ เท่านั้น แต่คุณกำลังสร้างระบบ AI ที่น่าเชื่อถือ ดีบักได้ และปรับขนาดได้