What about knowledge graphs?

Build persistent intelligence networks using agents, functions, and knowledge graphs

Knowledge graphs in Modus

Your applications need more than just individual memory—they need shared organizational knowledge. While agents maintain their own operational state and memory across interactions, knowledge graphs provide something fundamentally different: shared institutional knowledge that captures relationships between entities, events, and data across your entire app.

At Hypermode, we recognize that knowledge graphs aren’t just storage—they’re becoming critical infrastructure for next-generation AI systems. That’s why we’ve invested deeply in Dgraph, bringing enterprise-grade graph capabilities to Modus applications.

This is where knowledge graphs transform your Modus deployment from isolated processes into a coordinated system with shared institutional memory.

What are knowledge networks?

Knowledge networks in Modus combine:

• Agent state: personal memory that each agent maintains across interactions
• Knowledge graphs: shared organizational knowledge that captures relationships between entities, events, and data across your entire app
• Functions: rapid operations for data processing and analysis
• AI models: advanced pattern recognition and decision-making capabilities

Think of it as the difference between what an individual process remembers versus what your organization knows. Agent state is personal memory—what happened to this specific agent, what conversations they’ve had, what tasks they’re tracking. Knowledge graphs are organizational intelligence—how entities relate to each other, which patterns connect to which outcomes, what relationships emerge across all operations.

Setting up your knowledge infrastructure

First, connect to your knowledge graph by adding this to your modus.json:

{
  "connections": {
    "dgraph": {
      "type": "dgraph",
      "connString": "dgraph://your-graph.hypermode.host:443?sslmode=verify-ca&bearertoken={{API_KEY}}"
    }
  },
  "models": {
    "text-generator": {
      "sourceModel": "meta-llama/Llama-3.2-3B-Instruct",
      "provider": "hugging-face",
      "connection": "hypermode"
    }
  }
}

Set your credentials in .env.dev.local:

MODUS_DGRAPH_API_KEY=your_graph_api_key_here

Building a comprehensive system

Let’s walk through a realistic scenario that demonstrates how all these components work together. You’re building a system to track anomalous Agent behavior in the simulated reality. The system needs to:

1. Rapidly import new Agent sightings and behavioral data
2. Find patterns across historical Agent encounters
3. Coordinate ongoing surveillance operations
4. Provide strategic analysis to the resistance

Step 1: Rapid data import

When new Agent activity is detected in the Matrix, you need to process it quickly. This is perfect for a stateless function:

Go

package main

import (
  "encoding/json"
  "fmt"
  "strings"

  "github.com/hypermodeinc/modus/sdk/go/pkg/dgraph"
  "github.com/hypermodeinc/modus/sdk/go/pkg/models"
  "github.com/hypermodeinc/modus/sdk/go/pkg/models/openai"
)

type AgentSighting struct {
  SightingID  string  `json:"sighting_id"`
  AgentName   string  `json:"agent_name"`
  Location    string  `json:"location"`
  Behavior    string  `json:"behavior"`
  ThreatLevel int     `json:"threat_level"`
  Timestamp   string  `json:"timestamp"`
  AiAnalysis  *string `json:"ai_analysis,omitempty"`
}

type ThreatAnalysisResponse struct {
  SightingCount    int      `json:"sighting_count"`
  ActiveAgents     []string `json:"active_agents"`
  ThreatAssessment string   `json:"threat_assessment"`
  Recommendations  []string `json:"recommendations"`
}

type SightingsData struct {
  Sightings []AgentSighting `json:"sightings"`
}

func ImportAgentSighting(s AgentSighting) (*string, error) {
  analysis, err := analyzeAgentWithAI(s.Behavior)
  if err != nil {
    return nil, err
  }
  s.AiAnalysis = &analysis

  data, err := json.Marshal(s)
  if err != nil {
    return nil, err
  }

  mutation := dgraph.NewMutation().WithSetJson(string(data))
  if _, err := dgraph.ExecuteMutations("dgraph", mutation); err != nil {
    return nil, err
  }

  res := fmt.Sprintf("Agent sighting processed: %s", s.SightingID)
  return &res, nil
}

