making some code improvements

Author: Aayush Kataria

.env.template

@@ -24,10 +24,10 @@ COSMOS_DB_LEASE_CONTAINER=leases
 COSMOS_DB_THROUGHPUT_MODE=serverless
 COSMOS_DB_AUTOSCALE_MAX_RU=1000
 
-# ---- Change Feed Thresholds (set to 0 to disable) ----
-THREAD_SUMMARY_EVERY_N=0
-FACT_EXTRACTION_EVERY_N=0
-USER_SUMMARY_EVERY_N=0
+# ---- Processing thresholds (set to 0 to disable) ----
+THREAD_SUMMARY_EVERY_N=10
+FACT_EXTRACTION_EVERY_N=1
+USER_SUMMARY_EVERY_N=20
 
 # ---- AI Foundry / Azure OpenAI ----
 AI_FOUNDRY_ENDPOINT=https://<your-account>.openai.azure.com/

README.md

@@ -137,7 +137,7 @@ memory.add_cosmos(user_id=USER, thread_id=THREAD, role="user", content="I love C
 memory.add_cosmos(user_id=USER, thread_id=THREAD, role="assistant", content="It is fantastic.")
 
 # Run the processing pipeline (thread summary + fact extraction + user summary)
-memory.flush(user_id=USER, thread_id=THREAD)
+memory.process_now(user_id=USER, thread_id=THREAD)
 
 # Search semantically across the stored memory
 hits = memory.search_cosmos(user_id=USER, query_text="Cosmos DB preferences", top=5)
@@ -169,12 +169,45 @@ See [`Samples/`](Samples/) for end-to-end scenarios (chat memory, RAG, multi-age
 |---|---|---|
 | **Turn** | One message (user or assistant) — the raw conversation atom | `add_cosmos(...)`, `add_local(...)` |
 | **Thread summary** | LLM-generated, incrementally updated rollup of a single thread | `generate_thread_summary(...)` |
-| **Fact** | Discrete, independently searchable assertion extracted from turns | `extract_facts(...)` |
+| **Fact** | Discrete, independently searchable assertion extracted from turns | `extract_memories(...)` |
+| **Procedural** | Behavioral rule / instruction the user wants followed | `extract_memories(...)` |
+| **Episodic** | Past situation → action → outcome experience (90-day TTL) | `extract_memories(...)` |
 | **User summary** | Cross-thread profile of what's known about a user | `generate_user_summary(...)`, `get_user_summary(...)` |
-| **Search** | Vector + full-text + filter; returns turns, summaries, and facts | `search_cosmos(...)` |
-| **Flush** | Run the full pipeline (summary → facts → user profile) for recent turns | `flush(...)`, `flush_and_wait(...)` |
+| **Search** | Vector + full-text + filter; returns any of the above | `search_cosmos(...)` |
+| **Process now** | Run the full pipeline (summary → facts → user profile) for recent turns | `process_now(...)`, `process_now_and_wait(...)` |
 
-All four memory kinds live in the same Cosmos container, partitioned by `(user_id, thread_id)`, distinguished by a `memory_type` discriminator.
+All memory kinds live in the same Cosmos container, partitioned by `(user_id, thread_id)`, distinguished by a `type` discriminator.
+
+### Memory Type Taxonomy
+
+The `extract_memories` pipeline classifies each item it pulls from the conversation into one of four buckets. Every memory carries a top-level `confidence` (0.0–1.0) so retrieval can suppress weakly-grounded extractions.
+
+| Bucket | Meaning | Storage type | TTL |
+|---|---|---|---|
+| Fact | Declarative knowledge ("user prefers dark mode") | `type="fact"` | none |
+| Procedural | Behavioral rule ("always confirm before deleting") | `type="procedural"` | none |
+| Episodic | Past experience: situation → action → outcome | `type="episodic"` | 90 days |
+| Unclassified | Item worth keeping but the LLM couldn't confidently classify | `type="fact"` + tag `sys:unclassified` | none |
+
+#### Confidence Scale
+
+| Range | Meaning |
+|---|---|
+| 0.9–1.0 | Directly stated and unambiguous |
+| 0.7–0.9 | Clearly implied, no contradicting evidence |
+| 0.5–0.7 | Inferred from context — plausible but not explicit |
+| < 0.5 | Should be in `unclassified` instead |
+
+Filter at retrieval time:
+
+```python
+results = memory.search_cosmos("user preferences", user_id="u1", min_confidence=0.7)
+high_conf_facts = memory.get_memories(user_id="u1", memory_type="fact", min_confidence=0.7)
+```
+
+### Auto-trigger (per-turn extraction)
+
+By default, the **InProcess processor** runs fact extraction after **every turn** (`FACT_EXTRACTION_EVERY_N=1`) and a thread summary every **10 turns** (`THREAD_SUMMARY_EVERY_N=10`). The trigger fires inside `push_to_cosmos()` after the turn is durably written, so calling `process_now()` is normally redundant — it remains as an explicit "flush now" hook. The Durable backend uses the same defaults via the change-feed function app.
 
 ---
 
@@ -186,8 +219,8 @@ Pick at construction time via the `processor=` kwarg.
 |---|---|---|
 | Infra | None — just `pip install` | Sibling Azure Function app |
 | Best for | Prototypes, low TPS, single-agent | Fleet / multi-agent / high TPS |
-| `flush()` | Synchronous, returns when done | No-op (work runs async on change feed) |
-| `flush_and_wait()` | Returns immediately after flush | Polls until summary visible (RU-costly; tests/demos) |
+| `process_now()` | Synchronous, returns when done | No-op (work runs async on change feed) |
+| `process_now_and_wait()` | Returns immediately after flush | Polls until summary visible (RU-costly; tests/demos) |
 
 ```python
 from agent_memory_toolkit import CosmosMemoryClient, DurableFunctionProcessor
@@ -208,8 +241,8 @@ memory = CosmosMemoryClient(..., processor=DurableFunctionProcessor())
 | `MemoryProcessor` | `agent_memory_toolkit` | Protocol that any processor backend implements |
 | `InProcessProcessor` | `agent_memory_toolkit` | Default backend — runs the pipeline in-process |
 | `DurableFunctionProcessor` | `agent_memory_toolkit` | Marker backend — work runs in sibling Function app via change feed |
-| `client.flush()` | — | Run the pipeline for recent turns (in-process) or no-op (remote) |
-| `client.flush_and_wait()` | — | Opt-in poll until processing completes; useful for tests/demos with the remote backend |
+| `client.process_now()` | — | Run the pipeline for recent turns (in-process) or no-op (remote) |
+| `client.process_now_and_wait()` | — | Opt-in poll until processing completes; useful for tests/demos with the remote backend |
 | `MemoryRecord`, `MemoryType`, `Role` | `agent_memory_toolkit` | Pydantic models / enums |
 
 Async equivalents (`AsyncInProcessProcessor`, `AsyncDurableFunctionProcessor`) live in `agent_memory_toolkit.aio`.
@@ -243,6 +276,6 @@ tests/                  Unit + integration tests (pytest)
 
 ## Migration notes
 
-- **`agent_memory_toolkit.processing.ProcessingClient` is removed.** Drop the import and call `client.flush()` (or `client.flush_and_wait()`) instead. Same for the async `AsyncProcessingClient`.
+- **`agent_memory_toolkit.processing.ProcessingClient` is removed.** Drop the import and call `client.process_now()` (or `client.process_now_and_wait()`) instead. Same for the async `AsyncProcessingClient`.
 - **New `processor=` kwarg.** Defaults to `InProcessProcessor()` — existing code keeps its current behavior with no edits.
 - **`adf_endpoint` / `adf_key` constructor kwargs are gone.** The SDK no longer makes HTTP calls to the Function app at runtime; the Function app reads from the Cosmos change feed.

Samples/Demo_function_app.ipynb

@@ -32,8 +32,8 @@
     "\n",
     "* `add_cosmos(..., memory_type=\"turn\")` → writes the raw turn to Cosmos.\n",
     "* The change-feed-triggered Function App reads new turns and runs orchestrators.\n",
-    "* `flush()` is a **debug-logged no-op** — the Function App owns processing.\n",
-    "* `flush_and_wait()` polls Cosmos for the summary doc; useful for demos / tests (RU-costly)."
+    "* `process_now()` is a **debug-logged no-op** — the Function App owns processing.\n",
+    "* `process_now_and_wait()` polls Cosmos for the summary doc; useful for demos / tests (RU-costly)."
    ]
   },
   {
@@ -215,7 +215,7 @@
    "source": [
     "## 4. Verify the SDK did NOT process locally\n",
     "\n",
-    "`flush()` is a debug-logged no-op when using `DurableFunctionProcessor`. The Function App's change-feed\n",
+    "`process_now()` is a debug-logged no-op when using `DurableFunctionProcessor`. The Function App's change-feed\n",
     "trigger fires asynchronously — usually within a second or two."
    ]
   },
@@ -236,14 +236,14 @@
      "name": "stdout",
      "output_type": "stream",
      "text": [
-      "flush() returned — no LLM call was made by the SDK.\n"
+      "process_now() returned — no LLM call was made by the SDK.\n"
      ]
     }
    ],
    "source": [
     "# No-op: the function app owns processing.\n",
-    "memory.flush(user_id=USER_ID, thread_id=THREAD_ID)\n",
-    "print(\"flush() returned — no LLM call was made by the SDK.\")"
+    "memory.process_now(user_id=USER_ID, thread_id=THREAD_ID)\n",
+    "print(\"process_now() returned — no LLM call was made by the SDK.\")"
    ]
   },
   {
@@ -253,7 +253,7 @@
    "source": [
     "## 5. Wait for the Function App to produce a summary\n",
     "\n",
-    "`flush_and_wait` polls Cosmos for the thread's summary doc until `timeout` seconds. This is **RU-costly**\n",
+    "`process_now_and_wait` polls Cosmos for the thread's summary doc until `timeout` seconds. This is **RU-costly**\n",
     "(repeated `get_memories(memory_type='summary')` queries) — only use it for demos and tests.\n",
     "\n",
     "> If this returns `False`, check that:\n",
@@ -453,7 +453,7 @@
    ],
    "source": [
     "print(\"Polling Cosmos for the auto-generated summary…\")\n",
-    "ok = memory.flush_and_wait(user_id=USER_ID, thread_id=THREAD_ID, timeout=120.0)\n",
+    "ok = memory.process_now_and_wait(user_id=USER_ID, thread_id=THREAD_ID, timeout=120.0)\n",
     "print(f\"Summary available: {ok}\")"
    ]
   },

