เอเจนต์บน API LLM แบบไม่เซ็นเซอร์: การเรียกใช้ฟังก์ชัน เฟรมเวิร์ก และเกราะป้องกัน

โครงสร้างของลูปการเรียกใช้ฟังก์ชัน

ตัดเฟรมเวิร์กออกและเอเจนต์ทุกตัวจะทำสิ่งเดียวกันสี่อย่าง คุณส่งการสนทนาพร้อมรายการ tools โมเดลตอบกลับเป็นข้อความ ซึ่งหมายถึงเสร็จสิ้น หรือเป็น tool_calls ซึ่งหมายถึงต้องการให้ดำเนินการบางอย่าง คุณเรียกใช้แต่ละการเรียกและเพิ่มข้อความ tool ต่อการเรียก โดยอ้างอิง id ของการเรียกนั้น จากนั้นคุณถามโมเดลอีกครั้งด้วยการสนทนาที่ยาวขึ้น

สัญญาณของสาขาคือ finish_reason เมื่อเท่ากับ tool_calls ข้อความผู้ช่วยจะมี content เป็น null และอาร์เรย์ tool_calls ซึ่งแต่ละรายการมี id, ชื่อฟังก์ชัน และ สตริงอาร์กิวเมนต์ ที่เป็น JSON หน้า อ้างอิง แสดงรูปแบบที่แน่นอน รายละเอียดสองจุดที่ทำให้คนสับสน: อาร์กิวเมนต์เป็นสตริงที่ต้องแยกวิเคราะห์ และข้อความผู้ช่วยที่มีการเรียกต้องอยู่ในประวัติก่อนผลลัพธ์เครื่องมือของคุณตามลำดับ

ลูปสมบูรณ์ตั้งแต่เริ่มต้น

สคริปต์ด้านล่างกำหนดฟังก์ชันหนึ่งฟังก์ชัน อธิบายใน JSON schema และทำลูปจนกว่าโมเดลจะผลิตข้อความธรรมดา ใช้ Python SDK ทางการที่ชี้ไปที่ API ทุกรอบจะเพิ่มข้อความผู้ช่วย จากนั้นข้อความเครื่องมือต่อการเรียก และตัวนับขั้นตอนจะหยุดพฤติกรรมที่หลุดควบคุม

Copyimport json, os
from openai import OpenAI

client = OpenAI(base_url="https://api.uncensoredapis.com/v1", api_key=os.environ["API_KEY"])

TICKETS = {"T-101": "open", "T-102": "closed"}

def ticket_status(ticket_id: str) -> str:
    return TICKETS.get(ticket_id, "unknown ticket")

TOOLS = [{
    "type": "function",
    "function": {
        "name": "ticket_status",
        "description": "Look up the status of a support ticket by id.",
        "parameters": {
            "type": "object",
            "properties": {"ticket_id": {"type": "string"}},
            "required": ["ticket_id"],
        },
    },
}]
REGISTRY = {"ticket_status": ticket_status}

messages = [{"role": "user", "content": "Are tickets T-101 and T-102 still open?"}]

for step in range(6):                      # hard cap on loop turns
    r = client.chat.completions.create(
        model="uncensored", messages=messages, tools=TOOLS, max_tokens=400)
    msg = r.choices[0].message
    messages.append(msg.model_dump(exclude_none=True))
    if r.choices[0].finish_reason != "tool_calls":
        print(msg.content)
        break
    for call in msg.tool_calls:
        fn = REGISTRY.get(call.function.name)
        try:
            args = json.loads(call.function.arguments)
            result = fn(**args) if fn else f"error: unknown tool {call.function.name}"
        except Exception as exc:
            result = f"error: {exc}"
        messages.append({"role": "tool", "tool_call_id": call.id, "content": str(result)})
else:
    print("stopped: step limit reached")

