สร้างเอเจนต์ LangGraph ที่จำกัดขอบเขตสำหรับวินิจฉัยเหตุขัดข้องของ API

แนวทางตัดสินใจ: วินิจฉัยโดยอัตโนมัติ แก้ไขอย่างรอบคอบ

เอเจนต์สำหรับรับมือเหตุขัดข้องของ API ช่วยประหยัดเวลาได้ด้วยการรวบรวมหลักฐานจากล็อก การตรวจสอบสถานะ และข้อมูลเมตาของการ deploy ล่าสุด แต่ไม่ควรรีสตาร์ตบริการ ย้อนกลับรุ่นที่เผยแพร่ หรือเปลี่ยนการกำหนดค่าโดยไม่แจ้งให้ทราบ การออกแบบที่ปลอดภัยกว่าคือแยกการวินิจฉัยแบบอ่านอย่างเดียวออกจากการแก้ไขที่มีผลเปลี่ยนแปลงระบบ จำกัดขอบเขตงานของเอเจนต์ และแสดงหลักฐานที่ล้มเหลวหรือขาดหายให้ผู้ปฏิบัติงานเห็น

นี่คือแนวทางการออกแบบเชิงปฏิบัติ ไม่ใช่การผสานระบบที่ผ่านการทดสอบหรือบริการที่พร้อมใช้งานใน production เนื้อหาภายในเครื่องมือด้านล่างเป็นเพียงตัวอย่าง ให้แทนที่ด้วยไคลเอ็นต์ observability ของคุณเองที่ผ่านการยืนยันตัวตนและจำกัดสิทธิ์แล้ว

ข้อกำหนดเบื้องต้นและเวิร์กโฟลว์

คุณต้องมี Python โมเดลแชตที่ใช้ร่วมกับ LangGraph ได้และกำหนดค่าไว้ในแอปพลิเคชันของคุณ รวมถึงฟังก์ชันไคลเอ็นต์สำหรับระบบวินิจฉัยที่คุณใช้อยู่ ตัวอย่างนี้สมมติว่าฟังก์ชันไคลเอ็นต์เหล่านั้นส่งคืนข้อมูลหรือโยน exceptions โดยไม่ได้กำหนดผู้ให้บริการโมเดล endpoint หรือการผสานเครื่องมือแบบใดแบบหนึ่งไว้โดยเฉพาะ

การทำงานเพื่อรับมือเหตุขัดข้องที่มีประโยชน์ควรเป็นวงจรที่มีข้อจำกัดชัดเจน:

1. รวบรวมสัญญาณที่กำหนดไว้ล่วงหน้าจำนวนเล็กน้อยสำหรับบริการและช่วงเวลาที่ได้รับผลกระทบ
2. ระบุผลลัพธ์แต่ละรายการว่าใช้ได้ ล้มเหลว หรือไม่พร้อมใช้งาน
3. ขอให้โมเดลสรุปหลักฐานและระบุสิ่งที่ยังไม่ทราบ
4. หยุดเมื่อถึงจำนวนขั้นตอนกราฟที่กำหนด และขอให้มนุษย์ตรวจสอบก่อนดำเนินการใด ๆ ที่เปลี่ยนสถานะระบบ

ฟิลด์สถานะมีความสำคัญต่อการปฏิบัติงาน เพราะการค้นหาล็อกที่ล้มเหลวต้องไม่ดูเหมือนการค้นหาที่ทำงานสำเร็จแต่ไม่พบเหตุการณ์ที่ตรงกัน

กำหนดเครื่องมือวินิจฉัยพร้อมสัญญาณความล้มเหลวที่ชัดเจน

กำหนดสัญญาการใช้งานเครื่องมือให้มีขอบเขตแคบ ระบุให้ต้องมีชื่อบริการและช่วงเวลาที่จำกัด หลีกเลี่ยงการให้โมเดลเข้าถึงการค้นหาได้อย่างอิสระ หากใช้การค้นหาที่กำหนดไว้ตายตัวได้ ให้ส่งคืนผลลัพธ์ที่มีโครงสร้าง เพื่อให้โหนดถัดไปและรายงานเหตุขัดข้องแยกหลักฐานที่ว่างเปล่าออกจากความล้มเหลวของเครื่องมือได้

from typing import Literal, TypedDict
from langchain.tools import tool

class Signal(TypedDict):
    status: Literal["ok", "error", "unavailable"]
    source: str
    data: str
    error: str | None