Samples/Demo_function_app_async.ipynb

@@ -186,7 +186,7 @@
    "id": "99f63822",
    "metadata": {},
    "source": [
-    "## 4. `flush()` is a no-op locally"
+    "## 4. `process_now()` is a no-op locally"
    ]
   },
   {
@@ -206,13 +206,13 @@
      "name": "stdout",
      "output_type": "stream",
      "text": [
-      "flush() returned — no LLM call was made by the SDK.\n"
+      "process_now() returned — no LLM call was made by the SDK.\n"
      ]
     }
    ],
    "source": [
-    "await memory.flush(user_id=USER_ID, thread_id=THREAD_ID)\n",
-    "print(\"flush() returned — no LLM call was made by the SDK.\")"
+    "await memory.process_now(user_id=USER_ID, thread_id=THREAD_ID)\n",
+    "print(\"process_now() returned — no LLM call was made by the SDK.\")"
    ]
   },
   {
@@ -407,7 +407,7 @@
    ],
    "source": [
     "print(\"Polling Cosmos for the auto-generated summary…\")\n",
-    "ok = await memory.flush_and_wait(user_id=USER_ID, thread_id=THREAD_ID, timeout=120.0)\n",
+    "ok = await memory.process_now_and_wait(user_id=USER_ID, thread_id=THREAD_ID, timeout=120.0)\n",
     "print(f\"Summary available: {ok}\")"
    ]
   },

