Microsoft Agent Framework-arbetsflöden – tillståndsisolering

I verkliga program är det viktigt att hantera tillståndet korrekt när du hanterar flera uppgifter eller begäranden. Utan korrekt isolering kan delat tillstånd mellan olika arbetsflödeskörningar leda till oväntat beteende, skadade data och konkurrensförhållanden. Den här artikeln beskriver hur du säkerställer tillståndsisolering i Microsoft Agent Framework-arbetsflöden och ger insikter om metodtips och vanliga fallgropar.

Föränderliga arbetsflödesbyggare jämfört med oföränderliga arbetsflöden

Arbetsflöden skapas av arbetsflödesbyggare. Arbetsflödesbyggare anses vanligtvis vara föränderliga, där man kan lägga till, ändra startexekutor eller andra konfigurationer när byggaren har skapats eller till och med efter att ett arbetsflöde har skapats. Å andra sidan är arbetsflöden oföränderliga eftersom när ett arbetsflöde har skapats kan det inte ändras (inget offentligt API för att ändra ett arbetsflöde).

Den här skillnaden är viktig eftersom den påverkar hur tillståndet hanteras i olika arbetsflödeskörningar. Vi rekommenderar inte att du återanvänder en enda arbetsflödesinstans för flera uppgifter eller begäranden, eftersom detta kan leda till oavsiktlig tillståndsdelning. I stället rekommenderar vi att du skapar en ny arbetsflödesinstans från byggaren för varje uppgift eller begäran för att säkerställa korrekt tillståndsisolering och trådsäkerhet.

Säkerställa tillståndsisolering i arbetsflödesbyggare

När en exekutorinstans skickas direkt till en arbetsflödesbyggare delas den körinstansen mellan alla arbetsflödesinstanser som skapats från byggaren. Detta kan leda till problem om exekutorinstansen innehåller tillstånd som inte ska delas mellan flera arbetsflödeskörningar. För att säkerställa korrekt tillståndsisolering och trådsäkerhet rekommenderar vi att du använder fabriksfunktioner som skapar en ny körinstans för varje arbetsflödesinstans.

Kommer snart...

Exempel som inte är trådsäkert:

executor_a = CustomExecutorA()
executor_b = CustomExecutorB()

workflow_builder = WorkflowBuilder()
# executor_a and executor_b are passed directly to the workflow builder
workflow_builder.add_edge(executor_a, executor_b)
workflow_builder.set_start_executor(executor_b)

# All workflow instances created from the builder will share the same executor instances
workflow_a = workflow_builder.build()
workflow_b = workflow_builder.build()

Trådsäkert exempel:

workflow_builder = WorkflowBuilder()
# Register executor factory functions with the workflow builder
workflow_builder.register_executor(factory_func=CustomExecutorA, name="executor_a")
workflow_builder.register_executor(factory_func=CustomExecutorB, name="executor_b")
# Add edges using registered factory function names
workflow_builder.add_edge("executor_a", "executor_b")
workflow_builder.set_start_executor("executor_b")

# Each workflow instance created from the builder will have its own executor instances
workflow_a = workflow_builder.build()
workflow_b = workflow_builder.build()

Tips/Råd

För att säkerställa korrekt tillståndsisolering och trådsäkerhet kontrollerar du också att körbara instanser som skapats av fabriksfunktioner inte delar föränderligt tillstånd.

Hantering av agenttillstånd

Agentkontexten hanteras via agenttrådar. Som standard får varje agent i ett arbetsflöde sin egen tråd om inte agenten hanteras av en anpassad köre. Mer information finns i Arbeta med agenter.

Agenttrådar sparas mellan körningar av arbetsflöden. Det innebär att om en agent anropas i den första körningen av ett arbetsflöde blir innehåll som genereras av agenten tillgängligt i efterföljande körningar av samma arbetsflödesinstans. Även om detta kan vara användbart för att upprätthålla kontinuiteten i en enda uppgift, kan det också leda till oavsiktlig tillståndsdelning om samma arbetsflödesinstans återanvänds för olika uppgifter eller begäranden. För att säkerställa att varje uppgift har isolerat agenttillstånd använder du agentfabriksfunktioner i arbetsflödesverktyget för att skapa en ny arbetsflödesinstans för varje uppgift eller begäran.

Kommer snart...

Exempel som inte är trådsäkert:

writer_agent = AzureOpenAIChatClient(credential=AzureCliCredential()).create_agent(
    instructions=(
        "You are an excellent content writer. You create new content and edit contents based on the feedback."
    ),
    name="writer_agent",
)
reviewer_agent = AzureOpenAIChatClient(credential=AzureCliCredential()).create_agent(
    instructions=(
        "You are an excellent content reviewer."
        "Provide actionable feedback to the writer about the provided content."
        "Provide the feedback in the most concise manner possible."
    ),
    name="reviewer_agent",
)

builder = WorkflowBuilder()
# writer_agent and reviewer_agent are passed directly to the workflow builder
builder.add_edge(writer_agent, reviewer_agent)
builder.set_start_executor(writer_agent)

# All workflow instances created from the builder will share the same agent
# instances and agent threads
workflow = builder.build()

Trådsäkert exempel:

def create_writer_agent() -> ChatAgent:
    """Factory function to create a Writer agent."""
    return AzureOpenAIChatClient(credential=AzureCliCredential()).create_agent(
        instructions=(
            "You are an excellent content writer. You create new content and edit contents based on the feedback."
        ),
        name="writer_agent",
    )

def create_reviewer_agent() -> ChatAgent:
    """Factory function to create a Reviewer agent."""
    return AzureOpenAIChatClient(credential=AzureCliCredential()).create_agent(
        instructions=(
            "You are an excellent content reviewer."
            "Provide actionable feedback to the writer about the provided content."
            "Provide the feedback in the most concise manner possible."
        ),
        name="reviewer_agent",
    )

builder = WorkflowBuilder()
# Register agent factory functions with the workflow builder
builder.register_agent(factory_func=create_writer_agent, name="writer_agent")
builder.register_agent(factory_func=create_reviewer_agent, name="reviewer_agent")
# Add edges using registered factory function names
builder.add_edge("writer_agent", "reviewer_agent")
builder.set_start_executor("writer_agent")

# Each workflow instance created from the builder will have its own agent
# instances and agent threads
workflow = builder.build()

Conclusion

Tillståndsisolering i Microsoft Agent Framework-arbetsflöden kan hanteras effektivt med hjälp av fabriksfunktioner med arbetsflödesbyggare för att skapa nya kör- och agentinstanser. Genom att skapa nya arbetsflödesinstanser för varje uppgift eller begäran kan du upprätthålla korrekt tillståndsisolering och undvika oavsiktlig tillståndsdelning mellan olika arbetsflödeskörningar.

Nästa steg

• Lär dig hur du visualiserar arbetsflöden.