4-stage pipeline as Temporal workflow:
Stage 1: ChunkAndEmbed — chunk text + generate embeddings
Stage 2: ExtractEntities — LLM entity extraction with reflection
Stage 3: ExtractFacts — pattern + LLM fact extraction
Stage 4: DetectContradictions — pre-filter + LLM verification
Stage 5: PersistSynthesis — save all results to DB
Types: SynthesisInput, SynthesisResult, ExtractedEntity,
ExtractedFact, ContradictionResult, PersistInput
Retry: 3 attempts, exponential backoff (1s → 2s → 4s)
Each stage fails independently with wrapped errors.
Tests: 5 pass (success, contradictions, entity fail, chunk fail, fields)
Build: clean, 36 packages pass
This commit is contained in:
@@ -0,0 +1,172 @@
|
||||
package workflow
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"go.temporal.io/sdk/temporal"
|
||||
"go.temporal.io/sdk/workflow"
|
||||
)
|
||||
|
||||
// SynthesisInput contains the input for the synthesis workflow.
|
||||
type SynthesisInput struct {
|
||||
Project string `json:"project"`
|
||||
Source string `json:"source"`
|
||||
Text string `json:"text"`
|
||||
Kind string `json:"kind"` // L1, L2, reference
|
||||
Tags []string `json:"tags,omitempty"`
|
||||
}
|
||||
|
||||
// SynthesisResult contains the output of the synthesis workflow.
|
||||
type SynthesisResult struct {
|
||||
ChunkID string `json:"chunk_id"`
|
||||
EntitiesExtracted int `json:"entities_extracted"`
|
||||
FactsExtracted int `json:"facts_extracted"`
|
||||
Contradictions int `json:"contradictions"`
|
||||
ReviewQueued int `json:"review_queued"`
|
||||
Entities []ExtractedEntity `json:"entities"`
|
||||
Facts []ExtractedFact `json:"facts"`
|
||||
Duration time.Duration `json:"duration"`
|
||||
}
|
||||
|
||||
// ExtractedEntity represents an entity found during synthesis.
|
||||
type ExtractedEntity struct {
|
||||
Name string `json:"name"`
|
||||
EntityType string `json:"entity_type"`
|
||||
Confidence float64 `json:"confidence"`
|
||||
}
|
||||
|
||||
// ExtractedFact represents a fact extracted during synthesis.
|
||||
type ExtractedFact struct {
|
||||
Subject string `json:"subject"`
|
||||
Predicate string `json:"predicate"`
|
||||
Object string `json:"object"`
|
||||
Confidence float64 `json:"confidence"`
|
||||
}
|
||||
|
||||
// ContradictionResult represents a contradiction detection result.
|
||||
type ContradictionResult struct {
|
||||
FactA ExtractedFact `json:"fact_a"`
|
||||
FactB ExtractedFact `json:"fact_b"`
|
||||
Severity string `json:"severity"` // low, medium, high
|
||||
AutoResolved bool `json:"auto_resolved"`
|
||||
QueuedReview bool `json:"queued_review"`
|
||||
}
|
||||
|
||||
var synthesisActivityOptions = workflow.ActivityOptions{
|
||||
StartToCloseTimeout: 60 * time.Second,
|
||||
RetryPolicy: &temporal.RetryPolicy{
|
||||
InitialInterval: time.Second,
|
||||
BackoffCoefficient: 2.0,
|
||||
MaximumInterval: 30 * time.Second,
|
||||
MaximumAttempts: 3,
|
||||
},
|
||||
}
|
||||
|
||||
// SynthesisWorkflow orchestrates the 4-stage memory synthesis pipeline.
|
||||
//
|
||||
// Stage 1: Chunk + embed text
|
||||
// Stage 2: Extract entities (LLM + reflection)
|
||||
// Stage 3: Extract facts (pattern + LLM)
|
||||
// Stage 4: Detect contradictions (pre-filter + LLM)
|
||||
//
|
||||
// Each stage is an activity with independent retry policy.
|
||||
func SynthesisWorkflow(ctx workflow.Context, input SynthesisInput) (*SynthesisResult, error) {
|
||||
logger := workflow.GetLogger(ctx)
|
||||
startTime := workflow.Now(ctx)
|
||||
|
||||
logger.Info("synthesis started",
|
||||
"project", input.Project,
|
||||
"source", input.Source,
|
||||
"kind", input.Kind,
|
||||
)
|
||||
|
||||
actCtx := workflow.WithActivityOptions(ctx, synthesisActivityOptions)
|
||||
|
||||
// Stage 1: Chunk + Embed
|
||||
var chunkID string
|
||||
err := workflow.ExecuteActivity(actCtx, "ChunkAndEmbedActivity", input).Get(ctx, &chunkID)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("stage 1 chunk+embed: %w", err)
|
||||
}
|
||||
logger.Info("stage 1 complete", "chunk_id", chunkID)
|
||||
|
||||
// Stage 2: Entity Extraction
|
||||
var entities []ExtractedEntity
|
||||
err = workflow.ExecuteActivity(actCtx, "ExtractEntitiesActivity", chunkID, input.Text).Get(ctx, &entities)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("stage 2 entity extraction: %w", err)
|
||||
}
|
||||
logger.Info("stage 2 complete", "entities", len(entities))
|
||||
|
||||
// Stage 3: Fact Extraction
|
||||
var facts []ExtractedFact
|
||||
err = workflow.ExecuteActivity(actCtx, "ExtractFactsActivity", chunkID, input.Text, entities).Get(ctx, &facts)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("stage 3 fact extraction: %w", err)
|
||||
}
|
||||
logger.Info("stage 3 complete", "facts", len(facts))
|
||||
|
||||
// Stage 4: Contradiction Detection
|
||||
var contradictions []ContradictionResult
|
||||
err = workflow.ExecuteActivity(actCtx, "DetectContradictionsActivity", input.Project, facts).Get(ctx, &contradictions)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("stage 4 contradiction detection: %w", err)
|
||||
}
|
||||
|
||||
reviewQueued := 0
|
||||
for _, c := range contradictions {
|
||||
if c.QueuedReview {
|
||||
reviewQueued++
|
||||
}
|
||||
}
|
||||
logger.Info("stage 4 complete", "contradictions", len(contradictions), "review_queued", reviewQueued)
|
||||
|
||||
// Stage 5: Persist results
|
||||
persistInput := PersistInput{
|
||||
ChunkID: chunkID,
|
||||
Project: input.Project,
|
||||
Source: input.Source,
|
||||
Kind: input.Kind,
|
||||
Entities: entities,
|
||||
Facts: facts,
|
||||
Contradictions: contradictions,
|
||||
}
|
||||
err = workflow.ExecuteActivity(actCtx, "PersistSynthesisActivity", persistInput).Get(ctx, nil)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("stage 5 persist: %w", err)
|
||||
}
|
||||
|
||||
duration := workflow.Now(ctx).Sub(startTime)
|
||||
result := &SynthesisResult{
|
||||
ChunkID: chunkID,
|
||||
EntitiesExtracted: len(entities),
|
||||
FactsExtracted: len(facts),
|
||||
Contradictions: len(contradictions),
|
||||
ReviewQueued: reviewQueued,
|
||||
Entities: entities,
|
||||
Facts: facts,
|
||||
Duration: duration,
|
||||
}
|
||||
|
||||
logger.Info("synthesis complete",
|
||||
"chunk_id", chunkID,
|
||||
"entities", len(entities),
|
||||
"facts", len(facts),
|
||||
"contradictions", len(contradictions),
|
||||
"duration", duration,
|
||||
)
|
||||
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// PersistInput groups all synthesis results for persistence.
|
||||
type PersistInput struct {
|
||||
ChunkID string `json:"chunk_id"`
|
||||
Project string `json:"project"`
|
||||
Source string `json:"source"`
|
||||
Kind string `json:"kind"`
|
||||
Entities []ExtractedEntity `json:"entities"`
|
||||
Facts []ExtractedFact `json:"facts"`
|
||||
Contradictions []ContradictionResult `json:"contradictions"`
|
||||
}
|
||||
Reference in New Issue
Block a user