Upgradehandbuch: Workflow-APIs und Request-Response System

Dieses Handbuch hilft Ihnen, Ihre Python-Workflows auf die neuesten API-Änderungen zu aktualisieren, die in Version 1.0.0b251104 eingeführt wurden.

Übersicht über Änderungen

Diese Version enthält zwei wichtige Verbesserungen am Workflowsystem:

1. Konsolidierte Workflowausführungs-APIs

Die Workflowausführungsmethoden wurden aus Gründen der Einfachheit vereinheitlicht:

  • Einheitliche run(..., stream=True) und run() Methoden: Ersetzen separater prüfpunktspezifischer Methoden (run_stream_from_checkpoint(), run_from_checkpoint())
  • Einzelne Schnittstelle: Verwenden des checkpoint_id Parameters zum Fortsetzen von Prüfpunkten anstelle separater Methoden
  • Flexible Prüfpunkterstellung: Konfigurieren des Prüfpunktspeichers zur Buildzeit oder Außerkraftsetzung zur Laufzeit
  • Klarere Semantik: Sich gegenseitig ausschließende message Parameter (neue Ausführung) und checkpoint_id (Fortsetzung)

2. Vereinfachtes Request-Response System

Das Anforderungsantwortsystem wurde optimiert:

  • Nicht mehr RequestInfoExecutor: Executors können jetzt Direktanfragen senden
  • Neuer @response_handler Dekorateur: Ersetze RequestResponse Nachrichten-Handler
  • Vereinfachte Anforderungstypen: Keine Vererbung von RequestInfoMessage erforderlich
  • Integrierte Funktionen: Alle Executoren unterstützen automatisch die Anforderungsantwortfunktion.
  • Übersichtlichere Workflowdiagramme: Entfernen von RequestInfoExecutor Knoten aus Ihren Workflows

Teil 1: Einheitliche Workflowausführungs-APIs

Es wird empfohlen, zuerst zu den konsolidierten Workflow-APIs zu migrieren, da dies die Grundlage für alle Workflowausführungsmuster bildet.

Fortsetzen ab Prüfpunkten

Vor (alte API):

# OLD: Separate method for checkpoint resume
async for event in workflow.run_stream_from_checkpoint(
    checkpoint_id="checkpoint-id",
    checkpoint_storage=checkpoint_storage
):
    print(f"Event: {event}")

After (Neue API):

# NEW: Unified method with checkpoint_id parameter
async for event in workflow.run(
    checkpoint_id="checkpoint-id",
    checkpoint_storage=checkpoint_storage,  # Optional if configured at build time
    stream=True,
):
    print(f"Event: {event}")

Wichtige Unterschiede:

  • Verwenden des checkpoint_id Parameters anstelle einer separaten Methode
  • Es können weder message noch checkpoint_id bereitgestellt werden, da sie sich gegenseitig ausschließen.
  • Muss entweder message (neue Ausführung) oder checkpoint_id (Fortsetzen) angeben.
  • checkpoint_storage ist optional, wenn Prüfpunktausführung zur Build-Zeit konfiguriert wurde

API ohne Streaming

Die Nicht-Streaming-Methode run() folgt demselben Muster:

Alt:

result = await workflow.run_from_checkpoint(
    checkpoint_id="checkpoint-id",
    checkpoint_storage=checkpoint_storage
)

Neu:

result = await workflow.run(
    checkpoint_id="checkpoint-id",
    checkpoint_storage=checkpoint_storage  # Optional if configured at build time
)

Fortsetzen des Prüfpunkts mit ausstehenden Anforderungen

Wenn die Ausführung ab einem Prüfpunkt mit ausstehenden Anforderungs-Info-Ereignissen fortgesetzt wird, gibt die API diese Ereignisse automatisch erneut aus. Sie können sie erfassen und darauf reagieren oder im selben Aufruf responses mit checkpoint_id bereitstellen.

Vor (altes Verhalten):

# OLD: Could provide responses directly during resume
responses = {
    "request-id-1": "user response data",
    "request-id-2": "another response"
}

async for event in workflow.run_stream_from_checkpoint(
    checkpoint_id="checkpoint-id",
    checkpoint_storage=checkpoint_storage,
    responses=responses  # No longer supported
):
    print(f"Event: {event}")

