Ingestion from APIs and Live Streams: Real-Time Knowledge

A static Knowledge Graph is a museum. It represents what was true when the ingestion ran. But for an AI agent in a fast-moving enterprise, "State" changes every minute. A ticket is closed on Jira. A PR is merged on GitHub. A Slack channel erupts with a new incident.

In this lesson, we will move from "Batch Ingestion" to Streaming Ingestion. We will learn how to build "Connectors" for Slack and GitHub, how to handle "Real-Time Updates" to node properties, and how to maintain graph consistency when the data is arriving in a chaotic, asynchronous stream.

---

1. The Connector Pattern: Monitoring the Source

To feed a graph in real-time, you need a Listener.

• Webhooks: The source (e.g., GitHub) "Pushes" a notification to your server whenever an event happens (e.g., Issue Opened).
• Polling: Your server "Pulls" data from an API every 60 seconds (useful for sources that don't support webhooks).

The Task: When an event arrives, your code must "Diff" it against the current graph and update the relevant nodes and edges.

---

2. Upsert Logic: MERGE vs. CREATE

In a live stream, you will often receive the same entity multiple times.

• Example: 50 Slack messages from "Sudeep."
• Bad: Creating 50 nodes named "Sudeep."
• Good: Using the UPSERT (Update or Insert) pattern. In Cypher, this is the MERGE command.

MERGE (p:Person {id: 'Sudeep'}) ON MATCH SET p.lastActive = timestamp()

This ensures that "Sudeep" remains a single, high-fidelity node while his "Last Active" property stays current.

---

3. Handling Event Ordering (The Time Problem)

In a live stream, Message #2 might arrive before Message #1 due to network lag.

• Scenario:
  • Message 1: "The status is Green." (Sent at 12:00)
  • Message 2: "The status is Red." (Sent at 12:01)
• If Message 2 arrives first, your graph will correctly show "Red." But when Message 1 arrives a second later, it might "Overwrite" it to "Green" (Incorrect).

The Solution: Always store a version or timestamp property on nodes. Only update if the incoming event is newer than the stored one.

graph TD
    S[Slack/GitHub API] -->|Webhook| LW[Listener Worker]
    LW -->|Queue| B[Buffer]
    B -->|Check Version| KG[(Knowledge Graph)]
    KG -->|Trigger| AA[AI Agent Re-indexing]
    
    style S fill:#f4b400,color:#fff
    style KG fill:#4285F4,color:#fff

---

4. The "Episodic" Memory Edge

For an AI agent, knowing that something changed isn't enough. It needs to know what changed.

Instead of just updating a property, you can add a Temporal Edge. (Sudeep) -[:POSTED {at: '12:00', text: '...'}]-> (SlackChannel)

By keeping these "Event Edges," you allow the AI to answer Temporal Reasoning questions: "What was the sequence of events leading up to the server crash?"

---

5. Implementation: A Pseudo-Slack Ingester with Python

Let's look at how we process a simulated message stream.

import time

# Simulation of a incoming event stream
event_stream = [
    {"user": "Sudeep", "msg": "Working on Part 1", "ts": 101},
    {"user": "Sudeep", "msg": "Finished Part 1", "ts": 105}
]

def process_stream_to_graph(event):
    # 1. Identify Entity
    print(f"MERGE (p:Person {{name: '{event['user']}'}})")
    
    # 2. Create Temporal Fact
    print(f"""
        MATCH (p:Person {{name: '{event['user']}'}})
        CREATE (p)-[:POSTED {{text: '{event['msg']}', time: {event['ts']}}}]->(:Event)
    """)
    
    # 3. Update Current State
    print(f"MATCH (p:Person {{name: '{event['user']}'}}) SET p.lastMsg = '{event['msg']}'")

for e in event_stream:
    process_stream_to_graph(e)
    # The 'lastMsg' on Sudeep will correctly end up as 'Finished Part 1'

---

6. Summary and Exercises

Live ingestion is about State Management.

• Webhooks provide the push.
• MERGE prevents duplication.
• Timestamps prevent out-of-order errors.
• Temporal Edges provide a history that the AI can "Search" through.

Exercises

1. Polling vs. Webhook: You are building a graph of "Stock Prices." Which ingestion method (Polling or Webhook) is more appropriate for a 1-second update cycle?
2. State Audit: If a user changes their name in Slack, your graph will have two nodes if you use "Name" as the ID. What "Slack Property" should you use as the permanent node ID to prevent this?
3. Latency Thinking: If your graph updates in 10ms but your AI agent's "Summary" is cached for 1 hour, is your system truly "Real-Time"?

In the next lesson, we will look at how to handle the inevitable errors in these streams: Cleansing and Conflict Resolution.