12-Faktor-Agenten 8: Übernehmen Sie die Kontrolle über Ihren Kontrollfluss.

Wenn Sie die Kontrolle über Ihren Kontrollfluss haben, können Sie viele interessante Funktionen implementieren.

Erstellen Sie benutzerdefinierte Kontrollstrukturen, die zu Ihrem speziellen Anwendungsfall passen. Bestimmte Arten von Toolaufrufen könnten ein Grund sein, aus einer Schleife herauszuspringen, auf die Antwort eines Menschen zu warten oder auf eine andere lang laufende Aufgabe zu warten (z. B. eine Trainingspipeline). Möglicherweise möchten Sie auch benutzerdefinierte Implementierungen der folgenden Funktionen integrieren:

• Zusammenfassung oder Zwischenspeicher der Ergebnisse von Toolaufrufen
• LLM-as-judge (Large Language Model as Judge) für strukturierte Ausgaben
• Komprimierung des Kontextfensters oder andere Speicherverwaltung
• Protokollierung, Verfolgung und Metriken
• Client-seitige Ratenbegrenzung
• Dauerhafter Ruhezustand / Anhalten / "Warten auf Ereignis

Das folgende Beispiel zeigt drei mögliche Kontrollflussmuster:

• request_clarification: Das Modell fordert weitere Informationen an, unterbricht die Schleife und wartet auf eine menschliche Antwort
• fetch_git_tags: das Modell fordert eine Liste von Git-Tags an, holt die Tags ab, fügt sie dem Kontextfenster hinzu und gibt sie direkt an das Modell zurück
• deploy_backend: Das Modell fordert die Bereitstellung des Backends an, was mit einem hohen Risiko verbunden ist; daher wird die Schleife unterbrochen und auf die Genehmigung des Menschen gewartet.

def handle_next_step(thread: Thread):
while True:
next_step = await determine_next_step(thread_to_prompt(thread))
# 为清晰起见，此处为内联代码 - 实际上你可以把它
# 放在一个方法里，使用异常来控制流程，或任何你喜欢的方式
if next_step.intent == 'request_clarification':
thread.events.append({
type: 'request_clarification',
data: nextStep,
})
await send_message_to_human(next_step)
await db.save_thread(thread)
# 异步步骤 - 中断循环，我们稍后会收到一个 webhook
break
elif next_step.intent == 'fetch_open_issues':
thread.events.append({
type: 'fetch_open_issues',
data: next_step,
})
issues = await linear_client.issues()
thread.events.append({
type: 'fetch_open_issues_result',
data: issues,
})
# 同步步骤 - 将新的上下文传递给 LLM 以确定下一步的行动
continue
elif next_step.intent == 'create_issue':
thread.events.append({
type: 'create_issue',
data: next_step,
})
await request_human_approval(next_step)
await db.save_thread(thread)
# 异步步骤 - 中断循环，我们稍后会收到一个 webhook
break

Dieser Modus ermöglicht es Ihnen, den Fluss der Intelligenzen nach Bedarf zu unterbrechen und wieder aufzunehmen, um einen natürlicheren Dialog und Arbeitsablauf zu schaffen.

typisches Beispiel - Für jedes KI-Framework auf dem Markt würde ich mir am meisten die Möglichkeit wünschen, eine funktionierende Intelligenz zu unterbrechen und später wieder aufzunehmen, insbesondere bei Tools Option und Werkzeuge Aufforderungen Zwischen diesen beiden Momenten.

Ohne diesen Grad an Wiederherstellbarkeit/Granularität gibt es keine Möglichkeit, einen Werkzeugaufruf zu überprüfen/genehmigen, bevor er ausgeführt wird, was bedeutet, dass Sie eine von mehreren Optionen wählen müssen:

1. Anhalten von Aufgaben im Speicher während des Wartens auf die Beendigung lang laufender Aufgaben (wie die while...sleep), und wenn der Prozess unterbrochen wird, muss er von Grund auf neu gestartet werden
2. Beschränkung der Berechtigungen von Intelligenz auf Anrufe mit geringem Risiko und geringer Bedeutung, wie z. B. Recherchen und Zusammenfassungen
3. Geben Sie der Intelligenz die Befugnis, wichtigere und nützlichere Aufgaben auszuführen, und verlassen Sie sich darauf, dass sie sich nicht durch reines Glück irrt (yolo).

Sie werden feststellen, dass dies nicht dasselbe ist wie Element 5 - Harmonisierung des Ausführungsstatus mit dem Geschäftsstatus im Gesang antworten Element 6 - Starten/Aussetzen/Fortsetzen mit einer einfachen API Sie sind eng miteinander verbunden, können aber unabhängig voneinander realisiert werden.