def query_logs(service: str, minutes: int) -> Signal:
    if not 1 <= minutes <= 60:
        return {
            "status": "error", "source": "logs", "data": "",
            "error": "minutes must be between 1 and 60",
        }
    try:
        # Replace with a read-only, scoped observability client call.
        rows = logs_client.search(service=service, minutes=minutes)
        return {
            "status": "ok", "source": "logs",
            "data": str(rows), "error": None,
        }
    except TimeoutError as exc:
        return {
            "status": "unavailable", "source": "logs", "data": "",
            "error": f"timeout: {exc}",
        }

@tool
def inspect_api(service: str, minutes: int = 15) -> dict:
    """Collect scoped, read-only API incident signals."""
    return query_logs(service, minutes)

เพิ่มเครื่องมือแยกสำหรับตรวจสอบสถานะหรือข้อมูลเมตาของการ deploy โดยใช้รูปแบบผลลัพธ์เดียวกัน อย่าดักจับ exceptions ทุกชนิดแล้วส่งคืนสตริงว่าง ควรเก็บบริบทของข้อผิดพลาดไว้ให้เพียงพอสำหรับวินิจฉัยการผสานระบบที่ขัดข้อง โดยไม่ใส่ความลับหรือเพย์โหลดที่ละเอียดอ่อนลงในล็อกหรือบริบทของโมเดล

จำกัดการทำงานและแสดงผลเมื่อการทำงานไม่สมบูรณ์

สำหรับเอเจนต์ที่สร้างด้วย create_agent ของ LangGraph ให้กำหนดขีดจำกัด recursion ของกราฟขณะเรียกใช้งาน ขีดจำกัดนี้จำกัดจำนวนขั้นตอนของกราฟ แต่ไม่ได้รับประกันว่าจะได้คำตอบที่มีประโยชน์ และใช้แทน timeouts ระดับเครื่องมือไม่ได้ จัดการกรณีที่ถึงขีดจำกัดในแอปพลิเคชันที่ครอบการทำงาน และส่งคืนสถานะ “ไม่สมบูรณ์” ที่ชัดเจน แทนการมองว่าการทำงานนั้นเป็นการวินิจฉัยที่เสร็จสมบูรณ์แล้ว

from langchain.agents import create_agent

agent = create_agent(
    model=your_configured_chat_model,
    tools=[inspect_api],
    system_prompt=(
        "Diagnose API incidents using available evidence. "
        "Report each source as ok, error, or unavailable. "
        "Do not claim a cause without supporting evidence. "
        "Do not perform remediation."
    ),
)

try:
    result = agent.invoke(
        {"messages": [{"role": "user", "content":
            "Investigate elevated errors for payments-api over the last 15 minutes."}]},
        config={"recursion_limit": 12},
    )
except Exception as exc:
    # Map known graph-limit and provider/tool failures to your app's
    # incident-run status; do not expose credentials or raw secrets.
    result = {"status": "incomplete", "reason": type(exc).__name__}

ค่าขีดจำกัดนี้เป็นเพียงตัวอย่าง ไม่ใช่คำแนะนำที่ใช้ได้กับทุกกรณี ให้เลือกค่าตามจำนวนเครื่องมือและรอบการทำงานของโมเดลที่อนุญาต พร้อมเพิ่ม timeouts และขีดจำกัดขนาดคำขอในระดับเครื่องมือและแอปพลิเคชัน บันทึก run ID บริการและช่วงเวลาที่เลือก เครื่องมือที่ลองใช้ สถานะของเครื่องมือ และสาเหตุที่ยุติการทำงาน สัญญาณเหล่านี้ช่วยให้ผู้ปฏิบัติงานแยกได้ว่าเป็นกรณี “ไม่พบข้อผิดพลาด” หรือ “ระบบแบ็กเอนด์ของล็อกหมดเวลา”

เพิ่มการตรวจสอบโดยมนุษย์ก่อนดำเนินการแก้ไข

หากภายหลังคุณเพิ่มเครื่องมือที่เปลี่ยนสถานะของระบบ production ให้ควบคุมการใช้เครื่องมือเหล่านั้นด้วย human-in-the-loop middleware ของ LangChain และ checkpointer มิดเดิลแวร์ที่มีเอกสารรองรับสามารถหยุดการเรียกใช้เครื่องมือที่ตรงตามเงื่อนไข เพื่อให้อนุมัติ แก้ไข ปฏิเสธ หรือส่งคำตอบโดยตรงได้ ระบบจะบันทึกสถานะไว้เพื่อให้การทำงานกลับมาดำเนินต่อหลังมีการตัดสินใจ การใช้งานใน production ต้องใช้ checkpointer แบบถาวร ไม่ใช่ตัวเลือกแบบ in-memory ที่แสดงไว้สำหรับการทำต้นแบบ ดู เอกสาร human-in-the-loop ของ LangChain