func analyzeAgentWithAI(behavior string) (string, error) {
  model, err := models.GetModel[openai.ChatModel]("text-generator")
  if err != nil {
    return "", err
  }

  prompt := "Analyze this Agent behavior pattern and assess threat level, behavioral changes, and tactical implications for resistance operations. Respond with no more than 3 short sentences."
  input, err := model.CreateInput(
    openai.NewSystemMessage(prompt),
    openai.NewUserMessage(behavior),
  )
  if err != nil {
    return "", err
  }
  input.Temperature = 0.3

  out, err := model.Invoke(input)
  if err != nil {
    return "", err
  }
  return strings.TrimSpace(out.Choices[0].Message.Content), nil
}

Now let’s deploy this data import function and test it:

mutation {
  importAgentSighting(
    sighting: {
      sightingId: "SIGHT-2025-001"
      agentName: "Smith"
      location: "Downtown Loop - Financial District"
      behavior: "Unusual pattern recognition algorithm detected.
                Agent displaying enhanced replication capabilities
                beyond normal parameters."
      threatLevel: 9
      timestamp: "2025-01-15T14:30:00Z"
    }
  )
}

Response:

{
  "data": {
    "importAgentSighting": "Agent sighting processed: SIGHT-2025-001"
  }
}

Step 2: Strategic analysis using organizational knowledge

Now that we’ve got the Agent sighting data in our knowledge graph, let’s analyze the broader threat landscape:

Go

func AnalyzeAgentPatterns(since string) (*ThreatAnalysisResponse, error) {
  query := dgraph.NewQuery(`
    query analyzeAgents($since: string) {
      sightings(func: ge(timestamp, $since)) {
        sighting_id
        agent_name
        location
        behavior
        threat_level
        timestamp
        ai_analysis
      }
    }
  `).WithVariable("$since", since)

  resp, err := dgraph.ExecuteQuery("dgraph", query)
  if err != nil {
    return nil, err
  }

  var data SightingsData
  if err := json.Unmarshal([]byte(resp.Json), &data); err != nil {
    return nil, err
  }

  assessment, err := generateThreatAssessment(data.Sightings)
  if err != nil {
    return nil, err
  }

  return &ThreatAnalysisResponse{
    SightingCount:    len(data.Sightings),
    ActiveAgents:     extractActiveAgents(data.Sightings),
    ThreatAssessment: assessment,
    Recommendations:  generateRecommendations(len(data.Sightings)),
  }, nil
}

func generateThreatAssessment(sightings []AgentSighting) (string, error) {
  model, err := models.GetModel[openai.ChatModel]("text-generator")
  if err != nil {
    return "", err
  }

  prompt := "Based on these Agent sightings, provide a strategic threat assessment focusing on behavioral patterns and risks to resistance operations. Respond with no more than 2 short sentences."
  data, err := json.Marshal(sightings)
  if err != nil {
    return "", err
  }

  input, err := model.CreateInput(
    openai.NewSystemMessage(prompt),
    openai.NewUserMessage("Agent surveillance data: "+string(data)),
  )
  if err != nil {
    return "", err
  }
  input.Temperature = 0.4

  out, err := model.Invoke(input)
  if err != nil {
    return "", err
  }

  return strings.TrimSpace(out.Choices[0].Message.Content), nil
}

func extractActiveAgents(sightings []AgentSighting) []string {
  seen := map[string]struct{}{}
  var agents []string
  for _, s := range sightings {
    if _, ok := seen[s.AgentName]; !ok {
      seen[s.AgentName] = struct{}{}
      agents = append(agents, s.AgentName)
    }
  }
  return agents
}

func generateRecommendations(count int) []string {
  switch {
  case count > 5:
    return []string{
      "Implement additional security measures.",
      "Enhance threat detection capabilities.",
      "Implement enhanced access control.",
    }
  case count > 2:
    return []string{
      "Enhance threat detection capabilities.",
      "Implement enhanced access control.",
    }
  default:
    return []string{"Implement additional security measures."}
  }
}

Let’s query our surveillance data:

query {
  analyzeAgentPatterns(timeRange: "2025-01-01T00:00:00Z") {
    sightingCount
    activeAgents
    threatAssessment
    recommendations
  }
}