สังเกตสิ่งที่ลูปทำเมื่อล้มเหลว: ข้อผิดพลาดกลายเป็นผลลัพธ์เครื่องมือเป็นสตริง สิ่งนี้ทำให้โมเดลอ่านปัญหาและลองอีกครั้ง เช่น แก้ไข ticket id ที่ผิด แทนที่โปรแกรมของคุณจะล้มเหลว เงื่อนไข for ... else จะทำงานเฉพาะเมื่อถึงขีดจำกัดโดยไม่มีการตอบสุดท้าย ซึ่งเป็นวิธีที่คุณตรวจจับโมเดลที่ติดอยู่ในวงกลม

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

การป้องกันการทำงานของการเรียกใช้ฟังก์ชัน

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

• อนุญาตโดยชื่อ ดูฟังก์ชันในรีจิสทรีที่คุณสร้าง อย่าใช้สตริงเพื่อ import หรือประเมินโค้ด
• แยกการอ่านจากการเขียน เครื่องมือแบบอ่านอย่างเดียวทำงานได้อย่างอิสระ ทุกสิ่งที่ส่งอีเมล ใช้เงิน ลบข้อมูล หรือแตะไฟล์ต้องการการอนุมัติจากมนุษย์อย่างชัดเจน
• ตรวจสอบอาร์กิวเมนต์ กับสคีมาที่คุณประกาศ: คีย์ที่จำเป็น ไม่มีคีย์เพิ่มเติม ประเภทและขีดจำกัดความยาว
• จำกัดเวลาและผลลัพธ์ ใช้ time-outs และตัดผลลัพธ์ก่อนป้อนกลับเพื่อให้ผลลัพธ์ฟังก์ชันขนาดใหญ่ไม่กินหน้าต่างบริบท 100,000 โทเคนของคุณ
• จำกัดจำนวนครั้งและต้นทุน จำกัดขั้นตอนต่องานและโทเคนต่อเซสชัน โดยใช้บล็อก usage
• บันทึกทุกการเรียก พร้อมชื่อ อาร์กิวเมนต์ และ outcome เพื่อให้คุณตรวจสอบพฤติกรรมได้

Copyimport json
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutTimeout

ALLOWED = {"ticket_status", "lookup_invoice"}      # read-only tools only
NEEDS_APPROVAL = {"refund_order"}                  # side effects wait for a human
POOL = ThreadPoolExecutor(max_workers=4)

def validate(args: dict, schema: dict) -> None:
    props = schema["properties"]
    for key in schema.get("required", []):
        if key not in args:
            raise ValueError(f"missing argument: {key}")
    for key, val in args.items():
        if key not in props:
            raise ValueError(f"unexpected argument: {key}")
        if props[key]["type"] == "string" and not isinstance(val, str):
            raise ValueError(f"{key} must be a string")
        if isinstance(val, str) and len(val) > 200:
            raise ValueError(f"{key} too long")

def run_tool(name, raw_args, registry, schemas, approve=lambda n, a: False):
    if name in NEEDS_APPROVAL:
        args = json.loads(raw_args)
        if not approve(name, args):
            return "denied: this action needs human approval"
    elif name not in ALLOWED:
        return f"error: tool {name} is not permitted"
    args = json.loads(raw_args)
    validate(args, schemas[name]["parameters"])
    future = POOL.submit(registry[name], **args)
    try:
        return str(future.result(timeout=5))[:2000]    # time and size limits
    except FutTimeout:
        return "error: tool timed out"

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

การเขียนสคีมาฟังก์ชันที่โมเดลใช้งานได้

สคีมาคือเอกสารเพียงอย่างเดียวที่โมเดลเห็น จึงสมควรได้รับการดูแลเหมือน API สาธารณะ ตั้งชื่อฟังก์ชันด้วยคำกริยาและคำนามที่ตรงกับงาน เช่น ticket_status หรือ lookup_invoice แทนป้ายกำกับทั่วไปเช่น run หรือ helper เขียน description เป็น instruction: ฟังก์ชันทำอะไร ส่งอะไรกลับ และเมื่อไหร่ไม่ควรใช้

