feat(service): 实现阶段 4 业务逻辑 — Agent、Runner、Assembler、ChatService

2026-06-10 15:13:21 +08:00
parent 658866a661
commit f532b21a7a
4 changed files with 706 additions and 0 deletions
--- a/internal/service/agent.go
+++ b/internal/service/agent.go
@@ -1 +1,422 @@
 package service
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"sync/atomic"
+
+	"ai-agent-scaffold-go/internal/model"
+)
+
+const maxToolCallIterations = 4
+
+// ChatModelWithTools 扩展接口，同时具备 ChatModel 和工具调用能力
+type ChatModelWithTools interface {
+	model.ChatModel
+	CallTool(ctx context.Context, name, arguments string) (string, error)
+}
+
+// ============================================================
+// LLMAgent — 基础 LLM Agent
+// ============================================================
+
+// LLMAgent 基于 LLM 的智能体，支持多轮工具调用
+type LLMAgent struct {
+	name        string
+	description string
+	instruction string
+	outputKey   string
+	chatModel   ChatModelWithTools
+}
+
+// NewLLMAgent 创建 LLM Agent
+func NewLLMAgent(name, instruction, description, outputKey string, chatModel ChatModelWithTools) *LLMAgent {
+	return &LLMAgent{
+		name:        name,
+		instruction: instruction,
+		description: description,
+		outputKey:   outputKey,
+		chatModel:   chatModel,
+	}
+}
+
+func (a *LLMAgent) Name() string      { return a.name }
+func (a *LLMAgent) OutputKey() string { return a.outputKey }
+
+// Run 同步执行
+func (a *LLMAgent) Run(ctx context.Context, content model.ChatContent) (string, error) {
+	return a.runWithVars(ctx, content, map[string]string{})
+}
+
+// Stream 流式执行
+func (a *LLMAgent) Stream(ctx context.Context, content model.ChatContent, out chan<- string) error {
+	return a.streamWithVars(ctx, content, out, map[string]string{})
+}
+
+// runWithVars 同步执行，支持变量替换
+func (a *LLMAgent) runWithVars(ctx context.Context, content model.ChatContent, vars map[string]string) (string, error) {
+	messages := initialMessages(applyVars(a.instruction, vars), firstText(content))
+
+	for iter := 0; iter < maxToolCallIterations; iter++ {
+		reply, err := a.chatModel.Generate(ctx, messages)
+		if err != nil {
+			return "", err
+		}
+		if len(reply.ToolCalls) == 0 {
+			return reply.Content, nil
+		}
+		// 将 assistant 回复（含工具调用）加入消息历史
+		messages = append(messages, model.ChatMessage{
+			Role:      model.ChatRoleAssistant,
+			Content:   reply.Content,
+			ToolCalls: reply.ToolCalls,
+		})
+		// 执行工具调用，将结果加入消息历史
+		toolMessages, err := a.executeToolCalls(ctx, reply.ToolCalls)
+		if err != nil {
+			return "", err
+		}
+		messages = append(messages, toolMessages...)
+	}
+	return "", fmt.Errorf("agent %q exceeded tool-call iteration limit %d", a.name, maxToolCallIterations)
+}
+
+// streamWithVars 流式执行，支持变量替换
+func (a *LLMAgent) streamWithVars(ctx context.Context, content model.ChatContent, out chan<- string, vars map[string]string) error {
+	messages := initialMessages(applyVars(a.instruction, vars), firstText(content))
+
+	for iter := 0; iter < maxToolCallIterations; iter++ {
+		events, errs := a.chatModel.Stream(ctx, messages)
+
+		var (
+			finalText strings.Builder
+			toolCalls []model.ChatToolCall
+			done      bool
+		)
+		streamErr := error(nil)
+
+	streamLoop:
+		for {
+			select {
+			case <-ctx.Done():
+				streamErr = ctx.Err()
+				break streamLoop
+			case ev, ok := <-events:
+				if !ok {
+					break streamLoop
+				}
+				if ev.Done {
+					done = true
+				}
+				if ev.Delta != "" {
+					finalText.WriteString(ev.Delta)
+					select {
+					case out <- ev.Delta:
+					case <-ctx.Done():
+						streamErr = ctx.Err()
+						break streamLoop
+					}
+				}
+				if len(ev.ToolCalls) > 0 {
+					toolCalls = append(toolCalls, ev.ToolCalls...)
+				}
+			case err, ok := <-errs:
+				if ok && err != nil {
+					streamErr = err
+				}
+				break streamLoop
+			}
+		}
+
+		if streamErr != nil {
+			return streamErr
+		}
+
+		// 没有工具调用，说明模型已完成回复
+		if len(toolCalls) == 0 {
+			if !done {
+				return fmt.Errorf("agent %q stream closed without completion", a.name)
+			}
+			return nil
+		}
+
+		// 处理工具调用
+		messages = append(messages, model.ChatMessage{
+			Role:      model.ChatRoleAssistant,
+			Content:   finalText.String(),
+			ToolCalls: toolCalls,
+		})
+		toolMessages, err := a.executeToolCalls(ctx, toolCalls)
+		if err != nil {
+			return err
+		}
+		messages = append(messages, toolMessages...)
+	}
+
+	return fmt.Errorf("agent %q exceeded tool-call iteration limit %d", a.name, maxToolCallIterations)
+}
+
+// executeToolCalls 执行一组工具调用
+func (a *LLMAgent) executeToolCalls(ctx context.Context, calls []model.ChatToolCall) ([]model.ChatMessage, error) {
+	out := make([]model.ChatMessage, 0, len(calls))
+	for _, call := range calls {
+		result, err := a.chatModel.CallTool(ctx, call.Name, call.Arguments)
+		if err != nil {
+			return nil, fmt.Errorf("tool %q: %w", call.Name, err)
+		}
+		out = append(out, model.ChatMessage{
+			Role:       model.ChatRoleTool,
+			Content:    result,
+			ToolCallID: call.ID,
+			Name:       call.Name,
+		})
+	}
+	return out, nil
+}
+
+// ============================================================
+// 工作流子 Agent 接口
+// ============================================================
+
+// workflowSubAgent 工作流内部使用的 Agent 扩展接口
+type workflowSubAgent interface {
+	model.Agent
+	OutputKey() string
+	runWithVars(ctx context.Context, content model.ChatContent, vars map[string]string) (string, error)
+	streamWithVars(ctx context.Context, content model.ChatContent, out chan<- string, vars map[string]string) error
+}
+
+// ============================================================
+// SequentialAgent — 顺序工作流
+// ============================================================
+
+// SequentialAgent 顺序执行子 Agent，前一个的输出可通过 OutputKey 传递给后一个
+type SequentialAgent struct {
+	name        string
+	description string
+	subAgents   []workflowSubAgent
+}
+
+// NewSequentialAgent 创建顺序工作流 Agent
+func NewSequentialAgent(name, description string, subs []model.Agent) *SequentialAgent {
+	wrapped := make([]workflowSubAgent, 0, len(subs))
+	for _, s := range subs {
+		wrapped = append(wrapped, s.(workflowSubAgent))
+	}
+	return &SequentialAgent{name: name, description: description, subAgents: wrapped}
+}
+
+func (a *SequentialAgent) Name() string      { return a.name }
+func (a *SequentialAgent) OutputKey() string { return "" }
+
+func (a *SequentialAgent) Run(ctx context.Context, content model.ChatContent) (string, error) {
+	return a.runWithVars(ctx, content, map[string]string{})
+}
+
+func (a *SequentialAgent) Stream(ctx context.Context, content model.ChatContent, out chan<- string) error {
+	return a.streamWithVars(ctx, content, out, map[string]string{})
+}
+
+func (a *SequentialAgent) runWithVars(ctx context.Context, content model.ChatContent, vars map[string]string) (string, error) {
+	scope := cloneVars(vars)
+	var last string
+	for _, sub := range a.subAgents {
+		text, err := sub.runWithVars(ctx, content, scope)
+		if err != nil {
+			return "", err
+		}
+		last = text
+		if key := sub.OutputKey(); key != "" {
+			scope[key] = text
+		}
+	}
+	return last, nil
+}
+
+func (a *SequentialAgent) streamWithVars(ctx context.Context, content model.ChatContent, out chan<- string, vars map[string]string) error {
+	scope := cloneVars(vars)
+	for i, sub := range a.subAgents {
+		if i == len(a.subAgents)-1 {
+			return sub.streamWithVars(ctx, content, out, scope)
+		}
+		text, err := sub.runWithVars(ctx, content, scope)
+		if err != nil {
+			return err
+		}
+		if key := sub.OutputKey(); key != "" {
+			scope[key] = text
+		}
+	}
+	return nil
+}
+
+// ============================================================
+// ParallelAgent — 并行工作流
+// ============================================================
+
+// ParallelAgent 并行执行所有子 Agent 并汇总结果
+type ParallelAgent struct {
+	name        string
+	description string
+	subAgents   []workflowSubAgent
+}
+
+// NewParallelAgent 创建并行工作流 Agent
+func NewParallelAgent(name, description string, subs []model.Agent) *ParallelAgent {
+	wrapped := make([]workflowSubAgent, 0, len(subs))
+	for _, s := range subs {
+		wrapped = append(wrapped, s.(workflowSubAgent))
+	}
+	return &ParallelAgent{name: name, description: description, subAgents: wrapped}
+}
+
+func (a *ParallelAgent) Name() string      { return a.name }
+func (a *ParallelAgent) OutputKey() string { return "" }
+
+func (a *ParallelAgent) Run(ctx context.Context, content model.ChatContent) (string, error) {
+	return a.runWithVars(ctx, content, map[string]string{})
+}
+
+func (a *ParallelAgent) Stream(ctx context.Context, content model.ChatContent, out chan<- string) error {
+	return a.streamWithVars(ctx, content, out, map[string]string{})
+}
+
+func (a *ParallelAgent) runWithVars(ctx context.Context, content model.ChatContent, vars map[string]string) (string, error) {
+	parts := make([]string, 0, len(a.subAgents))
+	for _, sub := range a.subAgents {
+		text, err := sub.runWithVars(ctx, content, vars)
+		if err != nil {
+			return "", err
+		}
+		parts = append(parts, fmt.Sprintf("[%s] %s", sub.Name(), text))
+	}
+	return strings.Join(parts, "\n"), nil
+}
+
+func (a *ParallelAgent) streamWithVars(ctx context.Context, content model.ChatContent, out chan<- string, vars map[string]string) error {
+	text, err := a.runWithVars(ctx, content, vars)
+	if err != nil {
+		return err
+	}
+	select {
+	case out <- text:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
+// ============================================================
+// LoopAgent — 循环工作流
+// ============================================================
+
+// LoopAgent 循环执行子 Agent，最多执行 maxIterations 次
+type LoopAgent struct {
+	name          string
+	description   string
+	subAgents     []workflowSubAgent
+	maxIterations int
+}
+
+// NewLoopAgent 创建循环工作流 Agent
+func NewLoopAgent(name, description string, subs []model.Agent, maxIterations int) *LoopAgent {
+	if maxIterations <= 0 {
+		maxIterations = 3
+	}
+	wrapped := make([]workflowSubAgent, 0, len(subs))
+	for _, s := range subs {
+		wrapped = append(wrapped, s.(workflowSubAgent))
+	}
+	return &LoopAgent{
+		name:          name,
+		description:   description,
+		subAgents:     wrapped,
+		maxIterations: maxIterations,
+	}
+}
+
+func (a *LoopAgent) Name() string      { return a.name }
+func (a *LoopAgent) OutputKey() string { return "" }
+
+func (a *LoopAgent) Run(ctx context.Context, content model.ChatContent) (string, error) {
+	return a.runWithVars(ctx, content, map[string]string{})
+}
+
+func (a *LoopAgent) Stream(ctx context.Context, content model.ChatContent, out chan<- string) error {
+	return a.streamWithVars(ctx, content, out, map[string]string{})
+}
+
+func (a *LoopAgent) runWithVars(ctx context.Context, content model.ChatContent, vars map[string]string) (string, error) {
+	var last string
+	for i := 0; i < a.maxIterations; i++ {
+		parts := make([]string, 0, len(a.subAgents))
+		for _, sub := range a.subAgents {
+			text, err := sub.runWithVars(ctx, content, vars)
+			if err != nil {
+				return "", err
+			}
+			parts = append(parts, fmt.Sprintf("[%s] %s", sub.Name(), text))
+		}
+		last = strings.Join(parts, "\n")
+	}
+	return last, nil
+}
+
+func (a *LoopAgent) streamWithVars(ctx context.Context, content model.ChatContent, out chan<- string, vars map[string]string) error {
+	text, err := a.runWithVars(ctx, content, vars)
+	if err != nil {
+		return err
+	}
+	select {
+	case out <- text:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
+// ============================================================
+// 工具函数
+// ============================================================
+
+var sessionCounter atomic.Uint64
+
+// cloneVars 克隆变量映射
+func cloneVars(vars map[string]string) map[string]string {
+	out := make(map[string]string, len(vars)+4)
+	for k, v := range vars {
+		out[k] = v
+	}
+	return out
+}
+
+// applyVars 替换模板中的 {key} 占位符
+func applyVars(template string, vars map[string]string) string {
+	if template == "" || len(vars) == 0 {
+		return template
+	}
+	out := template
+	for k, v := range vars {
+		out = strings.ReplaceAll(out, "{"+k+"}", v)
+	}
+	return out
+}
+
+// initialMessages 构建初始消息列表（system + user）
+func initialMessages(instruction, userText string) []model.ChatMessage {
+	messages := make([]model.ChatMessage, 0, 2)
+	if strings.TrimSpace(instruction) != "" {
+		messages = append(messages, model.ChatMessage{Role: model.ChatRoleSystem, Content: instruction})
+	}
+	messages = append(messages, model.ChatMessage{Role: model.ChatRoleUser, Content: userText})
+	return messages
+}
+
+// firstText 从 ChatContent 中提取第一段文本
+func firstText(content model.ChatContent) string {
+	if len(content.Texts) == 0 {
+		return ""
+	}
+	return content.Texts[0].Message
+}
--- a/internal/service/assembler.go
+++ b/internal/service/assembler.go
@@ -1 +1,110 @@
 package service
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"strings"
+	"time"
+
+	"ai-agent-scaffold-go/internal/config"
+	"ai-agent-scaffold-go/internal/llm"
+	"ai-agent-scaffold-go/internal/model"
+)
+
+// AssembleAll 从配置表批量组装 Agent
+func AssembleAll(ctx context.Context, tables map[string]model.AiAgentConfigTable, timeout time.Duration) ([]model.RegisteredAgent, error) {
+	var agents []model.RegisteredAgent
+	for _, table := range tables {
+		agent, err := assembleOne(ctx, table, timeout)
+		if err != nil {
+			return nil, fmt.Errorf("assemble %s: %w", table.AppName, err)
+		}
+		agents = append(agents, *agent)
+	}
+	return agents, nil
+}
+
+// assembleOne 组装单个 Agent 配置
+func assembleOne(ctx context.Context, table model.AiAgentConfigTable, timeout time.Duration) (*model.RegisteredAgent, error) {
+	apiCfg := table.Module.AiAPI
+	completionsURL := strings.TrimRight(apiCfg.BaseURL, "/") + "/" + strings.TrimLeft(apiCfg.CompletionsPath, "/")
+
+	// 1. 创建 OpenAI 客户端
+	client := llm.NewOpenAIClient(completionsURL, apiCfg.APIKey, table.Module.ChatModel.Model, timeout)
+
+	// 2. 创建 ChatModel（当前无外部工具，后续可扩展）
+	chatModel := llm.NewChatModelAdapter(client, nil)
+
+	// 3. 构建 Agent 映射表
+	agentMap := map[string]model.Agent{}
+	for _, agentCfg := range table.Module.Agents {
+		agent := NewLLMAgent(agentCfg.Name, agentCfg.Instruction, agentCfg.Description, agentCfg.OutputKey, chatModel)
+		agentMap[agentCfg.Name] = agent
+	}
+
+	// 4. 构建 Workflow Agent
+	for _, wfCfg := range table.Module.AgentWorkflows {
+		subs := make([]model.Agent, 0, len(wfCfg.SubAgents))
+		for _, subName := range wfCfg.SubAgents {
+			sub, ok := agentMap[subName]
+			if !ok {
+				return nil, fmt.Errorf("workflow %q references unknown agent %q", wfCfg.Name, subName)
+			}
+			subs = append(subs, sub)
+		}
+		var wfAgent model.Agent
+		switch wfCfg.Type {
+		case model.WorkflowTypeSequential:
+			wfAgent = NewSequentialAgent(wfCfg.Name, wfCfg.Description, subs)
+		case model.WorkflowTypeParallel:
+			wfAgent = NewParallelAgent(wfCfg.Name, wfCfg.Description, subs)
+		case model.WorkflowTypeLoop:
+			wfAgent = NewLoopAgent(wfCfg.Name, wfCfg.Description, subs, wfCfg.MaxIterations)
+		default:
+			return nil, fmt.Errorf("unknown workflow type: %s", wfCfg.Type)
+		}
+		agentMap[wfCfg.Name] = wfAgent
+	}
+
+	// 5. 解析入口 Agent
+	entryName := table.Module.Runner.AgentName
+	entryAgent, ok := agentMap[entryName]
+	if !ok {
+		return nil, fmt.Errorf("entry agent %q not found", entryName)
+	}
+
+	// 6. 创建 Runner
+	runner := NewRunner(table.AppName, entryAgent)
+
+	return &model.RegisteredAgent{
+		AppName:   table.AppName,
+		AgentID:   table.Agent.AgentID,
+		AgentName: table.Agent.AgentName,
+		AgentDesc: table.Agent.AgentDesc,
+		Runner:    runner,
+	}, nil
+}
+
+// LoadAndAssemble 从配置文件路径列表加载并组装所有 Agent
+func LoadAndAssemble(ctx context.Context, paths []string, timeout time.Duration) ([]model.RegisteredAgent, error) {
+	merged := make(map[string]model.AiAgentConfigTable)
+	for _, raw := range paths {
+		path := strings.TrimSpace(raw)
+		if path == "" {
+			continue
+		}
+		expanded := os.ExpandEnv(path)
+		tables, err := config.LoadAgentTablesFile(expanded)
+		if err != nil {
+			return nil, err
+		}
+		for name, table := range tables {
+			merged[name] = table
+		}
+	}
+	if len(merged) == 0 {
+		return nil, fmt.Errorf("no agent tables loaded")
+	}
+	return AssembleAll(ctx, merged, timeout)
+}
--- a/internal/service/chat.go
+++ b/internal/service/chat.go
@@ -1 +1,110 @@
 package service
+
+import (
+	"fmt"
+	"sort"
+
+	"ai-agent-scaffold-go/internal/model"
+	"ai-agent-scaffold-go/pkg/types"
+)
+
+// ChatService 聊天服务
+type ChatService struct {
+	registry model.AgentRegistry
+	sessions model.SessionStore
+}
+
+// NewChatService 创建聊天服务
+func NewChatService(registry model.AgentRegistry, sessions model.SessionStore) *ChatService {
+	return &ChatService{registry: registry, sessions: sessions}
+}
+
+// QueryAgentConfigList 查询已注册的 Agent 列表
+func (s *ChatService) QueryAgentConfigList() []model.AgentSummary {
+	registered := s.registry.List()
+	sort.Slice(registered, func(i, j int) bool {
+		return registered[i].AgentID < registered[j].AgentID
+	})
+	agents := make([]model.AgentSummary, 0, len(registered))
+	for _, agent := range registered {
+		agents = append(agents, model.AgentSummary{
+			AgentID:   agent.AgentID,
+			AgentName: agent.AgentName,
+			AgentDesc: agent.AgentDesc,
+		})
+	}
+	return agents
+}
+
+// CreateSession 为指定 Agent 和用户创建会话
+func (s *ChatService) CreateSession(agentID, userID string) (string, error) {
+	if sessionID, ok := s.sessions.Get(userID, agentID); ok {
+		return sessionID, nil
+	}
+	registered, ok := s.registry.Get(agentID)
+	if !ok || registered.Runner == nil {
+		return "", types.NewAppError(types.CodeAgentNotFound, types.InfoAgentNotFound)
+	}
+	sessionID, err := registered.Runner.CreateSession(userID)
+	if err != nil {
+		return "", err
+	}
+	if err := s.sessions.Set(userID, agentID, sessionID); err != nil {
+		return "", err
+	}
+	return sessionID, nil
+}
+
+// HandleMessage 处理同步聊天消息
+func (s *ChatService) HandleMessage(agentID, userID, sessionID, message string) ([]string, error) {
+	content := model.ChatContent{Texts: []model.TextPart{{Message: message}}}
+	return s.handleCommand(agentID, userID, sessionID, message, content)
+}
+
+// HandleMessageStream 处理流式聊天消息
+func (s *ChatService) HandleMessageStream(agentID, userID, sessionID, message string) (<-chan string, <-chan error) {
+	content := model.ChatContent{Texts: []model.TextPart{{Message: message}}}
+	return s.handleCommandStream(agentID, userID, sessionID, message, content)
+}
+
+func (s *ChatService) handleCommand(agentID, userID, sessionID, message string, content model.ChatContent) ([]string, error) {
+	registered, sessionID, err := s.resolveRunnerSession(agentID, userID, sessionID, message, content)
+	if err != nil {
+		return nil, err
+	}
+	return registered.Runner.Run(userID, sessionID, content)
+}
+
+func (s *ChatService) handleCommandStream(agentID, userID, sessionID, message string, content model.ChatContent) (<-chan string, <-chan error) {
+	registered, sessionID, err := s.resolveRunnerSession(agentID, userID, sessionID, message, content)
+	if err != nil {
+		outputs := make(chan string)
+		errs := make(chan error, 1)
+		errs <- err
+		close(outputs)
+		close(errs)
+		return outputs, errs
+	}
+	return registered.Runner.Stream(userID, sessionID, content)
+}
+
+func (s *ChatService) resolveRunnerSession(agentID, userID, sessionID, message string, content model.ChatContent) (model.RegisteredAgent, string, error) {
+	registered, ok := s.registry.Get(agentID)
+	if !ok || registered.Runner == nil {
+		return model.RegisteredAgent{}, "", types.NewAppError(types.CodeAgentNotFound, types.InfoAgentNotFound)
+	}
+	if sessionID == "" {
+		var err error
+		sessionID, err = s.CreateSession(agentID, userID)
+		if err != nil {
+			return model.RegisteredAgent{}, "", err
+		}
+	}
+	if len(content.Texts) == 0 && message != "" {
+		content.Texts = []model.TextPart{{Message: message}}
+	}
+	if len(content.Texts) == 0 {
+		return model.RegisteredAgent{}, "", fmt.Errorf("chat content is required")
+	}
+	return registered, sessionID, nil
+}
--- a/internal/service/runner.go
+++ b/internal/service/runner.go
@@ -1 +1,68 @@
 package service
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"ai-agent-scaffold-go/internal/model"
+)
+
+// RunnerImpl Runner 的默认实现
+type RunnerImpl struct {
+	appName string
+	agent   model.Agent
+}
+
+// NewRunner 创建 Runner
+func NewRunner(appName string, agent model.Agent) *RunnerImpl {
+	return &RunnerImpl{
+		appName: appName,
+		agent:   agent,
+	}
+}
+
+// CreateSession 创建会话 ID
+func (r *RunnerImpl) CreateSession(userID string) (string, error) {
+	if strings.TrimSpace(userID) == "" {
+		return "", fmt.Errorf("user id is required")
+	}
+	next := sessionCounter.Add(1)
+	return fmt.Sprintf("%s:%s:%d", r.appName, userID, next), nil
+}
+
+// Run 同步执行
+func (r *RunnerImpl) Run(userID, sessionID string, content model.ChatContent) ([]string, error) {
+	if strings.TrimSpace(userID) == "" || strings.TrimSpace(sessionID) == "" {
+		return nil, fmt.Errorf("user id and session id are required")
+	}
+	output, err := r.agent.Run(context.Background(), content)
+	if err != nil {
+		return nil, err
+	}
+	if output == "" {
+		return []string{}, nil
+	}
+	return []string{output}, nil
+}
+
+// Stream 流式执行
+func (r *RunnerImpl) Stream(userID, sessionID string, content model.ChatContent) (<-chan string, <-chan error) {
+	outputs := make(chan string, 8)
+	errs := make(chan error, 1)
+
+	go func() {
+		defer close(outputs)
+		defer close(errs)
+
+		if strings.TrimSpace(userID) == "" || strings.TrimSpace(sessionID) == "" {
+			errs <- fmt.Errorf("user id and session id are required")
+			return
+		}
+		if err := r.agent.Stream(context.Background(), content, outputs); err != nil {
+			errs <- err
+		}
+	}()
+
+	return outputs, errs
+}