Nach (Neues Verhalten):

# Capture re-emitted pending requests
requests: dict[str, Any] = {}

async for event in workflow.run(checkpoint_id="checkpoint-id", stream=True):
    if event.type == "request_info":
        # Pending requests are automatically re-emitted
        print(f"Pending request re-emitted: {event.request_id}")
        requests[event.request_id] = event.data

# Collect user responses
responses: dict[str, Any] = {}
for request_id, request_data in requests.items():
    response = handle_request(request_data)  # Your logic here
    responses[request_id] = response

# Send responses back to workflow
async for event in workflow.run(responses=responses, stream=True):
    if event.type == "output":
        print(f"Workflow output: {event.data}")

Vollständiges Beispiel für "Human-in-the-Loop"

Nachfolgend sehen Sie ein vollständiges Beispiel für die Wiederaufnahme eines Prüfpunkts mit ausstehender menschlicher Genehmigung.

from agent_framework import (
    Executor,
    FileCheckpointStorage,
    WorkflowBuilder,
    handler,
    response_handler,
)

# ... (Executor definitions omitted for brevity)

async def run_interactive_session(
    workflow: Workflow,
    initial_message: str | None = None,
    checkpoint_id: str | None = None,
) -> str:
    """Run workflow until completion, handling human input interactively."""

    requests: dict[str, HumanApprovalRequest] = {}
    responses: dict[str, str] | None = None
    completed_output: str | None = None

    while True:
        # Determine which API to call
        if responses:
            # Send responses from previous iteration
            event_stream = workflow.run(responses=responses, stream=True)
            requests.clear()
            responses = None
        else:
            # Start new run or resume from checkpoint
            if initial_message:
                event_stream = workflow.run(initial_message, stream=True)
            elif checkpoint_id:
                event_stream = workflow.run(checkpoint_id=checkpoint_id, stream=True)
            else:
                raise ValueError("Either initial_message or checkpoint_id required")

        # Process events
        async for event in event_stream:
            if event.type == "status":
                print(event)
            if event.type == "output":
                completed_output = event.data
            if event.type == "request_info":
                if isinstance(event.data, HumanApprovalRequest):
                    requests[event.request_id] = event.data

        # Check completion
        if completed_output:
            break

        # Prompt for user input if we have pending requests
        if requests:
            responses = prompt_for_responses(requests)
            continue

        raise RuntimeError("Workflow stopped without completing or requesting input")

    return completed_output

Teil 2: Vereinfachtes Request-Response System

Aktualisieren Sie nach der Migration zu den einheitlichen Workflow-APIs Ihre Anforderungsantwortmuster, um das neue integrierte System zu verwenden.

1. Importe aktualisieren

Before:

from agent_framework import (
    RequestInfoExecutor,
    RequestInfoMessage,
    RequestResponse,
    # ... other imports
)

After:

from agent_framework import (
    response_handler,
    # ... other imports
    # Remove: RequestInfoExecutor, RequestInfoMessage, RequestResponse
)

2. Aktualisierungsanforderungstypen

Before:

from dataclasses import dataclass
from agent_framework import RequestInfoMessage

@dataclass
class UserApprovalRequest(RequestInfoMessage):
    """Request for user approval."""
    prompt: str = ""
    context: str = ""

After:

from dataclasses import dataclass

@dataclass
class UserApprovalRequest:
    """Request for user approval."""
    prompt: str = ""
    context: str = ""

3. Workflowdiagramm aktualisieren

Before:

# Old pattern: Required RequestInfoExecutor in workflow
approval_executor = ApprovalRequiredExecutor(id="approval")
request_info_executor = RequestInfoExecutor(id="request_info")

workflow = (
    WorkflowBuilder(start_executor=approval_executor)
    .add_edge(approval_executor, request_info_executor)
    .add_edge(request_info_executor, approval_executor)
    .build()
)

After:

# New pattern: Direct request-response capabilities
approval_executor = ApprovalRequiredExecutor(id="approval")

workflow = (
    WorkflowBuilder(start_executor=approval_executor)
    .build()
)

4. Senden von Aktualisierungsanforderungen

Before:

