为智能体工作流构建图路由¶

edges := workflow.Concat(
    workflow.Chain(workflow.Start, classifyNode),
    []workflow.Edge{
        {From: classifyNode, To: responseA, Route: workflow.StringRoute("output-1")},
        {From: classifyNode, To: responseB, Route: workflow.StringRoute("output-2")},
        {From: classifyNode, To: responseC, Route: workflow.StringRoute("output-3")},
    },
)
rootAgent, _ := workflowagent.New(workflowagent.Config{
    Name:  "routing_workflow",
    Edges: edges,
})

// newFunctionNodePipeline demonstrates workflow.NewFunctionNode as the primary
// v2 node type. A FunctionNode wraps a plain Go function: the function returns
// a typed value, and the framework automatically wraps it in a session.Event,
// setting event.Output. The successor node receives this value as its typed
// input parameter.
//
// This is the direct Go equivalent of the Python FunctionNode:
//
//  def my_function_node(node_input: str):
//      return Event(output=node_input.upper())
func newFunctionNodePipeline() (agent.Agent, error) {
    upperFn := func(_ agent.Context, input string) (string, error) {
        return strings.ToUpper(input), nil
    }

    suffixFn := func(_ agent.Context, input string) (string, error) {
        return input + " IS AWESOME!", nil
    }

    // workflow.NewFunctionNode wraps each function as a graph node.
    // workflow.Chain wires them in order: START → upper → suffix.
    // The output of upperFn is delivered as the typed input of suffixFn
    // via event.Output — no session state writes are needed.
    nodeA := workflow.NewFunctionNode("upper", upperFn, workflow.NodeConfig{})
    nodeB := workflow.NewFunctionNode("suffix", suffixFn, workflow.NodeConfig{})

    return workflowagent.New(workflowagent.Config{
        Name:        "function_node_pipeline",
        Description: "Demonstrates workflow.NewFunctionNode data flow via Event.Output.",
        Edges:       workflow.Chain(workflow.Start, nodeA, nodeB),
    })
}

// newSequentialNodes builds a two-step sequential workflow using the v2 graph
// engine. workflow.Chain wires the nodes in order; each node's typed return
// value is forwarded to the next node via event.Output.
//
// This is the Go equivalent of:
//
//  edges=[("START", task_A_node, task_B_node)]
func newSequentialNodes() (agent.Agent, error) {
    // task_A_node: transforms the user's input.
    taskANode := workflow.NewFunctionNode("task_A_node",
        func(_ agent.Context, input string) (string, error) {
            return "Summary: " + strings.TrimSpace(input), nil
        },
        workflow.NodeConfig{},
    )

    // task_B_node: receives task A's output as its typed input and produces
    // the final result. No session state reads needed.
    taskBNode := workflow.NewFunctionNode("task_B_node",
        func(_ agent.Context, summary string) (string, error) {
            return strings.ToUpper(summary), nil
        },
        workflow.NodeConfig{},
    )

    return workflowagent.New(workflowagent.Config{
        Name:        "sequential_workflow",
        Description: "Runs task A then task B in order via workflow.Chain.",
        Edges:       workflow.Chain(workflow.Start, taskANode, taskBNode),
    })
}

// classifyNode 根据消息发出 Routes=[]string{"BUG"}、
// ["CUSTOMER_SUPPORT"] 或 ["LOGISTICS"] 的事件。
edges := workflow.Concat(
    workflow.Chain(workflow.Start, processMessage, classifyNode),
    []workflow.Edge{
        {From: classifyNode, To: bugHandler,       Route: workflow.StringRoute("BUG")},
        {From: classifyNode, To: supportHandler,   Route: workflow.StringRoute("CUSTOMER_SUPPORT")},
        {From: classifyNode, To: logisticsHandler, Route: workflow.StringRoute("LOGISTICS")},
    },
)
rootAgent, _ := workflowagent.New(workflowagent.Config{
    Name:  "routing_workflow",
    Edges: edges,
})

eb := workflow.NewEdgeBuilder()
eb.Add(workflow.Start, processMessage)
eb.Add(processMessage, classifyNode)
eb.AddRoute(classifyNode, bugHandler,       workflow.StringRoute("BUG"))
eb.AddRoute(classifyNode, supportHandler,   workflow.StringRoute("CUSTOMER_SUPPORT"))
eb.AddRoute(classifyNode, logisticsHandler, workflow.StringRoute("LOGISTICS"))

rootAgent, _ := workflowagent.New(workflowagent.Config{
    Name:  "routing_workflow",
    Edges: eb.Build(),
})

gatherNode := workflow.NewJoinNode("gather")

eb := workflow.NewEdgeBuilder()
eb.AddFanOut(workflow.Start, researchNodeA, researchNodeB, researchNodeC)
eb.AddFanIn(gatherNode, researchNodeA, researchNodeB, researchNodeC)
eb.Add(gatherNode, formatNode)
eb.Add(formatNode, synthesisNode)

rootAgent, _ := workflowagent.New(workflowagent.Config{
    Name:  "research_pipeline",
    Edges: eb.Build(),
})

