Store memory using Letta's official agent API. Sends message directly to the Letta agent (following official design). The agent will internally decide what to store using archival_memory_insert. For long inputs, text is split into chunks and sent sequentially.
(self, text: str, **kwargs)
| 437 | return result |
| 438 | |
| 439 | def memorize(self, text: str, **kwargs) -> MemoryBuildResult: |
| 440 | """Store memory using Letta's official agent API. |
| 441 | |
| 442 | Sends message directly to the Letta agent (following official design). |
| 443 | The agent will internally decide what to store using archival_memory_insert. |
| 444 | |
| 445 | For long inputs, text is split into chunks and sent sequentially. |
| 446 | """ |
| 447 | context_id = self._get_context_id() |
| 448 | agent_id = self._ensure_agent(context_id) |
| 449 | |
| 450 | # Split input into chunks if needed |
| 451 | chunks = self._chunk_text_by_tokens( |
| 452 | text, |
| 453 | self.memorize_chunk_tokens, |
| 454 | self.memorize_chunk_overlap_tokens |
| 455 | ) |
| 456 | |
| 457 | if not chunks: |
| 458 | chunks = [text] if text else [] |
| 459 | |
| 460 | total_time_cost = 0.0 |
| 461 | all_memory_entries = [] |
| 462 | all_passages = [] |
| 463 | all_function_calls = [] |
| 464 | all_internal_monologue = [] |
| 465 | total_usage = {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0} |
| 466 | last_assistant_message = "" |
| 467 | |
| 468 | for chunk_idx, chunk in enumerate(chunks): |
| 469 | # Send chunk directly to agent (official Letta design - agent decides what to store) |
| 470 | start_time = time.time() |
| 471 | try: |
| 472 | response = self._client.user_message(agent_id=agent_id, message=chunk) |
| 473 | chunk_time = time.time() - start_time |
| 474 | total_time_cost += chunk_time |
| 475 | except Exception as e: |
| 476 | logger.error(f"Letta user_message failed for chunk {chunk_idx}: {e}") |
| 477 | return MemoryBuildResult( |
| 478 | success=False, |
| 479 | method="letta", |
| 480 | action="user_message", |
| 481 | input_content=text, |
| 482 | stored_content="", |
| 483 | extraction_result=f"[Error at chunk {chunk_idx}: {e}]", |
| 484 | all_passages=all_passages, |
| 485 | memory_entries=all_memory_entries, |
| 486 | chunk_count=chunk_idx, |
| 487 | time_cost=total_time_cost, |
| 488 | extra={"agent_id": agent_id, "error": str(e), "failed_chunk": chunk_idx}, |
| 489 | ) |
| 490 | |
| 491 | # Parse response |
| 492 | parsed = self._parse_letta_response(response) |
| 493 | |
| 494 | # Record token usage |
| 495 | self._record_usage_to_tracker(parsed, "memorize", chunk_time) |
| 496 |
nothing calls this directly
no test coverage detected