class ApprovalRequiredExecutor(Executor):
    @handler
    async def process(self, message: str, ctx: WorkflowContext[UserApprovalRequest]) -> None:
        request = UserApprovalRequest(
            prompt=f"Please approve: {message}",
            context="Important operation"
        )
        await ctx.send_message(request)

After:

class ApprovalRequiredExecutor(Executor):
    @handler
    async def process(self, message: str, ctx: WorkflowContext) -> None:
        request = UserApprovalRequest(
            prompt=f"Please approve: {message}",
            context="Important operation"
        )
        await ctx.request_info(request_data=request, response_type=bool)

5. Aktualisieren der Antwortbehandlung

Before:

class ApprovalRequiredExecutor(Executor):
    @handler
    async def handle_approval(
        self,
        response: RequestResponse[UserApprovalRequest, bool],
        ctx: WorkflowContext[Never, str]
    ) -> None:
        if response.data:
            await ctx.yield_output("Approved!")
        else:
            await ctx.yield_output("Rejected!")

After:

class ApprovalRequiredExecutor(Executor):
    @response_handler
    async def handle_approval(
        self,
        original_request: UserApprovalRequest,
        approved: bool,
        ctx: WorkflowContext
    ) -> None:
        if approved:
            await ctx.yield_output("Approved!")
        else:
            await ctx.yield_output("Rejected!")

Zusammenfassung der Vorteile

Einheitliche Workflow-APIs

  1. Vereinfachte Schnittstelle: Einzelne Methode für erste Ausführungen sowie Fortsetzen von Prüfpunkten
  2. Klarere Semantik: Sich gegenseitig ausschließende Parameter machen die Absicht explizit
  3. Flexible Prüfpunktausführung: Konfigurieren zur Build-Zeit oder Außerkraftsetzung zur Laufzeit
  4. Reduzierte kognitive Belastung: Weniger Methoden zum Merken und Pflegen

Anforderungs-Antwort-System

  1. Vereinfachte Architektur: Keine Notwendigkeit für separate RequestInfoExecutor Komponenten
  2. Typsicherheit: Direkte Typspezifizierung in request_info() Aufrufen
  3. Übersichtlicherer Code: Weniger Importe und einfachere Workflowdiagramme
  4. Bessere Leistung: Reduzierter Nachrichtenweiterleitungsaufwand
  5. Erweitertes Debuggen: Klarerer Ausführungsfluss und Fehlerbehandlung

Testen Ihrer Migration

Teil 1 Prüfliste: Workflow-APIs

  1. Api-Aufrufe aktualisieren: Ersetzen durch run_stream_from_checkpoint()run(checkpoint_id=..., stream=True)
  2. Api-Aufrufe aktualisieren: Ersetzen durch run_from_checkpoint()run(checkpoint_id=...)
  3. Aktuellen Fortsetzungs-Shape verwenden: Antworten beim Fortsetzen und Beantworten eines Aufrufs mit workflow.run(responses=..., stream=True) oder zusammen mit checkpoint_id dieser übergeben
  4. Hinzufügen der Ereigniserfassung: Implementieren von Logik zum Erfassen erneut ausgegebener request_info Ereignisse (event.type == "request_info")
  5. Testprüfpunkt-Fortsetzung: Überprüfen, ob ausstehende Anforderungen erneut ausgegeben und ordnungsgemäß verarbeitet werden

Teil 2 Prüfliste: Request-Response System

  1. Importe überprüfen: Sicherstellen, dass keine alten Importe verbleiben (RequestInfoExecutor, RequestInfoMessage, RequestResponse)
  2. Überprüfen von Anforderungstypen: Entfernen der RequestInfoMessage Vererbung bestätigen
  3. Test-Workflow-Diagramm: Überprüfen des Entfernens von RequestInfoExecutor Knoten
  4. Handler validieren: Sicherstellen, dass @response_handler Decorators angewendet werden
  5. End-to-End testen: Ausführen vollständiger Workflowszenarien

Nächste Schritte

Nach Abschluss der Migration:

  1. Überprüfen des aktualisierten Lernprogramms zu Anforderungen und Antworten
  2. Erkunden von erweiterten Mustern im Benutzerhandbuch
  3. Sehen Sie sich aktualisierte Beispiele im Repository an

Weitere Hilfe finden Sie in der Agent Framework-Dokumentation oder wenden Sie sich an das Team und die Community.

  • Last updated on