• เก็บพารามิเตอร์ให้น้อยและแบน มีฟิลด์เรียบง่ายสองหรือสามฟิลด์ที่เรียกได้ถูกต้องบ่อยกว่าวัตถุแบบซ้อนกันที่มีสิบฟิลด์
• ใช้ enums สำหรับชุดปิด เช่น "enum": ["open", "closed"] เพื่อให้โมเดลไม่สามารถสร้างค่าขึ้นมาเองได้
• อธิบายแต่ละคุณสมบัติ ด้วยหน่วยหรือตัวอย่าง: "Ticket id เช่น T-101" ชนะสตริงเปล่า
• ส่งผลลัพธ์แบบกระชับ สตริงหรือวัตถุ JSON ขนาดเล็กทำงานได้ดีที่สุด; การส่งข้อมูลขนาดใหญ่เผาผลาญหน้าต่างบริบทเพื่อประโยชน์เล็กน้อย
• ชอบเครื่องมือแคบหลายตัว แทนเครื่องมือ omnibus เดียวที่มีโหมดแฟลก

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

การตั้งค่า LangChain

คลาส ChatOpenAI ของ LangChain รับค่า base_url จึงไม่ต้องใช้แพ็กเกจผู้กำหนดค่าแบบกำหนดเอง ตั้งค่า model เป็น uncensored ส่งคีย์จากสภาพแวดล้อมของคุณ และผูกฟังก์ชันด้วย bind_tools ติดตั้ง langchain-openai และ langchain-core

Copyimport os
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

@tool
def ticket_status(ticket_id: str) -> str:
    """Return the status of a support ticket."""
    return {"T-101": "open", "T-102": "closed"}.get(ticket_id, "unknown ticket")

llm = ChatOpenAI(
    model="uncensored",
    base_url="https://api.uncensoredapis.com/v1",
    api_key=os.environ["API_KEY"],
    temperature=0.3,
    max_tokens=400,
)
llm_with_tools = llm.bind_tools([ticket_status])

ai = llm_with_tools.invoke("What is the status of ticket T-101?")
for call in ai.tool_calls:
    print(call["name"], call["args"], "->", ticket_status.invoke(call["args"]))

ผลลัพธ์ของ invoke คือ AIMessage ซึ่งรายการ tool_calls มีชื่อและอาร์กิวเมนต์ที่แยกวิเคราะห์แล้ว ดังนั้นคุณจึงข้ามขั้นตอน JSON แบบแมนนวลได้ เพื่อให้ได้ลูปอัตโนมัติเต็มรูปแบบ ให้ส่งวัตถุ llm เดียวกันไปยังผู้สร้างเอเจนต์ของ LangChain; เอนด์พอยต์ต้องเคารพการเรียกใช้เครื่องมือเท่านั้น ซึ่งมันทำ หากเอเจนต์ดูเหมือนเพิกเฉยต่อเครื่องมือ ให้ตรวจสอบว่ามี docstring และ type hints เนื่องจากสิ่งเหล่านี้กลายเป็นสคีมาที่โมเดลเห็น

การตั้งค่า LlamaIndex

LlamaIndex มี OpenAILike สำหรับเซิร์ฟเวอร์ที่ปฏิบัติตามรูปแบบการเสร็จสิ้นการสนทนา การตั้งค่าสามอย่างสำคัญ api_base ชี้ไปที่ API is_chat_model=True เลือกเส้นทางแชท และ is_function_calling_model=True บอกเฟรมเวิร์กว่าสามารถส่งเครื่องมือได้ ตั้ง context_window เป็น 100000 เพื่อให้ตัวแยกข้อความและบัฟเฟอร์หน่วยความจำปรับขนาดตัวเองได้อย่างถูกต้อง ติดตั้ง llama-index-llms-openai-like

Copyimport os
from llama_index.llms.openai_like import OpenAILike
from llama_index.core.tools import FunctionTool

def ticket_status(ticket_id: str) -> str:
    """Return the status of a support ticket."""
    return {"T-101": "open", "T-102": "closed"}.get(ticket_id, "unknown ticket")