// newParallelFanOut builds a fan-out / join workflow using the v2 graph engine.
// Three research nodes run in parallel from Start; workflow.NewJoinNode waits
// for all of them to complete and emits a map[nodeName]output to the format
// node, which assembles the results for a synthesis node.
//
// Graph topology:
//
//  START ─┬─> research_A ──┐
//         ├─> research_B ──┼─> gather (JoinNode) ─> format ─> synthesis
//         └─> research_C ──┘
//
// Python equivalent:
//
//  edges=[
//      ("START", research_A, my_join_node),
//      ("START", research_B, my_join_node),
//      ("START", research_C, my_join_node),
//      (my_join_node, format_node),
//      (format_node, synthesis_node),
//  ]
func newParallelFanOut() (agent.Agent, error) {
    researchA := workflow.NewFunctionNode("research_A",
        func(_ agent.Context, _ any) (string, error) {
            return "Fact about renewable energy.", nil
        },
        workflow.NodeConfig{},
    )
    researchB := workflow.NewFunctionNode("research_B",
        func(_ agent.Context, _ any) (string, error) {
            return "Fact about electric vehicles.", nil
        },
        workflow.NodeConfig{},
    )
    researchC := workflow.NewFunctionNode("research_C",
        func(_ agent.Context, _ any) (string, error) {
            return "Fact about carbon capture.", nil
        },
        workflow.NodeConfig{},
    )

    // workflow.NewJoinNode waits for all predecessors (research_A, research_B,
    // research_C) to complete and emits a map[nodeName]output to its successor.
    gatherNode := workflow.NewJoinNode("gather")

    // formatNode receives map[string]any from gatherNode and assembles a
    // combined prompt string.
    formatNode := workflow.NewFunctionNode("format",
        func(_ agent.Context, results map[string]any) (string, error) {
            return fmt.Sprintf("A: %v\nB: %v\nC: %v",
                results["research_A"],
                results["research_B"],
                results["research_C"],
            ), nil
        },
        workflow.NodeConfig{},
    )

    synthesisNode := workflow.NewFunctionNode("synthesis",
        func(_ agent.Context, prompt string) (string, error) {
            return "Combined report: " + prompt, nil
        },
        workflow.NodeConfig{},
    )

    // EdgeBuilder.AddFanOut fans workflow.Start out to all three research nodes.
    // EdgeBuilder.AddFanIn routes all three research nodes into gatherNode.
    eb := workflow.NewEdgeBuilder()
    eb.AddFanOut(workflow.Start, researchA, researchB, researchC)
    eb.AddFanIn(gatherNode, researchA, researchB, researchC)
    eb.Add(gatherNode, formatNode)
    eb.Add(formatNode, synthesisNode)

    return workflowagent.New(workflowagent.Config{
        Name:        "fan_out_workflow",
        Description: "Parallel research fan-out with JoinNode barrier and synthesis.",
        Edges:       eb.Build(),
    })
}

innerNode, _ := workflow.NewAgentNode(innerWorkflowAgent, workflow.NodeConfig{})

outerEdges := workflow.Chain(workflow.Start, outerStepNode, innerNode, finalNode)
rootAgent, _ := workflowagent.New(workflowagent.Config{
    Name:  "parent_workflow",
    Edges: outerEdges,
})

// newNestedWorkflows shows how to nest one workflowagent inside another using
// the v2 graph engine. The inner workflowagent is wrapped with
// workflow.NewAgentNode and placed as a node in the outer graph's edge slice.
// From the outer graph's perspective the inner workflow is a single node that
// runs to completion before the edge to finalNode is followed.
//
// Python equivalent:
//
//  root_agent = Workflow(
//      name="parent_workflow",
//      edges=[("START", task_A1, workflow_B, final_node)],
//  )
func newNestedWorkflows() (agent.Agent, error) {
    // --- Inner workflow B ---
    innerStep1 := workflow.NewFunctionNode("inner_step_1",
        func(_ agent.Context, input string) (string, error) {
            return "[ES] " + input, nil // simulate translation to Spanish
        },
        workflow.NodeConfig{},
    )
    innerStep2 := workflow.NewFunctionNode("inner_step_2",
        func(_ agent.Context, spanish string) (string, error) {
            return "[EN] " + spanish, nil // simulate translation back to English
        },
        workflow.NodeConfig{},
    )

    // workflowB is a self-contained inner graph.
    workflowB, err := workflowagent.New(workflowagent.Config{
        Name:        "workflow_B",
        Description: "Translates input to Spanish then back to English.",
        Edges:       workflow.Chain(workflow.Start, innerStep1, innerStep2),
    })
    if err != nil {
        return nil, fmt.Errorf("workflowB: %w", err)
    }

    // --- Outer graph ---
    taskA1 := workflow.NewFunctionNode("task_A1",
        func(_ agent.Context, input string) (string, error) {
            return "Summary: " + strings.TrimSpace(input), nil
        },
        workflow.NodeConfig{},
    )

    finalNode := workflow.NewFunctionNode("final_node",
        func(_ agent.Context, result string) (string, error) {
            return "Final: " + result, nil
        },
        workflow.NodeConfig{},
    )

    // workflow.NewAgentNode wraps workflowB so it can be placed as a node
    // in the outer graph's edges slice.
    innerNode, err := workflow.NewAgentNode(workflowB, workflow.NodeConfig{})
    if err != nil {
        return nil, fmt.Errorf("NewAgentNode(workflowB): %w", err)
    }

    return workflowagent.New(workflowagent.Config{
        Name:        "parent_workflow",
        Description: "Runs task_A1 then the nested workflow_B then final_node.",
        Edges:       workflow.Chain(workflow.Start, taskA1, innerNode, finalNode),
        SubAgents:   []agent.Agent{workflowB},
    })
}