Response:

{
  "data": {
    "analyzeAgentPatterns": {
      "sightingCount": 1,
      "activeAgents": ["Smith"],
      "threatAssessment": "Initial Agent Smith sighting shows enhanced
                          replication capabilities beyond standard parameters.
                          Requires additional surveillance data for pattern
                          analysis and threat escalation assessment.",
      "recommendations": [
        "Continue surveillance monitoring",
        "Increase Agent activity detection"
      ]
    }
  }
}

Step 3: Automated processing with asynchronous coordination

Now let’s enhance our system to automatically coordinate surveillance when new data arrives. We’ll deploy persistent surveillance agents and upgrade our import function to trigger them:

Go

import (
  "time"
  "github.com/hypermodeinc/modus/sdk/go/pkg/agents"
)

type SurveillanceAgent struct {
  agents.AgentBase
  MonitoredSectors []string  `json:"monitored_sectors"`
  SightingsTracked int       `json:"sightings_tracked"`
  RecentActivities []string  `json:"recent_activities"`
  LastSweepTime    time.Time `json:"last_sweep_time"`
}

func (s *SurveillanceAgent) Name() string {
  return "SurveillanceAgent"
}

func (s *SurveillanceAgent) OnInitialize() error {
  s.MonitoredSectors = []string{"Downtown Loop", "Megacity Financial", "Industrial District"}
  s.SightingsTracked = 0
  s.RecentActivities = []string{}
  s.LastSweepTime = time.Now()
  return nil
}

func (s *SurveillanceAgent) OnReceiveMessage(msgName string, data *string) (*string, error) {
  switch msgName {
  case "continuous_surveillance":
    return s.processNewIntelligence()
  case "get_status":
    return s.getOperationalStatus()
  }
  return nil, fmt.Errorf("unrecognized directive: %s", msgName)
}

func (s *SurveillanceAgent) processNewIntelligence() (*string, error) {
  query := dgraph.NewQuery(`
        query getRecentSightings($since: string) {
            sightings(func: ge(timestamp, $since)) {
                agent_name
                threat_level
                location
            }
        }
    `).WithVariable("$since", s.LastSweepTime.Format(time.RFC3339))

  _, err := dgraph.ExecuteQuery("dgraph", query)
  if err != nil {
    return nil, err
  }

  s.LastSweepTime = time.Now()
  s.SightingsTracked++
  activity := fmt.Sprintf("Auto surveillance at %s", s.LastSweepTime.Format("15:04:05"))
  s.RecentActivities = append(s.RecentActivities, activity)

  if len(s.RecentActivities) > 3 {
    s.RecentActivities = s.RecentActivities[1:]
  }

  result := fmt.Sprintf("Data processed automatically.\nTracking %d sightings. Matrix integrity: COMPROMISED", s.SightingsTracked)
  return &result, nil
}

func (s *SurveillanceAgent) getOperationalStatus() (*string, error) {
  status := fmt.Sprintf(`Surveillance Agent Status:
- Operational: Active
- Monitoring %d sectors: %s
- Last sweep: %s
- Tracking %d ongoing sightings
- Recent activities: %s`,
    len(s.MonitoredSectors),
    strings.Join(s.MonitoredSectors, ", "),
    s.LastSweepTime.Format("2006-01-02 15:04:05"),
    s.SightingsTracked,
    strings.Join(s.RecentActivities, ", "),
  )
  return &status, nil
}

func (s *SurveillanceAgent) GetState() *string { return nil }
func (s *SurveillanceAgent) SetState(state *string) {}

func init() {
  agents.Register(&SurveillanceAgent{})
}

func DeploySurveillanceAgent() (string, error) {
  agentInfo, err := agents.Start("SurveillanceAgent")
  if err != nil {
    return "", err
  }
  return agentInfo.Name + " - " + agentInfo.Id, nil
}

func GetSurveillanceStatus(agentId string) (string, error) {
  result, err := agents.SendMessage(agentId, "get_status")
  if err != nil {
    return "", err
  }
  if result == nil {
    return "", fmt.Errorf("no response from agent")
  }
  return *result, nil
}

