Python: Combining Output of previous to previous step in current step

[monuminu]

Hi Team

Current i have a workflow where i have document analysis step which will extract the document and summarizes it. Output of this will be passed to Policy verifier step . After that the next step is Fraud Detection , in this step i need output of both Policy verification step and document analysis step. how can i achive this.

    WorkflowBuilder()
    .set_start_executor(document_analyzer)
    .add_edge(document_analyzer, policy_verifier)
    .add_edge(policy_verifier, fraud_detector)
    .add_edge(fraud_detector, decision_maker)
    .build()

@handler
    async def detect_fraud(
        self, 
        policy_verification: PolicyVerification, 
        ctx: WorkflowContext[FraudAssessment]
    ) -> None:
        """
        Assess fraud risk for the claim.
        
        Args:
            policy_verification: Policy verification from previous stage
            ctx: Workflow context for sending output
        """

[ekzhu]

You can use add_fan_in_edges from document_analyzer and policy_verifier to fraud_detector.

.add_fan_in_edges([document_analyzer, policy_verifier], fraud_detector)

This requires the output types of document_analyzer and policy_verifier to be the same, and the handler in fraud_detector in this case should handle a list of that output type.

Alternatively, you can add edges directly to fraud_detector:

.add_edge(document_analyzer, fraud_detector)
`add_edge(policy_verifier, fraud_detector)

But in fraud_detector, you need to keep track of the previous inputs from either one of the source executors, and only start the real work when both inputs are available.