// draft carries the working document through the refinement loop.
type draft struct {
    Text string `json:"text"`
}

// criticResult is emitted by the critic node with the review verdict and
// optional suggestions. The router reads Verdict to set Event.Routes.
type criticResult struct {
    Verdict     string `json:"verdict"`     // "REFINE" or "DONE"
    Suggestions string `json:"suggestions"` // non-empty when Verdict == "REFINE"
}

// writeDraft is the initial writer node: produces the first draft from the
// user's topic. Its typed return value becomes the input to the critic node
// via Event.Output — no session state writes needed.
func writeDraft(_ agent.Context, topic string) (draft, error) {
    // In a real workflow this would call an LLM; here we return a stub.
    return draft{Text: "Draft about " + topic + ": placeholder content."}, nil
}

// reviewDraft is the critic node: inspects the draft and returns a verdict.
// "DONE" exits the loop; "REFINE" triggers a back-edge to the refiner.
func reviewDraft(_ agent.Context, d draft) (criticResult, error) {
    // Simulate a critic: approve once the draft contains "improved".
    if strings.Contains(d.Text, "improved") {
        return criticResult{Verdict: "DONE"}, nil
    }
    return criticResult{
        Verdict:     "REFINE",
        Suggestions: "Add more detail and mark the text as improved.",
    }, nil
}

// routeVerdict reads the critic's verdict and sets Event.Routes so the
// graph engine dispatches to either the refiner or the done node.
// Returning nil suppresses the automatic terminal event.
func routeVerdict(ctx agent.Context, r criticResult, emit func(*session.Event) error) (any, error) {
    ev := session.NewEvent(ctx, ctx.InvocationID())
    ev.Routes = []string{r.Verdict}
    ev.Output = r // forward the full result to the chosen successor
    if err := emit(ev); err != nil {
        return nil, err
    }
    return nil, nil
}

// refineDraft applies the critic's suggestions and returns the improved draft.
// Its output feeds back to the critic node via the back-edge.
func refineDraft(_ agent.Context, r criticResult) (draft, error) {
    return draft{Text: "improved draft incorporating: " + r.Suggestions}, nil
}

// reportDone is the terminal node, reached only when the critic is satisfied.
func reportDone(_ agent.Context, r criticResult) (string, error) {
    return "Refinement complete. Final verdict: " + r.Verdict, nil
}

// newLoopEscalate builds an iterative document-refinement workflow using the
// graph engine. The critic node emits a route ("REFINE" or "DONE") and the
// engine dispatches to either the refiner (which loops back to the critic via
// a back-edge) or the terminal done node.
//
// Graph topology:
//
//  START → writer → critic → router ─┬─ "REFINE" → refiner ──┐
//                                     └─ "DONE"   → done       │
//                   ▲_______________________________┘ (back-edge)
//
// Python equivalent:
//
//  edges=[
//      ("START", writer_node, critic_node, router),
//      (router, {"REFINE": refiner_node, "DONE": done_node}),
//      (refiner_node, critic_node),  # back-edge creates the loop
//  ]
func newLoopEscalate() (agent.Agent, error) {
    writerNode := workflow.NewFunctionNode("writer", writeDraft, workflow.NodeConfig{})
    criticNode := workflow.NewFunctionNode("critic", reviewDraft, workflow.NodeConfig{})
    routerNode := workflow.NewEmittingFunctionNode("router", routeVerdict, workflow.NodeConfig{})
    refinerNode := workflow.NewFunctionNode("refiner", refineDraft, workflow.NodeConfig{})
    doneNode := workflow.NewFunctionNode("done", reportDone, workflow.NodeConfig{})

    // Build the edges. The back-edge from refinerNode to criticNode creates
    // the loop; the graph engine re-activates criticNode with a fresh
    // lifecycle on each iteration.
    eb := workflow.NewEdgeBuilder()
    eb.Add(workflow.Start, writerNode)
    eb.Add(writerNode, criticNode)
    eb.Add(criticNode, routerNode)
    eb.AddRoute(routerNode, refinerNode, workflow.StringRoute("REFINE"))
    eb.AddRoute(routerNode, doneNode, workflow.StringRoute("DONE"))
    eb.AddRoute(refinerNode, criticNode, workflow.Default) // back-edge: loop back for another review

    return workflowagent.New(workflowagent.Config{
        Name:        "iterative_writer",
        Description: "Writes then iteratively refines a document using a critic/refiner loop.",
        Edges:       eb.Build(),
    })
}