DEVOCEAN

// Request 요청 구조체
type Request struct {
    Message string `json:"message"`
}

// ResponseChunk 에이전트 응답 스트리밍 단위
type ResponseChunk struct {
    Name    string `json:"name"`
    Content string `json:"content"`
}

// Agent 에이전트 인터페이스
type Agent interface {
    Run(input Request) <-chan ResponseChunk
}

import (
    "math/rand"
    "time"
)

type mockAgent struct {
    name string
}

func NewMockAgent() Agent {
    return &mockAgent{
        name: "Echo",
    }
}

func (m *mockAgent) Run(input Request) <-chan ResponseChunk {
    // 출력 채널 생성
    out := make(chan ResponseChunk)

    // 고루틴을 이용한 비동기 작업 수행
    go func() {
        defer close(out)

        // 입력 문자열의 각 문자마다 랜덤 지연 후 출력 채널로 전송 (긴 지연시간, 중간 스트리밍 시뮬레이션)
        for _, v := range input.Message {
            time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
            out <- ResponseChunk{Name: m.name, Content: string(v)}
        }
    }()

    return out
}

package main

import (
    "github.com/gin-gonic/gin"
    "io"
    "net/http"
)

func main() {
    r := gin.Default()

    // POST /echo 엔드포인트
    r.POST("/echo", func(c *gin.Context) {
        var req Request
        if err := c.ShouldBind(&req); err != nil {
            c.AbortWithStatusJSON(http.StatusBadRequest, gin.H{"error": err.Error()})
            return
        }

        agent := NewMockAgent()
        out := agent.Run(req)	// 에이전트 실행하고 즉시 응답 채널 반환
        sse(c, out)				// 채널을 SSE 응답으로 스트리밍
        return
    })

    r.Run()
}

func sse(c *gin.Context, ch <-chan ResponseChunk) bool {
    c.Writer.Header().Set("Content-Type", "text/event-stream")
    c.Writer.Header().Set("Cache-Control", "no-cache")
    c.Writer.Header().Set("Connection", "keep-alive")

    if f, ok := c.Writer.(http.Flusher); ok {
        f.Flush()
    }

    return c.Stream(func(w io.Writer) bool {
        if chunk, ok := <-ch; ok {
            // 채널에서 읽은 데이터를 SSE 이벤트로 전송
            c.SSEvent("", chunk)
            return true
        } else {
            // 채널이 닫히면 스트리밍 종료
            return false
        }
    })
}

➜  curl -N -H 'Content-Type: application/json' \
   -d '{"message": "안녕하세요"}' \
   'localhost:8080/echo'
data:{"name":"Echo","content":"안"}

data:{"name":"Echo","content":"녕"}

data:{"name":"Echo","content":"하"}

data:{"name":"Echo","content":"세"}

data:{"name":"Echo","content":"요"}

➜  

type chatgptAgent struct {
    name   string
    client openai.Client
}

func NewChatGptAgent() Agent {
    apiKey := "..."

    return &chatgptAgent{
        name:   "ChatGPT",
        client: openai.NewClient(option.WithAPIKey(apiKey)),
    }
}

func (c *chatgptAgent) Run(input Request) <-chan ResponseChunk {
    out := make(chan ResponseChunk)

    go func() {
        defer close(out)

        stream := c.client.Chat.Completions.NewStreaming(context.TODO(), openai.ChatCompletionNewParams{
            Messages: []openai.ChatCompletionMessageParamUnion{openai.UserMessage(input.Message)},
            Model:    openai.ChatModelGPT4_1Mini,
        })

        for stream.Next() {
            chunk := stream.Current()

            if len(chunk.Choices) == 0 {
                continue
            }

            contentChunk := chunk.Choices[0].Delta.Content
            if contentChunk == "" {
                continue
            }

            out <- ResponseChunk{
                Name:    c.name,
                Content: contentChunk,
            }
        }
    }()

    return out
}

type claudeAgent struct {
    name   string
    client anthropic.Client
}