แนวทางที่ใช้ได้จริงคืออนุญาตให้การวินิจฉัยแบบอ่านอย่างเดียวทำงานต่อได้โดยไม่หยุด และกำหนดให้ขออนุมัติสำหรับเครื่องมือรีสตาร์ต ย้อนกลับรุ่น หรือเปลี่ยนการกำหนดค่า หากปฏิเสธข้อเสนอให้ดำเนินการ ให้ปฏิเสธการดำเนินการนั้นโดยตรง เอกสารเตือนว่า “respond” ใช้สำหรับดำเนินการในฐานะเครื่องมือ ไม่ใช่สำหรับปฏิเสธการเรียกใช้ที่มีผลเปลี่ยนแปลงระบบ ตรวจสอบอาร์กิวเมนต์ที่แก้ไขแล้วใน implementation ของเครื่องมือคุณเอง เพราะมิดเดิลแวร์ตามเอกสารไม่ได้ตรวจสอบการแก้ไขกับ schema ที่แสดงไว้

กรณีความล้มเหลวที่ควรออกแบบรองรับ

• ผลลัพธ์ว่างเปล่า: รายงานว่า “ไม่มีเหตุการณ์ที่ตรงกันส่งกลับมา” ไม่ใช่ “บริการทำงานปกติ”
• เครื่องมือหมดเวลาหรือไม่ได้รับอนุญาต: เก็บแหล่งข้อมูลที่ล้มเหลวและหมวดหมู่ข้อผิดพลาดไว้ อย่าลองใหม่ซ้ำไปเรื่อย ๆ โดยไม่แจ้งให้ทราบ
• เรียกใช้เครื่องมือซ้ำ: จำกัดการทำงานของกราฟและควบคุมอินพุตของเครื่องมือ หากเพิ่มการลองใหม่ ให้กำหนดนโยบายการลองใหม่ที่ชัดเจนและมีจำนวนครั้งน้อย พร้อมแสดงจำนวนครั้งที่ลอง
• กลับมาทำต่อจาก interrupt ไม่ได้: ตรวจสอบว่ามีการกำหนดค่า checkpointer และใช้ตัวระบุ thread เดิมเมื่อต้องการทำงานต่อ เอกสารต้นทางกำหนดให้ใช้ checkpointing สำหรับการตรวจสอบโดยมนุษย์
• สรุปสุดท้ายชวนให้เข้าใจผิด: กำหนดให้รายงานอ้างอิงสัญญาณที่รวบรวมได้ พร้อมระบุแหล่งข้อมูลที่ไม่พร้อมใช้งานและคำถามที่ยังไม่ได้คำตอบ

รายการตรวจสอบก่อนเริ่มใช้งานเพื่อรับมือเหตุขัดข้อง

ก่อน deploy ให้ตรวจสอบว่าเครื่องมือวินิจฉัยอ่านข้อมูลได้อย่างเดียวและมีขอบเขตชัดเจน จำกัดช่วงเวลาและปริมาณการค้นหา มองเห็นข้อผิดพลาดของเครื่องมือได้ กำหนดขีดจำกัดการทำงานของกราฟ และหยุดรอการตัดสินใจจากมนุษย์ก่อนทุกการดำเนินการที่เปลี่ยนสถานะระบบ ทดสอบเส้นทางเมื่อหมดเวลา เมื่อได้ผลลัพธ์ว่าง เมื่อปฏิเสธการดำเนินการ และเมื่อกลับมาทำต่อหลังถูกหยุดชั่วคราว เอเจนต์ที่ดีไม่ได้แค่เสนอหาสาเหตุรากที่ดูเป็นไปได้ แต่ต้องแสดงหลักฐานที่รวบรวมได้จริงและระบุจุดที่การวินิจฉัยสิ้นสุดลง

แหล่งข้อมูล

• LangChain: Human-in-the-loop middleware — การตัดสินใจจาก interrupt การตรวจสอบเครื่องมือตามเงื่อนไข การทำ checkpoint และการกลับมาทำงานต่อหลังหยุดชั่วคราว