From f532b21a7ac460eaca4dfaff6044d6f84c86158c Mon Sep 17 00:00:00 2001 From: hhs <386998068@qq.com> Date: Wed, 10 Jun 2026 15:13:21 +0800 Subject: [PATCH] =?UTF-8?q?feat(service):=20=E5=AE=9E=E7=8E=B0=E9=98=B6?= =?UTF-8?q?=E6=AE=B5=204=20=E4=B8=9A=E5=8A=A1=E9=80=BB=E8=BE=91=20?= =?UTF-8?q?=E2=80=94=20Agent=E3=80=81Runner=E3=80=81Assembler=E3=80=81Chat?= =?UTF-8?q?Service?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/service/agent.go | 421 ++++++++++++++++++++++++++++++++++ internal/service/assembler.go | 109 +++++++++ internal/service/chat.go | 109 +++++++++ internal/service/runner.go | 67 ++++++ 4 files changed, 706 insertions(+) diff --git a/internal/service/agent.go b/internal/service/agent.go index 6d43c33..28e4393 100644 --- 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 +} diff --git a/internal/service/assembler.go b/internal/service/assembler.go index 6d43c33..18c7629 100644 --- 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) +} diff --git a/internal/service/chat.go b/internal/service/chat.go index 6d43c33..3ba5e9c 100644 --- 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 +} diff --git a/internal/service/runner.go b/internal/service/runner.go index 6d43c33..cc72af5 100644 --- 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 +}