Agenci na API LLM bez cenzury: wywoływanie funkcji, frameworki i zabezpieczenia

Anatomia pętli wywoływania narzędzi

Zdejmij frameworki i każdy agent wykonuje te same cztery rzeczy. Wysyłasz rozmowę wraz z listą tools. Model odpowiada albo tekstem, co oznacza zakończenie, albo tool_calls, co oznacza chęć wykonania czegoś. Uruchamiasz każde wywołanie i dołączasz jedną wiadomość tool dla każdego wywołania, cytując id wywołania. Następnie pytasz model ponownie dłuższą rozmową.

Sygnałem do gałęzi jest finish_reason. Gdy wynosi tool_calls, wiadomość asystenta ma content ustawione na null i tablicę tool_calls, gdzie każdy wpis zawiera identyfikator, nazwę funkcji i ciąg argumentów w formacie JSON. Dokładny kształt pokazuje strona referencyjna. Dwa szczegóły często mylą: argumenty to ciąg do sparsowania, a wiadomość asystenta zawierająca wywołania musi pozostać w historii przed wynikami narzędzi, w odpowiedniej kolejności.

Pełna pętla od zera

Poniższy skrypt definiuje jedną funkcję, opisuje ją w schemacie JSON i pętli się, aż model wyprodukuje zwykły tekst. Używa oficjalnego SDK Pythona wskazującego na API. Każda iteracja dołącza wiadomość asystenta, następnie wiadomość narzędzia dla każdego wywołania, a licznik kroków zatrzymuje niekontrolowane zachowanie.

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")

Zauważ, co pętla robi w przypadku błędu: błędy stają się wynikami narzędzi jako ciągi. Pozwala to modelowi przeczytać problem i spróbować ponownie, np. poprawić nieprawidłowy identyfikator zgłoszenia, zamiast crashu programu. Klauzula for ... else wykonuje się tylko, jeśli limit zostanie osiągnięty bez ostatecznej odpowiedzi, co pozwala wykryć model utknęły w kółko.

Trzymaj prompty skupione. Krótki prompt systemowy określający, kiedy wywołać narzędzie, a kiedy odpowiedzieć bezpośrednio, zmniejsza zbędne wywołania, a każde pominięte wywołanie oszczędza rundę, tokeny i jedno miejsce w limicie zapytań.

Zabezpieczanie wykonywania narzędzi

Wynik modelu to niezaufane dane wejściowe. Nazwa narzędzia lub argument wybrany przez model mogą być błędne, dziwne lub wpływane przez tekst, który model przeczytał ze strony internetowej lub od użytkownika. Traktuj każde wywołanie jak zapytanie przychodzące z publicznego internetu.

• Lista dozwolonych według nazwy. Sprawdź funkcję w rejestrze, który zbudowałeś. Nigdy nie używaj ciągu do importowania lub oceny kodu.
• Oddziel odczyt od zapisu. Narzędzia tylko do odczytu działają swobodnie. Wszystko, co wysyła maile, wydaje pieniądze, usuwa dane lub dotyka plików, wymaga wyraźnej zgody człowieka.
• Waliduj argumenty względem opublikowanego schematu: wymagane klucze, brak dodatkowych kluczy, typy i limity długości.
• Ogranicz czas i wynik. Zastosuj limity czasu i obetnij wyniki przed ich ponownym podaniem, aby ogromny wynik narzędzia nie zjadł twojego okna kontekstu 100 000 tokenów.
• Ogranicz iteracje i koszt. Ogranicz kroki na zadanie i tokeny na sesję, używając bloku usage.
• Loguj każde wywołanie z nazwą, argumentami i wynikiem, abyś mógł zauditować zachowanie.

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"

Traktuj wyniki narzędzi jako niezaufane. Jeśli narzędzie zwraca tekst ze strony zawierający instrukcje, model może je wykonać. Owiń taką zawartość wyraźnymi separatorami w wiadomości narzędzia i wskaż w promptzie systemowym, że wynik narzędzia to dane, nie polecenia.

Pisanie schematów narzędzi, których może użyć model

Schemat to jedyna dokumentacja, jaką widzi model, więc zasługuje na taką samą staranność jak publiczne API. Nazwij funkcje czasownikami i rzeczownikami pasującymi do zadania, np. ticket_status lub lookup_invoice, zamiast ogólnych etykiet jak run lub helper. Napisz description jako instrukcję: co robi funkcja, co zwraca i kiedy z niej nie korzystać.

• Trzymaj mało parametrów i płasko. Dwa lub trzy proste pola są wywoływane poprawnie znacznie częściej niż zagnieżdżony obiekt z dziesięcioma.
• Używaj wyliczeń dla zamkniętych zbiorów, np. "enum": ["open", "closed"], aby model nie wymyślił wartości.
• Opisz każdą właściwość z jednostką lub przykładem: "Identyfikator zgłoszenia, np. T-101" jest lepszy niż sam ciąg.
• Zwracaj kompaktowe wyniki. Ciągi lub małe obiekty JSON działają najlepiej; długie wypisy zużywają kontekst bez wielkiej korzyści.
• Preferuj wiele wąskich narzędzi zamiast jednego uniwersalnego z flagą trybu.