func NewClaudeAgent() Agent {
    apiKey := "..."

    return &claudeAgent{
        name:   "Claude",
        client: anthropic.NewClient(option.WithAPIKey(apiKey)),
    }
}

func (c *claudeAgent) Run(input Request) <-chan ResponseChunk {
    out := make(chan ResponseChunk)

    go func() {
        defer close(out)

        stream := c.client.Messages.NewStreaming(context.TODO(), anthropic.MessageNewParams{
            MaxTokens: 1024,
            Model:     anthropic.ModelClaudeSonnet4_0,
            Messages:  []anthropic.MessageParam{anthropic.NewUserMessage(anthropic.NewTextBlock(input.Message))},
        })

        for stream.Next() {
            event := stream.Current()

            switch eventVariant := event.AsAny().(type) {
            case anthropic.ContentBlockDeltaEvent:
                switch deltaVariant := eventVariant.Delta.AsAny().(type) {
                case anthropic.TextDelta:
                    if deltaVariant.Text == "" {
                        continue
                    }

                    out <- ResponseChunk{
                        Name:    c.name,
                        Content: deltaVariant.Text,
                    }
                }

            }
        }
    }()

    return out
}

type geminiAgent struct {
    name   string
    client *genai.Client
}

func NewGeminiAgent() Agent {
    apiKey := "..."
    client, _ := genai.NewClient(context.TODO(), &genai.ClientConfig{
        APIKey:  apiKey,
        Backend: genai.BackendGeminiAPI,
    })

    return &geminiAgent{
        name:   "Gemini",
        client: client,
    }
}

func (g *geminiAgent) Run(input Request) <-chan ResponseChunk {
    out := make(chan ResponseChunk)

    go func() {
        defer close(out)

        iter := g.client.Models.GenerateContentStream(context.TODO(), "gemini-2.5-flash-lite", genai.Text(input.Message), nil)

        for chunk := range iter {
            if len(chunk.Candidates) == 0 || len(chunk.Candidates[0].Content.Parts) == 0 {
                continue
            }

            contentChunk := chunk.Candidates[0].Content.Parts[0]
            if contentChunk.Text == "" {
                continue
            }

            out <- ResponseChunk{
                Name:    g.name,
                Content: contentChunk.Text,
            }
        }
    }()

    return out
}

func spawnAgents(input Request, agents []Agent) []<-chan ResponseChunk {
    var chs []<-chan ResponseChunk

    for _, a := range agents {
        ch := a.Run(input)	// 각 에이전트를 비동기로 시작
        chs = append(chs, ch)
    }

    return chs
}

// FanInChunk 병합 채널용 스트리밍 구조체
type FanInChunk struct {
    Order int           // 에이전트 순서
    Chunk ResponseChunk // 실제 데이터
    Done  bool          // 완료 신호
}

func fanIn(chs []<-chan ResponseChunk) <-chan FanInChunk {
    var wg sync.WaitGroup
    wg.Add(len(chs))

    out := make(chan FanInChunk)
    for i, v := range chs {
        // 각 채널을 동시에 읽기
        go func(order int, ch <-chan ResponseChunk) {
            for chunk := range ch {
                // 읽은 데이터를 채널 인덱스와 함께 출력 채널로 전달
                out <- FanInChunk{Order: order, Chunk: chunk, Done: false}
            }

            // 채널이 닫히면 완료 신호 전달
            out <- FanInChunk{Order: order, Chunk: ResponseChunk{}, Done: true}
            wg.Done()
        }(i, v)
    }

    go func() {
        wg.Wait()
        close(out)
    }()

    return out
}