Samples/scenario_counter_tuning.py

@@ -17,8 +17,8 @@
 +--------------------------+---------+------------------------------------+
 
 This sample writes a transcript long enough to cross the thread-level
-thresholds and uses :meth:`flush_and_wait` to *poll* Cosmos until the
-function app has produced a summary. ``flush_and_wait`` is intended for
+thresholds and uses :meth:`process_now_and_wait` to *poll* Cosmos until the
+function app has produced a summary. ``process_now_and_wait`` is intended for
 demos/tests; production code should not poll Cosmos for orchestration
 state.
 
@@ -80,14 +80,14 @@ def main() -> None:
         )
 
     # ------------------------------------------------------------------
-    # In durable mode .flush() is a no-op locally — the function app's
+    # In durable mode .process_now() is a no-op locally — the function app's
     # change-feed trigger has already begun work asynchronously. Use
-    # flush_and_wait() to *poll* for the summary doc; it's RU-costly so
+    # process_now_and_wait() to *poll* for the summary doc; it's RU-costly so
     # use it only in demos/tests.
     # ------------------------------------------------------------------
     print("\nWaiting (poll) for thread summary to be written by function app...")
     t0 = time.monotonic()
-    ok = client.flush_and_wait(
+    ok = client.process_now_and_wait(
         user_id=user_id,
         thread_id=thread_id,
         timeout=90.0,

Samples/scenario_remote_processor.py

@@ -14,9 +14,9 @@
 
 Behavior notes
 --------------
-* :meth:`CosmosMemoryClient.flush` is a debug-logged no-op when using the
+* :meth:`CosmosMemoryClient.process_now` is a debug-logged no-op when using the
   durable processor — the function app owns processing.
-* :meth:`CosmosMemoryClient.flush_and_wait` *polls* Cosmos for the resulting
+* :meth:`CosmosMemoryClient.process_now_and_wait` *polls* Cosmos for the resulting
   summary; that costs RUs so it's opt-in and intended for demos / tests.
 """
 from __future__ import annotations
@@ -59,12 +59,12 @@ def main() -> None:
         print(f"  wrote {role:>9}: {content}")
 
     # No-op locally; the change-feed trigger in the function app does the work.
-    client.flush(user_id=user_id, thread_id=thread_id)
+    client.process_now(user_id=user_id, thread_id=thread_id)
 
     # Optional: block until the function app has produced a summary for this
     # thread. Polls Cosmos every 0.5s until ``timeout`` — RU-costly, demo only.
     print("\nWaiting for the function app to produce a summary...")
-    ok = client.flush_and_wait(user_id=user_id, thread_id=thread_id, timeout=60.0)
+    ok = client.process_now_and_wait(user_id=user_id, thread_id=thread_id, timeout=60.0)
     print(f"Summary available: {ok}")
 
 

Samples/scenario_remote_processor_async.py

@@ -51,11 +51,11 @@ async def main() -> None:
             print(f"  wrote {role:>9}: {content}")
 
         # No-op locally; the change-feed trigger in the function app does the work.
-        await client.flush(user_id=user_id, thread_id=thread_id)
+        await client.process_now(user_id=user_id, thread_id=thread_id)
 
         # Optional: poll until the function app produces a summary. RU-costly.
         print("\nWaiting for the function app to produce a summary...")
-        ok = await client.flush_and_wait(
+        ok = await client.process_now_and_wait(
             user_id=user_id, thread_id=thread_id, timeout=60.0
         )
         print(f"Summary available: {ok}")