Gdy wywołanie nie przejdzie walidacji, zwróć komunikat walidacji jako wynik narzędzia zamiast zgłaszania wyjątku. Model zwykle naprawia wywołanie w kolejnej turze, ponieważ może przeczytać, co było nie tak. Jeśli zawali się ten sam sposób dwukrotnie, przepisz opis, nie pętlę.

Konfiguracja LangChain

Klasa LangChain ChatOpenAI akceptuje base_url, więc nie jest potrzebny pakiet dostawcy. Ustaw model na uncensored, przekaż klucz ze środowiska i powiąż narzędzia za pomocą bind_tools. Zainstaluj langchain-openai i 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"]))

Wynikiem invoke jest AIMessage, którego lista tool_calls zawiera sparsowane nazwy i argumenty, więc pomijasz krok ręcznego parsowania JSON. Aby uzyskać pełną pętlę autonomiczną, przekaż ten sam obiekt llm do konstruktorów agentów LangChain; endpoint musi tylko wspierać wywoływanie funkcji, co robi. Jeśli agent wydaje się ignorować narzędzia, sprawdź, czy występują docstringi i typy hintów, ponieważ stają się one schematem, który widzi model.

Konfiguracja LlamaIndex

LlamaIndex dostarcza OpenAILike dla serwerów przestrzegających formatu uzupełniania czatu. Trzy ustawienia mają znaczenie. api_base wskazuje na API, is_chat_model=True wybiera trasę czatu, a is_function_calling_model=True informuje framework, że może wysyłać narzędzia. Ustaw context_window na 100000, aby podział tekstu i bufory pamięci dostosowywały się poprawnie. Zainstaluj 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)

Stąd możesz przekazać ten sam llm do pracownika agenta lub silnika zapytań. Pamiętaj, że potoki pobierania mogą wciskać wiele fragmentów w jednym promptcie; przy oknie 100 000 tokenów dzielonym między prompt a uzupełnienie, ogranicz liczbę pobranych fragmentów i trzymaj max_tokens na umiarkowanym poziomie.

Zarządzanie agentami przy ograniczonym budżecie

Agent zamienia jedną wiadomość użytkownika na kilka zapytań, więc zaplanuj pojemność. Zadanie z pięciema rundami wywoływania funkcji kosztuje sześć wywołań modelu. Dziesięć równoległych zadań generuje więc sześćdziesiąt zapytań w krótkim przedziale czasu, a limit 300 zapytań na minutę zostaje osiągnięty szybciej niż w aplikacji czatu. Umieść mechanizm throttle oparty na wiadrze żetonów z poradnikiem błędów i limitów przed swoim klientem.

Kontekszt rośnie z każdą rundą, ponieważ każdy wynik narzędzia jest dołączany. Podsumuj lub usuń stare wyniki narzędzi, gdy zostały użyte, i obserwuj usage.prompt_tokens; gdy zbliża się do 50 000, skompresuj historię. Pod względem kosztu przyjmij ilustracyjne zadanie 6 wywołań średnio 3000 tokenów promptu i 200 tokenów uzupełnienia: to 18 000 tokenów wejściowych (około 0,0045 USD przy 0,25 USD za milion) i 1200 tokenów wyjściowych (około 0,0012 USD przy 1,00 USD za milion). Te liczby tokenów to założenia do sum, nie pomiary. Jeśli jesteś nowy w endpointzie, zacznij od dokumentacji i darmowego kredytu próbnego.

Testowanie i debugowanie agenta

Agenci zawalają w sposób, jaki testy jednostkowe pojedynczej funkcji pomijają, więc testuj rozmowę. Zapisuj pełną listę wiadomości dla każdej rundy, w tym argumenty i wyniki narzędzi, i odtwarzaj błędy z dziennika. Trzy nawyki szybko się opłacą.

1. Zaśmiec model. Zastąp klienta tak, aby zwracał zaszyte komunikaty tool_calls. To pozwala przetestować rejestr, walidację i zakończenie pętli bez sieci i bez wydanych tokenów.
2. Stałe próbkowanie. Użyj niskiej temperature, np. 0.2, w zadaniach sterowanych funkcjami. Mniejsza losowość daje bardziej powtarzalne argumenty JSON, a później możesz ją zwiększyć w krokach wymagających generowania tekstu.
3. Weryfikuj trajektorie. Dla kilku stałych promptów sprawdzaj, które narzędzia zostały wywołane i w jakiej kolejności, a nie tylko końcowy tekst.

Obserwuj również finish_reason. Wartość length w trakcie wywoływania narzędzia oznacza, że argumenty zostały ucięte; zwiększ max_tokens (do 32 000), aby JSON się uzupełnił. Strumieniowanie działa również z narzędziami — gromadź fragmenty argumentów według indeksu wywołania i parsuj je dopiero po zamknięciu strumienia. Aby odświeżyć wiedzę na temat tych fragmentów, zobacz sekcję strumieniowania w dokumentacji.