llm = OpenAILike(
    model="uncensored",
    api_base="https://api.uncensoredapis.com/v1",
    api_key=os.environ["API_KEY"],
    is_chat_model=True,
    is_function_calling_model=True,
    context_window=100000,
    max_tokens=400,
)

tool = FunctionTool.from_defaults(fn=ticket_status)
response = llm.predict_and_call([tool], "Check ticket T-102 for me.")
print(response)

จากจุดนี้คุณสามารถส่ง llm เดียวกันไปยัง worker เอเจนต์หรือเครื่องค้นหาข้อมูล โปรดจำไว้ว่าเส้นทางการดึงข้อมูลสามารถบรรจุชิ้นส่วนจำนวนมากลงในพรอมต์เดียวได้ ด้วยหน้าต่าง 100,000 โทเคนที่แชร์ระหว่างพรอมต์และการเสร็จสิ้น ให้จำกัดจำนวนชิ้นส่วนที่ดึงและเก็บ max_tokens ให้ต่ำ

การจัดการเอเจนต์ภายใต้งบประมาณ

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

บริบทเติบโตในทุกรอบเนื่องจากผลลัพธ์เครื่องมือแต่ละรายการถูกเพิ่ม สรุปหรือลบเอาต์พุตเครื่องมือเก่าเมื่อใช้แล้ว และจับตาดู usage.prompt_tokens; เมื่อเข้าใกล้ 50,000 ให้บีบอัดประวัติ สำหรับต้นทุน ให้สมมติว่างานตัวอย่างมีการเรียก 6 ครั้งโดยเฉลี่ย 3,000 โทเคนพรอมต์และ 200 โทเคนเสร็จสิ้น: นั่นคือ 18,000 โทเคนอินพุต (ประมาณ $0.0045 ที่ $0.25 ต่อล้าน) และ 1,200 โทเคนเอาต์พุต (ประมาณ $0.0012 ที่ $1.00 ต่อล้าน) จำนวนโทเคนเหล่านี้เป็นสมมติฐานสำหรับผลรวม ไม่ใช่การวัด หากคุณใหม่ต่อเอนด์พอยต์ ให้เริ่มที่ เอกสาร และเครดิตทดลองใช้ฟรี

การทดสอบและแก้ไขข้อบกพร่องของเอเจนต์

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

1. จำลองโมเดล Stub ไคลเอนต์เพื่อให้ส่งข้อความ tool_calls ที่บันทึกไว้ สิ่งนี้ทดสอบทะเบียน การตรวจสอบ และความสิ้นสุดของลูปของคุณโดยไม่ต้องใช้เครือข่ายหรือใช้จ่าย
2. ตรึงการสุ่มตัวอย่าง ใช้ temperature ต่ำ เช่น 0.2 สำหรับงานที่ขับเคลื่อนโดยเครื่องมือ ความสุ่มที่ต่ำกว่าให้ JSON อาร์กิวเมนต์ที่ทำซ้ำได้มากขึ้น และคุณสามารถเพิ่มค่านี้ในภายหลังสำหรับขั้นตอนที่เน้นการเขียน
3. ตรวจสอบเส้นทาง สำหรับพรอมต์คงที่จำนวนหนึ่ง ให้ตรวจสอบว่าเครื่องมือใดถูกเรียกและตามลำดับใด ไม่ใช่เพียงข้อความสุดท้าย

จับตาดู finish_reason ด้วย ค่า length ระหว่างการเรียกใช้เครื่องมือหมายความว่าอาร์กิวเมนต์ถูกตัด; เพิ่ม max_tokens (สูงสุด 32,000) และ JSON จะสมบูรณ์ สตรีมมิงทำงานกับเครื่องมือได้เช่นกัน แต่สะสมเศษอาร์กิวเมนต์ตามดัชนีการเรียกและแยกวิเคราะห์เฉพาะหลังจากสตรีมปิด สำหรับการทบทวนเกี่ยวกับเศษเหล่านั้น ดู ส่วนสตรีมมิงของเอกสารอ้างอิง