Now let’s enhance our original import function to automatically trigger surveillance:

Go

func ImportAgentSighting(s AgentSighting, surveillanceAgentId *string) (*string, error) {
  analysis, err := analyzeAgentWithAI(s.Behavior)
  if err != nil {
    return nil, err
  }
  s.AiAnalysis = &analysis

  data, err := json.Marshal(s)
  if err != nil {
    return nil, err
  }

  mutation := dgraph.NewMutation().WithSetJson(string(data))
  if _, err := dgraph.ExecuteMutations("dgraph", mutation); err != nil {
    return nil, err
  }

  if surveillanceAgentId != nil {
    if err := agents.SendMessageAsync(*surveillanceAgentId, "continuous_surveillance"); err != nil {
      return nil, err
    }
  }

  res := fmt.Sprintf("Agent sighting processed: %s", s.SightingID)
  return &res, nil
}

Deploy your surveillance agent:

mutation {
  deploySurveillanceAgent
}

Response:

{
  "data": {
    "deploySurveillanceAgent": "agent_neo_001"
  }
}

Step 4: Coordinated processing

Now when you import new Agent sightings, surveillance automatically triggers:

mutation {
  importAgentSighting(
    sighting: {
      sightingId: "SIGHT-2025-002"
      agentName: "Brown"
      location: "Megacity Financial - Server Room B12"
      behavior: "Agent Brown detected implementing advanced
                countermeasures against known resistance
                encryption protocols. Adaptive learning
                subroutines active."
      threatLevel: 8
      timestamp: "2025-01-15T15:45:00Z"
    }
  )
}

Response:

{
  "data": {
    "importAgentSighting": "Agent sighting processed: SIGHT-2025-002"
  }
}

Each import automatically triggers the surveillance agent through asynchronous messaging. Check the surveillance status:

query {
  surveillanceStatus(agentId: "agent_neo_001")
}

Response:

{
  "data": {
    "surveillanceStatus":
      "Surveillance Agent Status:
        - Operational: Active
        - Monitoring 3 sectors: Downtown Loop,
          Megacity Financial, Industrial District
        - Last sweep: 2025-01-15 15:45:22
        - Tracking 2 ongoing sightings
        - Recent activities: Auto surveillance at 14:30:05,
          Auto surveillance at 15:45:22"
  }
}

Step 5: Enhanced threat analysis

Query the strategic analysis to see patterns across automatically processed data:

query {
  analyzeAgentPatterns(timeRange: "2025-01-15T00:00:00Z") {
    sightingCount
    activeAgents
    threatAssessment
    recommendations
  }
}

Response:

{
  "data": {
    "analyzeAgentPatterns": {
      "sightingCount": 2,
      "activeAgents": ["Smith", "Brown"],
      "threatAssessment":
        "Critical escalation detected. Agent Smith's
        enhanced replication capabilities combined with
        Agent Brown's encryption countermeasures indicates
        coordinated Matrix defense upgrade. Systematic
        pattern suggests machines adapting to resistance
        operations.",
      "recommendations": [
        "Emergency extraction protocols",
        "Activate deep cover cells",
        "Increase surveillance sweeps"
      ]
    }
  }
}

Conclusion

You’ve just built a complete automated surveillance network that demonstrates the power of coordinated systems. By combining functions for rapid data processing, knowledge graphs for organizational memory, AI models for enhanced analysis, and agents for persistent processing—all coordinated through asynchronous messaging—you’ve created something far more powerful than any single component could achieve.

Your system now automatically processes Agent sightings, triggers surveillance operations, builds organizational knowledge over time, and provides AI-enhanced threat analysis across all accumulated data. The surveillance agent maintains persistent memory across system failures while the knowledge graph captures relationships that no single sighting could reveal.

This isn’t just a database with some AI on top—it’s a coordinated system where each component enhances the others, creating emergent capabilities that scale with your operations. Welcome to the real world.

Next steps

Ready to deploy your surveillance network against the machines? Check out:

• Dgraph integration guide for advanced graph operations
• Agent coordination patterns for multi-agent workflows
• Production deployment for scaling your knowledge network