func orchestrate(in <-chan FanInChunk) <-chan ResponseChunk {
	out := make(chan ResponseChunk)

	go func() {
		defer close(out)

		buf := make([][]ResponseChunk, 0) // 각 에이전트의 청크를 저장할 버퍼
		cur := 0                          // 현재 출력 대상 에이전트 인덱스

		for chunk := range in {
			// 0. 버퍼 확장
			for len(buf) <= chunk.Order {
				buf = append(buf, []ResponseChunk{})
			}

			// 1. 현재 순서가 아니면: 버퍼에 저장
			if chunk.Order != cur {
				if !chunk.Done {
					buf[chunk.Order] = append(buf[chunk.Order], chunk.Chunk)
				}

				continue
			}

			// 2. 현재 순서: 버퍼 flush 후 처리
			for _, buffered := range buf[chunk.Order] {
				out <- buffered
			}
			buf[chunk.Order] = nil

			if chunk.Done {
				cur++
			} else {
				out <- chunk.Chunk
			}
		}

		// 3. 남은 버퍼 flush
		for cur < len(buf) {
			for _, buffered := range buf[cur] {
				out <- buffered
			}
			cur++
		}
	}()

	return out
}

func main() {
    r := gin.Default()

	// 출력 순서는 아래 정의된 대로
    agents := []Agent{
        NewChatGptAgent(), 
        NewClaudeAgent(),
        NewGeminiAgent(),
    }

    r.POST("/orchestrate", func(c *gin.Context) {
        var req Request
        if err := c.ShouldBind(&req); err != nil {
            c.AbortWithStatusJSON(http.StatusBadRequest, gin.H{"error": err.Error()})
            return
        }

        chs := spawnAgents(req, agents)	// 병렬 실행
        merged := fanIn(chs)				// 채널 병합
        out := orchestrate(merged)		// 순서 조율

        sse(c, out)
        return
    })

    r.Run()
}

[Agent1] ───────┐
[Agent2] ───────┼──────→ [fanIn] ───────→ [orchestrate] ───────→ [SSE Stream]
[Agent3] ───────┘

            (병렬 실행)             (무작위)                (순서 보장)

➜  curl -N -H 'Content-Type: application/json' \
   -d '{"message": "What is the meaning of life?"}' \
   'localhost:8080/orchestrate'
data:{"name":"ChatGPT","content":"The"}			# ChatGPT 에이전트 응답

data:{"name":"ChatGPT","content":" meaning"}

data:{"name":"ChatGPT","content":" of"}

data:{"name":"ChatGPT","content":" life"}

data:{"name":"ChatGPT","content":" is"}

data:{"name":"ChatGPT","content":" a"}

data:{"name":"ChatGPT","content":" deeply"}

data:{"name":"ChatGPT","content":" personal"}

data:{"name":"ChatGPT","content":" and"}

data:{"name":"ChatGPT","content":" philosophical"}

data:{"name":"ChatGPT","content":" question"}

...

data:{"name":"Claude","content":"This"}			# Claude 에이전트 응답

data:{"name":"Claude","content":" is one of humanity"}

data:{"name":"Claude","content":"'s oldest and most profound questions, an"}

data:{"name":"Claude","content":"d there's no single answer"}

...

data:{"name":"Gemini","content":"Ah"}			# Gemini 에이전트 응답

data:{"name":"Gemini","content":", the age-old question! \"What is the meaning of life?\" It"}

data:{"name":"Gemini","content":"'s a question that has been pondered by philosophers, theologians, scientists, artists, and everyday people for millennia. And the truth is, there's no single, universally agreed-upon answer.\n\nHere's a breakdown of why it's"}

...

➜   

2025/10/13 14:54:44 orchestrator started
2025/10/13 14:54:44 ChatGPT agent started
2025/10/13 14:54:44 Claude agent started
2025/10/13 14:54:44 Gemini agent started
2025/10/13 14:54:46 ChatGPT agent ended (elapsed: 2.481247291s)
2025/10/13 14:54:47 Gemini agent ended (elapsed: 3.520516333s)
2025/10/13 14:54:50 Claude agent ended (elapsed: 6.467166166s)
2025/10/13 14:54:50 orchestrator ended (elapsed: 6.469606583s)	    # 가장 느린 에이전트 시간과 동일
[GIN] 2025/10/13 - 14:54:50 | 200 |  6.471028125s |             ::1 | POST     "/orchestrate"