rock/poimen-workflows
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
447951daca7cefb6cc7d83cbf03d83a01777e851
poimen-workflows/docs/TEMPORAL_GRAPH_RAG_WIRING.md
T

1014 lines
31 KiB
Markdown
Raw Normal View History

docs: temporal + graph RAG integration with unified query
2026-09-05 05:45:17 -07:00
# Temporal.io + Graph RAG Integration for Workflows
## Architecture
```
┌─────────────────────────────────┐
│ Workflow API Request │
│ POST /workflows/{id}/query │
└──────────────┬──────────────────┘
│ {query, search_type, relation_type, version}
▼
┌─────────────────────────────────┐
│ Temporal Workflow Orchestrator│
│ - Start execution │
│ - Route to activities │
│ - Aggregate results │
└──────────────┬──────────────────┘
│
┌──────┴──────┐
▼ ▼
┌────────────┐ ┌─────────────────┐
│ Activity 1 │ │ Activity 2 │
│ Retrieve │ │ Query Graph RAG │
│ Entities │ │ for Relations │
└────────────┘ └─────────────────┘
│ │
▼ ▼
┌────────────────────────────────┐
│ Memory API │
│ /memory/query/semantic │
│ /memory/entities/{id}/v.. │
└────────────────────────────────┘
│
▼
┌────────────────────────────────┐
│ Graph RAG (PostgreSQL) │
│ - workflow_relations │
│ - workflow_versions │
│ - relation_changes │
└────────────────────────────────┘
```
---
## 1. Temporal Workflow Definition
**File:** `statemachine/workflow_graph_query.go`
```go
package statemachine
import (
"context"
"time"
"go.temporal.io/sdk/workflow"
"github.com/rockliang/poimen/workflows/action"
"github.com/rockliang/poimen/workflows/pkg/db"
)
// WorkflowGraphQueryInput defines query parameters
type WorkflowGraphQueryInput struct {
WorkflowID string `json:"workflow_id"`
Query string `json:"query"`
SearchType string `json:"search_type"` // entities, edges, all
RelationType string `json:"relation_type"` // data-flow, dependency, etc
Version int `json:"version"` // canvas version
ConfidenceFloor float64 `json:"confidence_floor"`
TopK int `json:"top_k"`
FindPaths bool `json:"find_paths"`
TargetNodeID string `json:"target_node_id"`
MaxPathDepth int `json:"max_path_depth"`
RankingProfile string `json:"ranking_profile"` // default, recency, accuracy
IncludeReasoning bool `json:"include_reasoning"`
}
// WorkflowGraphQueryOutput aggregates results
type WorkflowGraphQueryOutput struct {
WorkflowID string `json:"workflow_id"`
Query string `json:"query"`
Version int `json:"version"`
ExecutionTimeMs int64 `json:"execution_time_ms"`
Results []EdgeWithWording `json:"results"`
Paths []QueryPath `json:"paths"`
TotalCount int `json:"total_count"`
HasMore bool `json:"has_more"`
RankingProfile string `json:"ranking_profile"`
}
type QueryPath struct {
SourceID string `json:"source_id"`
TargetID string `json:"target_id"`
PathCount int `json:"path_count"`
Distance int `json:"distance"`
Confidence float64 `json:"total_confidence"`
PathNodes []string `json:"node_ids"`
}
// WorkflowGraphQuery is the main Temporal workflow
func WorkflowGraphQuery(ctx workflow.Context, input WorkflowGraphQueryInput) (WorkflowGraphQueryOutput, error) {
startTime := time.Now()
output := WorkflowGraphQueryOutput{
WorkflowID: input.WorkflowID,
Query: input.Query,
Version: input.Version,
RankingProfile: input.RankingProfile,
Results: []EdgeWithWording{},
}
// Activity options
opts := workflow.ActivityOptions{
StartToCloseTimeout: 120 * time.Second,
RetryPolicy: &temporal.RetryPolicy{
InitialInterval: 2 * time.Second,
BackoffCoefficient: 2.0,
MaxInterval: 10 * time.Second,
MaxAttempts: 3,
},
}
ctx = workflow.WithActivityOptions(ctx, opts)
// Step 1: Fetch Canvas + Relations from Database
var canvasData action.CanvasWithRelationsData
err := workflow.ExecuteActivity(
ctx,
action.FetchCanvasRelationsActivity,
action.FetchCanvasRelationsInput{
WorkflowID: input.WorkflowID,
Version: input.Version,
},
).Get(ctx, &canvasData)
if err != nil {
return output, err
}
// Step 2: Query Graph RAG based on search_type
var graphResults action.GraphRAGQueryOutput
graphQuery := action.GraphRAGQueryInput{
WorkflowID: input.WorkflowID,
Query: input.Query,
SearchType: input.SearchType,
RelationType: input.RelationType,
ConfidenceFloor: input.ConfidenceFloor,
TopK: input.TopK,
RankingProfile: input.RankingProfile,
Canvas: canvasData,
}
err = workflow.ExecuteActivity(
ctx,
action.QueryGraphRAGActivity,
graphQuery,
).Get(ctx, &graphResults)
if err != nil {
return output, err
}
output.Results = graphResults.Edges
output.TotalCount = graphResults.TotalCount
output.HasMore = graphResults.HasMore
// Step 3: If find_paths requested, calculate paths
if input.FindPaths && input.TargetNodeID != "" {
var pathResults action.PathfindingOutput
err = workflow.ExecuteActivity(
ctx,
action.FindRelationPathsActivity,
action.FindRelationPathsInput{
WorkflowID: input.WorkflowID,
Version: input.Version,
SourceNodes: extractSourceNodes(graphResults.Edges),
TargetNodeID: input.TargetNodeID,
MaxDepth: input.MaxPathDepth,
Canvas: canvasData,
Relations: graphResults.Edges,
},
).Get(ctx, &pathResults)
if err != nil {
return output, err
}
output.Paths = pathResults.Paths
}
// Step 4: If reasoning requested, explain results
if input.IncludeReasoning {
var reasoning action.ReasoningOutput
err = workflow.ExecuteActivity(
ctx,
action.ExplainGraphResultsActivity,
action.ExplainGraphResultsInput{
Query: input.Query,
Results: output.Results,
Paths: output.Paths,
Confidence: calculateAverageConfidence(output.Results),
},
).Get(ctx, &reasoning)
if err != nil {
// Log but don't fail if reasoning fails
workflow.GetLogger(ctx).Warn("Reasoning failed", "error", err)
}
}
output.ExecutionTimeMs = time.Since(startTime).Milliseconds()
return output, nil
}
func extractSourceNodes(edges []EdgeWithWording) []string {
nodeSet := make(map[string]bool)
for _, edge := range edges {
nodeSet[edge.Source] = true
}
var nodes []string
for node := range nodeSet {
nodes = append(nodes, node)
}
return nodes
}
func calculateAverageConfidence(edges []EdgeWithWording) float64 {
if len(edges) == 0 {
return 0.0
}
sum := 0.0
for _, edge := range edges {
sum += edge.RelationWording.Confidence
}
return sum / float64(len(edges))
}
```
---
## 2. Activity: Fetch Canvas + Relations
**File:** `action/fetch_canvas_relations.go`
```go
package action
import (
"context"
"fmt"
"github.com/rockliang/poimen/workflows/pkg/db"
)
type CanvasWithRelationsData struct {
WorkflowID string `json:"workflow_id"`
Version int `json:"version"`
Nodes []db.WorkflowNode `json:"nodes"`
Edges []EdgeWithWording `json:"edges"`
CreatedAt string `json:"created_at"`
Metadata map[string]interface{} `json:"metadata"`
}
type FetchCanvasRelationsInput struct {
WorkflowID string `json:"workflow_id"`
Version int `json:"version"`
}
// FetchCanvasRelationsActivity retrieves canvas + versioned relations
func FetchCanvasRelationsActivity(ctx context.Context, input FetchCanvasRelationsInput) (CanvasWithRelationsData, error) {
logger := newActivityLogger(ctx)
output := CanvasWithRelationsData{
WorkflowID: input.WorkflowID,
Version: input.Version,
Nodes: []db.WorkflowNode{},
Edges: []EdgeWithWording{},
}
logger.logf("info", "Fetching canvas relations for workflow %s version %d", input.WorkflowID, input.Version)
// Get database connection (injected via Temporal context)
dbConn, err := getDBFromContext(ctx)
if err != nil {
return output, fmt.Errorf("failed to get DB connection: %w", err)
}
// Fetch workflow version + canvas
var canvas db.Canvas
query := `
SELECT workflow_id, version, nodes, edges, created_at, metadata
FROM workflow_versions
WHERE workflow_id = $1 AND version = $2
`
rows, err := dbConn.Query(ctx, query, input.WorkflowID, input.Version)
if err != nil {
return output, fmt.Errorf("failed to query workflow_versions: %w", err)
}
defer rows.Close()
if !rows.Next() {
return output, fmt.Errorf("workflow version not found: %s v%d", input.WorkflowID, input.Version)
}
// Parse canvas JSONB
if err := rows.Scan(&canvas.WorkflowID, &canvas.Version, &canvas.Nodes, &canvas.Edges, &canvas.CreatedAt, &canvas.Metadata); err != nil {
return output, fmt.Errorf("failed to scan canvas: %w", err)
}
output.Nodes = canvas.Nodes
output.CreatedAt = canvas.CreatedAt.String()
// Fetch relations (edges with wording)
relQuery := `
SELECT id, source_node_id, target_node_id, relation_type, relation_label,
relation_wording, metadata, created_at
FROM workflow_relations
WHERE workflow_id = $1 AND version = $2
ORDER BY created_at
`
relRows, err := dbConn.Query(ctx, relQuery, input.WorkflowID, input.Version)
if err != nil {
return output, fmt.Errorf("failed to query workflow_relations: %w", err)
}
defer relRows.Close()
for relRows.Next() {
var edge EdgeWithWording
if err := relRows.Scan(
&edge.ID,
&edge.Source,
&edge.Target,
&edge.RelationType,
&edge.RelationLabel,
&edge.RelationWording,
&edge.Metadata,
&edge.CreatedAt,
); err != nil {
logger.logf("warn", "Failed to scan relation: %v", err)
continue
}
output.Edges = append(output.Edges, edge)
}
logger.logf("info", "Fetched %d nodes, %d relations", len(output.Nodes), len(output.Edges))
return output, nil
}
```
---
## 3. Activity: Query Graph RAG
**File:** `action/query_graph_rag.go`
```go
package action
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"net/http"
)
type GraphRAGQueryInput struct {
WorkflowID string `json:"workflow_id"`
Query string `json:"query"`
SearchType string `json:"search_type"` // entities, edges, all
RelationType string `json:"relation_type"`
ConfidenceFloor float64 `json:"confidence_floor"`
TopK int `json:"top_k"`
RankingProfile string `json:"ranking_profile"`
Canvas CanvasWithRelationsData `json:"canvas"`
}
type GraphRAGQueryOutput struct {
WorkflowID string `json:"workflow_id"`
Query string `json:"query"`
Edges []EdgeWithWording `json:"results"`
TotalCount int `json:"total_count"`
HasMore bool `json:"has_more"`
ExecutionMs int64 `json:"execution_time_ms"`
}
// QueryGraphRAGActivity calls Memory System unified query
func QueryGraphRAGActivity(ctx context.Context, input GraphRAGQueryInput) (GraphRAGQueryOutput, error) {
logger := newActivityLogger(ctx)
output := GraphRAGQueryOutput{
WorkflowID: input.WorkflowID,
Query: input.Query,
Edges: []EdgeWithWording{},
}
logger.logf("info", "Querying Graph RAG: %s (type: %s, confidence: %.2f)", input.Query, input.SearchType, input.ConfidenceFloor)
// Build unified query for Memory System
memoryQuery := buildMemoryUnifiedQuery(input)
// Call Memory System /memory/query endpoint
payload, err := json.Marshal(memoryQuery)
if err != nil {
return output, fmt.Errorf("failed to marshal query: %w", err)
}
// Get JWT token from context (set by worker)
token := ctx.Value("jwt_token").(string)
req, err := http.NewRequestWithContext(ctx, "POST", "http://localhost:8080/memory/query", bytes.NewReader(payload))
if err != nil {
return output, fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", token))
req.Header.Set("Content-Type", "application/json")
client := &http.Client{Timeout: 30 * time.Second}
resp, err := client.Do(req)
if err != nil {
return output, fmt.Errorf("failed to query Memory System: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != 200 {
body, _ := io.ReadAll(resp.Body)
return output, fmt.Errorf("Memory System returned %d: %s", resp.StatusCode, string(body))
}
// Parse response
var memoryResp struct {
Results []map[string]interface{} `json:"results"`
TotalCount int `json:"total_count"`
HasMore bool `json:"has_more"`
ExecutionMs int64 `json:"execution_time_ms"`
}
if err := json.NewDecoder(resp.Body).Decode(&memoryResp); err != nil {
return output, fmt.Errorf("failed to decode Memory response: %w", err)
}
// Map results to edges with wording
for _, result := range memoryResp.Results {
edge := EdgeWithWording{}
// Extract edge data from Memory result
if sourceID, ok := result["source_entity_id"].(string); ok {
edge.Source = sourceID
}
if targetID, ok := result["target_entity_id"].(string); ok {
edge.Target = targetID
}
if relType, ok := result["relation_type"].(string); ok {
edge.RelationType = relType
}
if fact, ok := result["fact"].(string); ok {
edge.RelationLabel = fact
}
if conf, ok := result["confidence"].(float64); ok {
edge.RelationWording.Confidence = conf
}
output.Edges = append(output.Edges, edge)
}
output.TotalCount = memoryResp.TotalCount
output.HasMore = memoryResp.HasMore
output.ExecutionMs = memoryResp.ExecutionMs
logger.logf("info", "Graph RAG returned %d edges", len(output.Edges))
return output, nil
}
// buildMemoryUnifiedQuery constructs Memory System unified query format
func buildMemoryUnifiedQuery(input GraphRAGQueryInput) map[string]interface{} {
query := map[string]interface{}{
"query": input.Query,
"search_type": "edges", // Default to edges for workflow relations
"confidence_floor": input.ConfidenceFloor,
"top_k": input.TopK,
"find_paths": input.RelationType != "",
"ranking_profile": input.RankingProfile,
"detect_communities": false,
"discover_facets": false,
}
// Add relation type filter if specified
if input.RelationType != "" {
if query["facet_filters"] == nil {
query["facet_filters"] = map[string]interface{}{}
}
filters := query["facet_filters"].(map[string]interface{})
filters["relation_type"] = input.RelationType
}
return query
}
```
---
## 4. Activity: Find Relation Paths
**File:** `action/find_relation_paths.go`
```go
package action
import (
"context"
"fmt"
)
type FindRelationPathsInput struct {
WorkflowID string `json:"workflow_id"`
Version int `json:"version"`
SourceNodes []string `json:"source_nodes"`
TargetNodeID string `json:"target_node_id"`
MaxDepth int `json:"max_depth"`
Canvas CanvasWithRelationsData `json:"canvas"`
Relations []EdgeWithWording `json:"relations"`
}
type PathfindingOutput struct {
Paths []QueryPath `json:"paths"`
}
// FindRelationPathsActivity computes shortest paths in relation graph
func FindRelationPathsActivity(ctx context.Context, input FindRelationPathsInput) (PathfindingOutput, error) {
logger := newActivityLogger(ctx)
output := PathfindingOutput{Paths: []QueryPath{}}
if input.TargetNodeID == "" {
return output, fmt.Errorf("target_node_id required for pathfinding")
}
logger.logf("info", "Finding paths to %s (max depth: %d)", input.TargetNodeID, input.MaxDepth)
// Build adjacency list from relations
graph := buildRelationGraph(input.Relations)
// BFS from each source to target
for _, sourceNode := range input.SourceNodes {
if sourceNode == input.TargetNodeID {
continue
}
paths := bfsShortestPaths(graph, sourceNode, input.TargetNodeID, input.MaxDepth)
for _, path := range paths {
output.Paths = append(output.Paths, path)
}
}
logger.logf("info", "Found %d paths", len(output.Paths))
return output, nil
}
func buildRelationGraph(relations []EdgeWithWording) map[string][]string {
graph := make(map[string][]string)
for _, edge := range relations {
graph[edge.Source] = append(graph[edge.Source], edge.Target)
}
return graph
}
func bfsShortestPaths(graph map[string][]string, source, target string, maxDepth int) []QueryPath {
var paths []QueryPath
queue := [][]string{{source}}
visited := make(map[string]bool)
visited[source] = true
for len(queue) > 0 {
path := queue[0]
queue = queue[1:]
if len(path) > maxDepth {
continue
}
current := path[len(path)-1]
if current == target {
paths = append(paths, QueryPath{
SourceID: source,
TargetID: target,
Distance: len(path) - 1,
PathCount: len(paths) + 1,
PathNodes: path,
})
continue
}
for _, neighbor := range graph[current] {
newPath := append([]string{}, path...)
newPath = append(newPath, neighbor)
queue = append(queue, newPath)
}
}
return paths
}
```
---
## 5. API Handler: Wire Everything Together
**File:** `internal/api/workflows_graph_query.go`
```go
package api
import (
"context"
"encoding/json"
"net/http"
"time"
"go.temporal.io/sdk/client"
"github.com/rockliang/poimen/workflows/statemachine"
)
// QueryWorkflowGraph handles POST /workflows/{id}/query
func (s *Server) QueryWorkflowGraph(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
workflowID := r.PathValue("id")
if workflowID == "" {
http.Error(w, "Missing workflow ID", http.StatusBadRequest)
return
}
// Parse request
var input statemachine.WorkflowGraphQueryInput
if err := json.NewDecoder(r.Body).Decode(&input); err != nil {
http.Error(w, "Invalid request body", http.StatusBadRequest)
return
}
input.WorkflowID = workflowID
// Set defaults
if input.SearchType == "" {
input.SearchType = "edges"
}
if input.TopK == 0 {
input.TopK = 10
}
if input.ConfidenceFloor == 0 {
input.ConfidenceFloor = 0.5
}
if input.MaxPathDepth == 0 {
input.MaxPathDepth = 3
}
if input.RankingProfile == "" {
input.RankingProfile = "default"
}
// Start Temporal workflow
temporalClient := s.TemporalClient // injected in NewServer
ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second)
defer cancel()
we, err := temporalClient.ExecuteWorkflow(
ctx,
client.StartWorkflowOptions{
ID: workflowID + "-query-" + time.Now().Format("20060102150405"),
TaskQueue: "workflow-tasks",
},
statemachine.WorkflowGraphQuery,
input,
)
if err != nil {
http.Error(w, "Failed to start workflow", http.StatusInternalServerError)
return
}
// Wait for result
var result statemachine.WorkflowGraphQueryOutput
if err := we.Get(ctx, &result); err != nil {
http.Error(w, "Workflow failed", http.StatusInternalServerError)
return
}
// Return result
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
json.NewEncoder(w).Encode(result)
}
// GetWorkflowRelationVersions handles GET /workflows/{id}/relations/{edge_id}/versions
func (s *Server) GetWorkflowRelationVersions(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
workflowID := r.PathValue("id")
edgeID := r.PathValue("edge_id")
if workflowID == "" || edgeID == "" {
http.Error(w, "Missing workflow or edge ID", http.StatusBadRequest)
return
}
// Query database for edge versions
query := `
SELECT version_num, operation, snapshot, changed_at, changed_by, fields_changed
FROM workflow_relation_versions
WHERE workflow_id = $1 AND edge_id = $2
ORDER BY version_num
`
rows, err := s.DB.Query(r.Context(), query, workflowID, edgeID)
if err != nil {
http.Error(w, "Failed to query versions", http.StatusInternalServerError)
return
}
defer rows.Close()
type Version struct {
VersionNum int `json:"version_num"`
Operation string `json:"operation"`
Snapshot interface{} `json:"snapshot"`
ChangedAt string `json:"changed_at"`
ChangedBy string `json:"changed_by"`
FieldsChanged []string `json:"fields_changed"`
}
versions := []Version{}
for rows.Next() {
v := Version{}
if err := rows.Scan(&v.VersionNum, &v.Operation, &v.Snapshot, &v.ChangedAt, &v.ChangedBy, &v.FieldsChanged); err != nil {
continue
}
versions = append(versions, v)
}
response := map[string]interface{}{
"edge_id": edgeID,
"workflow_id": workflowID,
"versions": versions,
"total_versions": len(versions),
"current_version": len(versions),
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(response)
}
```
---
## 6. Database Schema for Graph RAG
**File:** `migrations/004_graph_rag_relations.sql`
```sql
-- Versioned relations with wording
CREATE TABLE workflow_relations (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
workflow_id UUID NOT NULL,
version INT NOT NULL,
source_node_id VARCHAR(255) NOT NULL,
target_node_id VARCHAR(255) NOT NULL,
relation_type VARCHAR(100), -- data-flow, dependency, conditional, parallel
relation_label TEXT,
relation_wording JSONB, -- {verb, source_output, target_input, confidence, etc}
metadata JSONB,
created_at TIMESTAMP DEFAULT NOW(),
FOREIGN KEY (workflow_id, version) REFERENCES workflow_versions(workflow_id, version),
INDEX (workflow_id, version),
INDEX (relation_type),
UNIQUE (workflow_id, version, source_node_id, target_node_id)
);
-- Relation change history for versioning
CREATE TABLE workflow_relation_versions (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
workflow_id UUID NOT NULL,
edge_id UUID NOT NULL REFERENCES workflow_relations(id),
version_num INT NOT NULL,
operation VARCHAR(50), -- CREATE, UPDATE, DELETE
snapshot JSONB, -- Full relation state at this version
changed_at TIMESTAMP DEFAULT NOW(),
changed_by VARCHAR(255),
fields_changed TEXT[],
INDEX (workflow_id, version_num),
UNIQUE (workflow_id, edge_id, version_num)
);
-- Graph RAG indexing metadata
CREATE TABLE workflow_rag_index (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
workflow_id UUID NOT NULL,
version INT NOT NULL,
indexed_entities JSONB, -- {node_ids, count}
indexed_edges JSONB, -- {edge_ids, count}
index_status VARCHAR(50), -- indexed, pending, failed
last_indexed_at TIMESTAMP,
embedding_model VARCHAR(100),
INDEX (workflow_id, version)
);
```
---
## 7. Worker Configuration
**File:** `cmd/server/main.go` (additions)
```go
import (
"context"
"go.temporal.io/sdk/client"
"go.temporal.io/sdk/worker"
"github.com/rockliang/poimen/workflows/statemachine"
"github.com/rockliang/poimen/workflows/action"
)
func main() {
// ... existing setup ...
// Temporal client
c, err := client.Dial(client.Options{
HostPort: os.Getenv("TEMPORAL_HOST_PORT"), // localhost:7233
})
if err != nil {
log.Fatal(err)
}
defer c.Close()
// Worker
w := worker.New(c, "workflow-tasks", worker.Options{})
// Register workflows
w.RegisterWorkflow(statemachine.WorkflowGraphQuery)
// Register activities with DB connection in context
w.RegisterActivityWithOptions(
func(ctx context.Context, input interface{}) (interface{}, error) {
// Inject DB into context
ctx = context.WithValue(ctx, "db", dbConn)
ctx = context.WithValue(ctx, "jwt_token", os.Getenv("JWT_TOKEN"))
return nil, nil
},
activity.RegisterOptions{Name: "setup"},
)
w.RegisterActivity(action.FetchCanvasRelationsActivity)
w.RegisterActivity(action.QueryGraphRAGActivity)
w.RegisterActivity(action.FindRelationPathsActivity)
w.RegisterActivity(action.ExplainGraphResultsActivity)
// Start worker
if err := w.Start(); err != nil {
log.Fatal(err)
}
defer w.Stop()
// Register API handler
server := api.NewServer(dbConn, c)
http.HandleFunc("POST /workflows/{id}/query", server.QueryWorkflowGraph)
http.HandleFunc("GET /workflows/{id}/relations/{edge_id}/versions", server.GetWorkflowRelationVersions)
log.Fatal(http.ListenAndServe(":8081", nil))
}
```
---
## 8. Request/Response Flow Example
### Request
```bash
curl -X POST http://localhost:8081/workflows/workflow-1/query \
-H "Content-Type: application/json" \
-d '{
"query": "how does code analysis flow into security scanning",
"search_type": "edges",
"relation_type": "data-flow",
"version": 3,
"confidence_floor": 0.7,
"top_k": 10,
"find_paths": true,
"target_node_id": "security-scan-1",
"max_path_depth": 3,
"ranking_profile": "default",
"include_reasoning": true
}'
```
### Execution Flow (Temporal)
1. **API Handler** → Starts `WorkflowGraphQuery` workflow
2. **Workflow** → Calls `FetchCanvasRelationsActivity`
- Query DB: `SELECT * FROM workflow_versions WHERE workflow_id=? AND version=?`
- Query DB: `SELECT * FROM workflow_relations WHERE workflow_id=? AND version=?`
- Return: Canvas + Relations data
3. **Workflow** → Calls `QueryGraphRAGActivity`
- Build unified query for Memory System
- POST to `/memory/query` with edges search
- Return: Ranked edges with wording
4. **Workflow** → If find_paths, calls `FindRelationPathsActivity`
- BFS pathfinding in relation graph
- Return: Query paths from sources to target
5. **Workflow** → If reasoning, calls `ExplainGraphResultsActivity`
- LLM explains why results match query
- Return: Reasoning narrative
6. **Workflow** → Aggregates results, returns to API
7. **API Handler** → Returns 200 + JSON
### Response
```json
{
"workflow_id": "workflow-1",
"query": "how does code analysis flow into security scanning",
"version": 3,
"execution_time_ms": 245,
"results": [
{
"id": "edge_analyze_scan",
"source": "analyze-code-1",
"target": "security-scan-1",
"relation_type": "data-flow",
"relation_label": "AnalyzeCode outputs metrics → SecurityScan requires code structure",
"relation_wording": {
"verb": "provides-input-for",
"source_output": "metrics (object)",
"target_input": "path (string)",
"confidence": 0.85
}
}
],
"paths": [
{
"source_id": "analyze-code-1",
"target_id": "security-scan-1",
"distance": 1,
"node_ids": ["analyze-code-1", "security-scan-1"]
}
],
"total_count": 1,
"has_more": false,
"ranking_profile": "default"
}
```
---
## 9. Error Handling
```go
// Temporal retry policy
RetryPolicy: &temporal.RetryPolicy{
InitialInterval: 2 * time.Second,
BackoffCoefficient: 2.0,
MaxInterval: 10 * time.Second,
MaxAttempts: 3,
}
// Memory System errors propagate
if resp.StatusCode != 200 {
return fmt.Errorf("Memory System %d: %s", resp.StatusCode, body)
}
// Activity timeouts
StartToCloseTimeout: 120 * time.Second
```
---
## 10. Testing
**File:** `tests/workflow_graph_query_test.go`
```go
func TestWorkflowGraphQuery(t *testing.T) {
// Setup Temporal test
testSuite := &testsuite.WorkflowTestSuite{}
env := testSuite.NewTestWorkflowEnvironment()
// Mock activities
env.OnActivity(action.FetchCanvasRelationsActivity, mock.MatchedBy(func(input interface{}) bool {
return true
})).Return(CanvasWithRelationsData{
Nodes: []db.WorkflowNode{...},
Edges: []EdgeWithWording{...},
}, nil)
env.OnActivity(action.QueryGraphRAGActivity, mock.MatchedBy(func(input interface{}) bool {
return true
})).Return(GraphRAGQueryOutput{
Edges: []EdgeWithWording{...},
TotalCount: 1,
}, nil)
// Execute
env.ExecuteWorkflow(statemachine.WorkflowGraphQuery, input)
// Assert
require.True(t, env.IsWorkflowCompleted())
require.NoError(t, env.GetWorkflowError())
var result statemachine.WorkflowGraphQueryOutput
err := env.GetWorkflowResult(&result)
require.NoError(t, err)
require.Equal(t, 1, result.TotalCount)
}
```
Reference in New Issue Copy Permalink