wlw/ai-agent
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Enhance workflow execution with detailed run logging and error handling, including workflow run status tracking and node tracing

Browse Source
This commit is contained in:
mlogclub
2026-06-23 22:57:18 +08:00
parent 6c02d6a79c
commit a670084374
8 changed files with 335 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/ai/application/runtime/service.go
+120 -25
View File
@@ -21,6 +21,12 @@ type Service struct {
prepare *prepareService
}
const (
workflowRunStatusCompleted = 1
workflowRunStatusInterrupted = 2
workflowRunStatusFailed = 3
)
func NewService() *Service {
catalog := newToolCatalog()
return &Service{
@@ -45,12 +51,16 @@ func (s *Service) Run(ctx context.Context, req Request) (*Summary, error) {
AIConfig: req.AIConfig,
})
if err != nil {
if workflowResult != nil {
_, _ = writeWorkflowRun(req, workflow, workflowResult, err.Error())
}
return nil, err
}
if err := writeWorkflowRun(req, workflow, workflowResult, ""); err != nil {
workflowRunID, err := writeWorkflowRun(req, workflow, workflowResult, "")
if err != nil {
return nil, err
}
return toWorkflowSummary(workflowResult, req.AIConfig.ModelName), nil
return toWorkflowSummary(workflowResult, req.AIConfig.ModelName, workflow, workflowRunID), nil
}
func (s *Service) Resume(ctx context.Context, req ResumeRequest) (*Summary, error) {
@@ -67,9 +77,26 @@ func (s *Service) Resume(ctx context.Context, req ResumeRequest) (*Summary, erro
AIConfig: req.AIConfig,
}, interrupt.RequestData, firstWorkflowResumeText(req.ResumeData))
if err != nil {
if workflowResult != nil {
_, _ = writeWorkflowRun(Request{
Conversation: req.Conversation,
UserMessage: req.UserMessage,
AIAgent: req.AIAgent,
AIConfig: req.AIConfig,
}, workflow, workflowResult, err.Error())
}
return nil, err
}
return toWorkflowSummary(workflowResult, req.AIConfig.ModelName), nil
workflowRunID, err := writeWorkflowRun(Request{
Conversation: req.Conversation,
UserMessage: req.UserMessage,
AIAgent: req.AIAgent,
AIConfig: req.AIConfig,
}, workflow, workflowResult, "")
if err != nil {
return nil, err
}
return toWorkflowSummary(workflowResult, req.AIConfig.ModelName, workflow, workflowRunID), nil
}
toolSet, err := s.prepare.prepareToolsForResume(req)
if err != nil {
@@ -99,27 +126,34 @@ func firstWorkflowResumeText(data map[string]string) string {
return ""
}
func toWorkflowSummary(result *workflowexecutor.Result, modelName string) *Summary {
func toWorkflowSummary(result *workflowexecutor.Result, modelName string, workflow resolvedWorkflow, workflowRunID int64) *Summary {
if result == nil {
return nil
}
trace := map[string]any{
"status": result.Status,
"nodePath": result.NodePath,
"status": result.Status,
"workflowId": workflow.WorkflowID,
"workflowVersionId": workflow.VersionID,
"workflowRunId": workflowRunID,
"nodePath": result.NodePath,
}
traceData, _ := json.Marshal(trace)
return &Summary{
Status: result.Status,
ReplyText: result.ReplyText,
ModelName: modelName,
PromptTokens: result.PromptTokens,
CompletionTokens: result.CompletionTokens,
RetrieverCount: result.RetrieverCount,
TraceData: string(traceData),
CheckPointID: result.CheckPointID,
CheckPointData: result.CheckPointData,
Interrupted: result.Interrupted,
Interrupts: toWorkflowInterruptSummaries(result.Interrupts),
Status: result.Status,
ReplyText: result.ReplyText,
ModelName: modelName,
PromptTokens: result.PromptTokens,
CompletionTokens: result.CompletionTokens,
RetrieverCount: result.RetrieverCount,
WorkflowID: workflow.WorkflowID,
WorkflowVersionID: workflow.VersionID,
WorkflowRunID: workflowRunID,
WorkflowNodePath: append([]string(nil), result.NodePath...),
TraceData: string(traceData),
CheckPointID: result.CheckPointID,
CheckPointData: result.CheckPointData,
Interrupted: result.Interrupted,
Interrupts: toWorkflowInterruptSummaries(result.Interrupts),
}
}
@@ -138,9 +172,9 @@ func toWorkflowInterruptSummaries(items []workflowexecutor.InterruptSummary) []I
return ret
}
func writeWorkflowRun(req Request, workflow resolvedWorkflow, result *workflowexecutor.Result, errorMessage string) error {
func writeWorkflowRun(req Request, workflow resolvedWorkflow, result *workflowexecutor.Result, errorMessage string) (int64, error) {
if result == nil {
return nil
return 0, nil
}
now := time.Now()
endedAt := now
@@ -148,29 +182,42 @@ func writeWorkflowRun(req Request, workflow resolvedWorkflow, result *workflowex
for _, node := range workflow.Definition.Nodes {
nodeTypes[node.ID] = node.Type
}
return sqls.WithTransaction(func(ctx *sqls.TxContext) error {
runStatus := workflowRunStatus(result.Status, errorMessage)
var runID int64
err := sqls.WithTransaction(func(ctx *sqls.TxContext) error {
run := &models.AIWorkflowRun{
WorkflowID: workflow.WorkflowID,
WorkflowVersionID: workflow.VersionID,
ConversationID: req.Conversation.ID,
AIAgentID: req.AIAgent.ID,
MessageID: req.UserMessage.ID,
Status: 1,
Status: runStatus,
StartedAt: now,
EndedAt: &endedAt,
InterruptType: firstWorkflowInterruptType(result),
InterruptNodeID: firstWorkflowInterruptNodeID(result),
ErrorMessage: errorMessage,
}
if err := repositories.AIWorkflowRunRepository.Create(ctx.Tx, run); err != nil {
return err
}
for _, nodeID := range result.NodePath {
runID = run.ID
nodeTraces := result.NodeTraces
if len(nodeTraces) == 0 {
nodeTraces = fallbackWorkflowNodeTraces(result.NodePath, nodeTypes, result.Status)
}
for _, nodeTrace := range nodeTraces {
nodeRun := &models.AIWorkflowNodeRun{
WorkflowRunID: run.ID,
NodeID: nodeID,
NodeType: nodeTypes[nodeID],
Status: 1,
NodeID: nodeTrace.NodeID,
NodeType: firstNonEmpty(nodeTrace.NodeType, nodeTypes[nodeTrace.NodeID]),
Status: workflowRunStatus(nodeTrace.Status, nodeTrace.ErrorMessage),
InputPreview: nodeTrace.InputPreview,
OutputPreview: nodeTrace.OutputPreview,
ErrorMessage: nodeTrace.ErrorMessage,
StartedAt: now,
EndedAt: &endedAt,
DurationMS: nodeTrace.DurationMS,
}
if err := repositories.AIWorkflowNodeRunRepository.Create(ctx.Tx, nodeRun); err != nil {
return err
@@ -178,4 +225,52 @@ func writeWorkflowRun(req Request, workflow resolvedWorkflow, result *workflowex
}
return nil
})
return runID, err
}
func workflowRunStatus(status string, errorMessage string) int {
if strings.TrimSpace(errorMessage) != "" || strings.TrimSpace(status) == "error" {
return workflowRunStatusFailed
}
switch strings.TrimSpace(status) {
case "interrupted":
return workflowRunStatusInterrupted
default:
return workflowRunStatusCompleted
}
}
func fallbackWorkflowNodeTraces(nodePath []string, nodeTypes map[string]string, status string) []workflowexecutor.NodeTrace {
ret := make([]workflowexecutor.NodeTrace, 0, len(nodePath))
for _, nodeID := range nodePath {
ret = append(ret, workflowexecutor.NodeTrace{
NodeID: nodeID,
NodeType: nodeTypes[nodeID],
Status: status,
})
}
return ret
}
func firstWorkflowInterruptType(result *workflowexecutor.Result) string {
if result == nil || len(result.Interrupts) == 0 {
return ""
}
return strings.TrimSpace(result.Interrupts[0].Type)
}
func firstWorkflowInterruptNodeID(result *workflowexecutor.Result) string {
if result == nil || len(result.Interrupts) == 0 {
return ""
}
return strings.TrimSpace(result.Interrupts[0].ID)
}
func firstNonEmpty(items ...string) string {
for _, item := range items {
if strings.TrimSpace(item) != "" {
return strings.TrimSpace(item)
}
}
return ""
}
internal/ai/application/runtime/tool_catalog_test.go
+13
View File
@@ -199,6 +199,19 @@ func TestServiceRunExecutesPublishedWorkflow(t *testing.T) {
if nodeRunCount != 4 {
t.Fatalf("expected four workflow node runs, got %d", nodeRunCount)
}
var replyNodeRun models.AIWorkflowNodeRun
if err := sqls.DB().First(&replyNodeRun, "node_id = ?", "reply_1").Error; err != nil {
t.Fatalf("find reply node run: %v", err)
}
if replyNodeRun.InputPreview == "" || replyNodeRun.OutputPreview == "" {
t.Fatalf("expected node input/output previews, got input=%q output=%q", replyNodeRun.InputPreview, replyNodeRun.OutputPreview)
}
if !strings.Contains(replyNodeRun.OutputPreview, "workflow reply") {
t.Fatalf("expected reply output preview, got %q", replyNodeRun.OutputPreview)
}
if replyNodeRun.DurationMS < 0 {
t.Fatalf("unexpected negative duration: %d", replyNodeRun.DurationMS)
}
}
func setupWorkflowRuntimeTestDB(t *testing.T) {
internal/ai/application/runtime/types.go
+5
View File
@@ -16,6 +16,7 @@ type Request struct {
type ResumeRequest struct {
Conversation models.Conversation
UserMessage models.Message
AIAgent models.AIAgent
AIConfig models.AIConfig
CheckPointID string
@@ -46,6 +47,10 @@ type Summary struct {
ToolCallCount int
ToolCodes []string
InvokedToolCodes []string
WorkflowID int64
WorkflowVersionID int64
WorkflowRunID int64
WorkflowNodePath []string
CheckPointID string
CheckPointData string
Interrupted bool
internal/ai/application/runtime/workflow_summary_test.go
+66 -2
View File
@@ -27,7 +27,7 @@ func TestToWorkflowSummaryPreservesInterruptCheckpoint(t *testing.T) {
Interrupts: []workflowexecutor.InterruptSummary{
{Type: "human_confirm", ID: "confirm_1", InfoPreview: `{"message":"请确认"}`},
},
}, "test-model")
}, "test-model", resolvedWorkflow{WorkflowID: 11, VersionID: 22}, 33)
if summary == nil || !summary.Interrupted {
t.Fatalf("expected interrupted summary, got %#v", summary)
@@ -38,6 +38,9 @@ func TestToWorkflowSummaryPreservesInterruptCheckpoint(t *testing.T) {
if summary.CheckPointData == "" {
t.Fatalf("expected checkpoint data")
}
if summary.WorkflowID != 11 || summary.WorkflowVersionID != 22 || summary.WorkflowRunID != 33 {
t.Fatalf("unexpected workflow identity: workflow=%d version=%d run=%d", summary.WorkflowID, summary.WorkflowVersionID, summary.WorkflowRunID)
}
if len(summary.Interrupts) != 1 || summary.Interrupts[0].ID != "confirm_1" {
t.Fatalf("unexpected interrupts: %#v", summary.Interrupts)
}
@@ -69,6 +72,7 @@ func TestServiceResumeUsesWorkflowCheckpointData(t *testing.T) {
summary, err := NewService().Resume(context.Background(), ResumeRequest{
Conversation: models.Conversation{ID: 1},
UserMessage: models.Message{ID: 2, Content: "确认"},
AIAgent: models.AIAgent{
ID: 1,
WorkflowVersionID: version.ID,
@@ -85,6 +89,66 @@ func TestServiceResumeUsesWorkflowCheckpointData(t *testing.T) {
if summary == nil || summary.Status != "completed" || summary.Interrupted {
t.Fatalf("unexpected summary: %#v", summary)
}
if summary.WorkflowRunID <= 0 {
t.Fatalf("expected workflow run id in resume summary")
}
var run models.AIWorkflowRun
if err := db.First(&run, summary.WorkflowRunID).Error; err != nil {
t.Fatalf("find resume workflow run: %v", err)
}
if run.MessageID != 2 || run.Status != workflowRunStatusCompleted {
t.Fatalf("unexpected resume workflow run: %#v", run)
}
}
func TestServiceRunWritesFailedWorkflowRun(t *testing.T) {
db := setupWorkflowResumeTestDB(t)
def := dsl.Definition{
SchemaVersion: 1,
EntryNodeID: "start_1",
Nodes: []dsl.Node{
{ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start"},
{ID: "bad_1", Type: "unsupported_node", Name: "Bad"},
},
Edges: []dsl.Edge{
{ID: "edge_start_bad", Source: "start_1", Target: "bad_1"},
},
}
version := models.AIWorkflowVersion{
WorkflowID: 9,
Version: 1,
Status: enums.StatusOk,
Definition: mustMarshalDefinition(t, def),
}
if err := db.Create(&version).Error; err != nil {
t.Fatalf("create workflow version: %v", err)
}
_, err := NewService().Run(context.Background(), Request{
Conversation: models.Conversation{ID: 10},
UserMessage: models.Message{ID: 20, Content: "hello"},
AIAgent: models.AIAgent{
ID: 30,
WorkflowVersionID: version.ID,
},
})
if err == nil {
t.Fatalf("expected workflow run error")
}
var run models.AIWorkflowRun
if err := db.First(&run, "workflow_version_id = ?", version.ID).Error; err != nil {
t.Fatalf("find failed workflow run: %v", err)
}
if run.Status != workflowRunStatusFailed || !strings.Contains(run.ErrorMessage, "unsupported workflow node type") {
t.Fatalf("unexpected failed workflow run: %#v", run)
}
var badNodeRun models.AIWorkflowNodeRun
if err := db.First(&badNodeRun, "workflow_run_id = ? AND node_id = ?", run.ID, "bad_1").Error; err != nil {
t.Fatalf("find failed node run: %v", err)
}
if badNodeRun.Status != workflowRunStatusFailed || badNodeRun.ErrorMessage == "" {
t.Fatalf("unexpected failed node run: %#v", badNodeRun)
}
}
func setupWorkflowResumeTestDB(t *testing.T) *gorm.DB {
@@ -105,7 +169,7 @@ func setupWorkflowResumeTestDB(t *testing.T) *gorm.DB {
_ = sqlDB.Close()
}
})
if err := db.AutoMigrate(&models.AIWorkflowVersion{}, &models.ConversationInterrupt{}); err != nil {
if err := db.AutoMigrate(&models.AIWorkflowVersion{}, &models.AIWorkflowRun{}, &models.AIWorkflowNodeRun{}, &models.ConversationInterrupt{}); err != nil {
t.Fatalf("auto migrate: %v", err)
}
sqls.SetDB(db)