my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	import logging
#2	import os
#3	from typing import Any, Dict, List, Optional
#4
#5	from dotenv import load_dotenv
#6	from fastapi import FastAPI, HTTPException
#7	from fastapi.responses import JSONResponse, RedirectResponse
#8	from pydantic import BaseModel, Field
#9
#10	from mem0 import Memory
#11
#12	logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
#13
#14	# Load environment variables
#15	load_dotenv()
#16
#17
#18	POSTGRES_HOST = os.environ.get("POSTGRES_HOST", "postgres")
#19	POSTGRES_PORT = os.environ.get("POSTGRES_PORT", "5432")
#20	POSTGRES_DB = os.environ.get("POSTGRES_DB", "postgres")
#21	POSTGRES_USER = os.environ.get("POSTGRES_USER", "postgres")
#22	POSTGRES_PASSWORD = os.environ.get("POSTGRES_PASSWORD", "postgres")
#23	POSTGRES_COLLECTION_NAME = os.environ.get("POSTGRES_COLLECTION_NAME", "memories")
#24
#25	NEO4J_URI = os.environ.get("NEO4J_URI", "bolt://neo4j:7687")
#26	NEO4J_USERNAME = os.environ.get("NEO4J_USERNAME", "neo4j")
#27	NEO4J_PASSWORD = os.environ.get("NEO4J_PASSWORD", "mem0graph")
#28
#29	MEMGRAPH_URI = os.environ.get("MEMGRAPH_URI", "bolt://localhost:7687")
#30	MEMGRAPH_USERNAME = os.environ.get("MEMGRAPH_USERNAME", "memgraph")
#31	MEMGRAPH_PASSWORD = os.environ.get("MEMGRAPH_PASSWORD", "mem0graph")
#32
#33	OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY")
#34	HISTORY_DB_PATH = os.environ.get("HISTORY_DB_PATH", "/app/history/history.db")
#35
#36	DEFAULT_CONFIG = {
#37	"version": "v1.1",
#38	"vector_store": {
#39	"provider": "pgvector",
#40	"config": {
#41	"host": POSTGRES_HOST,
#42	"port": int(POSTGRES_PORT),
#43	"dbname": POSTGRES_DB,
#44	"user": POSTGRES_USER,
#45	"password": POSTGRES_PASSWORD,
#46	"collection_name": POSTGRES_COLLECTION_NAME,
#47	},
#48	},
#49	"graph_store": {
#50	"provider": "neo4j",
#51	"config": {"url": NEO4J_URI, "username": NEO4J_USERNAME, "password": NEO4J_PASSWORD},
#52	},
#53	"llm": {"provider": "openai", "config": {"api_key": OPENAI_API_KEY, "temperature": 0.2, "model": "gpt-4.1-nano-2025-04-14"}},
#54	"embedder": {"provider": "openai", "config": {"api_key": OPENAI_API_KEY, "model": "text-embedding-3-small"}},
#55	"history_db_path": HISTORY_DB_PATH,
#56	}
#57
#58
#59	MEMORY_INSTANCE = Memory.from_config(DEFAULT_CONFIG)
#60
#61	app = FastAPI(
#62	title="Mem0 REST APIs",
#63	description="A REST API for managing and searching memories for your AI Agents and Apps.",
#64	version="1.0.0",
#65	)
#66
#67
#68	class Message(BaseModel):
#69	role: str = Field(..., description="Role of the message (user or assistant).")
#70	content: str = Field(..., description="Message content.")
#71
#72
#73	class MemoryCreate(BaseModel):
#74	messages: List[Message] = Field(..., description="List of messages to store.")
#75	user_id: Optional[str] = None
#76	agent_id: Optional[str] = None
#77	run_id: Optional[str] = None
#78	metadata: Optional[Dict[str, Any]] = None
#79
#80
#81	class SearchRequest(BaseModel):
#82	query: str = Field(..., description="Search query.")
#83	user_id: Optional[str] = None
#84	run_id: Optional[str] = None
#85	agent_id: Optional[str] = None
#86	filters: Optional[Dict[str, Any]] = None
#87
#88
#89	@app.post("/configure", summary="Configure Mem0")
#90	def set_config(config: Dict[str, Any]):
#91	"""Set memory configuration."""
#92	global MEMORY_INSTANCE
#93	MEMORY_INSTANCE = Memory.from_config(config)
#94	return {"message": "Configuration set successfully"}
#95
#96
#97	@app.post("/memories", summary="Create memories")
#98	def add_memory(memory_create: MemoryCreate):
#99	"""Store new memories."""
#100	if not any([memory_create.user_id, memory_create.agent_id, memory_create.run_id]):
#101	raise HTTPException(status_code=400, detail="At least one identifier (user_id, agent_id, run_id) is required.")
#102
#103	params = {k: v for k, v in memory_create.model_dump().items() if v is not None and k != "messages"}
#104	try:
#105	response = MEMORY_INSTANCE.add(messages=[m.model_dump() for m in memory_create.messages], **params)
#106	return JSONResponse(content=response)
#107	except Exception as e:
#108	logging.exception("Error in add_memory:") # This will log the full traceback
#109	raise HTTPException(status_code=500, detail=str(e))
#110
#111
#112	@app.get("/memories", summary="Get memories")
#113	def get_all_memories(
#114	user_id: Optional[str] = None,
#115	run_id: Optional[str] = None,
#116	agent_id: Optional[str] = None,
#117	):
#118	"""Retrieve stored memories."""
#119	if not any([user_id, run_id, agent_id]):
#120	raise HTTPException(status_code=400, detail="At least one identifier is required.")
#121	try:
#122	params = {
#123	k: v for k, v in {"user_id": user_id, "run_id": run_id, "agent_id": agent_id}.items() if v is not None
#124	}
#125	return MEMORY_INSTANCE.get_all(**params)
#126	except Exception as e:
#127	logging.exception("Error in get_all_memories:")
#128	raise HTTPException(status_code=500, detail=str(e))
#129
#130
#131	@app.get("/memories/{memory_id}", summary="Get a memory")
#132	def get_memory(memory_id: str):
#133	"""Retrieve a specific memory by ID."""
#134	try:
#135	return MEMORY_INSTANCE.get(memory_id)
#136	except Exception as e:
#137	logging.exception("Error in get_memory:")
#138	raise HTTPException(status_code=500, detail=str(e))
#139
#140
#141	@app.post("/search", summary="Search memories")
#142	def search_memories(search_req: SearchRequest):
#143	"""Search for memories based on a query."""
#144	try:
#145	params = {k: v for k, v in search_req.model_dump().items() if v is not None and k != "query"}
#146	return MEMORY_INSTANCE.search(query=search_req.query, **params)
#147	except Exception as e:
#148	logging.exception("Error in search_memories:")
#149	raise HTTPException(status_code=500, detail=str(e))
#150
#151
#152	@app.put("/memories/{memory_id}", summary="Update a memory")
#153	def update_memory(memory_id: str, updated_memory: Dict[str, Any]):
#154	"""Update an existing memory with new content.
#155
#156	Args:
#157	memory_id (str): ID of the memory to update
#158	updated_memory (str): New content to update the memory with
#159
#160	Returns:
#161	dict: Success message indicating the memory was updated
#162	"""
#163	try:
#164	return MEMORY_INSTANCE.update(memory_id=memory_id, data=updated_memory)
#165	except Exception as e:
#166	logging.exception("Error in update_memory:")
#167	raise HTTPException(status_code=500, detail=str(e))
#168
#169
#170	@app.get("/memories/{memory_id}/history", summary="Get memory history")
#171	def memory_history(memory_id: str):
#172	"""Retrieve memory history."""
#173	try:
#174	return MEMORY_INSTANCE.history(memory_id=memory_id)
#175	except Exception as e:
#176	logging.exception("Error in memory_history:")
#177	raise HTTPException(status_code=500, detail=str(e))
#178
#179
#180	@app.delete("/memories/{memory_id}", summary="Delete a memory")
#181	def delete_memory(memory_id: str):
#182	"""Delete a specific memory by ID."""
#183	try:
#184	MEMORY_INSTANCE.delete(memory_id=memory_id)
#185	return {"message": "Memory deleted successfully"}
#186	except Exception as e:
#187	logging.exception("Error in delete_memory:")
#188	raise HTTPException(status_code=500, detail=str(e))
#189
#190
#191	@app.delete("/memories", summary="Delete all memories")
#192	def delete_all_memories(
#193	user_id: Optional[str] = None,
#194	run_id: Optional[str] = None,
#195	agent_id: Optional[str] = None,
#196	):
#197	"""Delete all memories for a given identifier."""
#198	if not any([user_id, run_id, agent_id]):
#199	raise HTTPException(status_code=400, detail="At least one identifier is required.")
#200	try:
#201	params = {
#202	k: v for k, v in {"user_id": user_id, "run_id": run_id, "agent_id": agent_id}.items() if v is not None
#203	}
#204	MEMORY_INSTANCE.delete_all(**params)
#205	return {"message": "All relevant memories deleted"}
#206	except Exception as e:
#207	logging.exception("Error in delete_all_memories:")
#208	raise HTTPException(status_code=500, detail=str(e))
#209
#210
#211	@app.post("/reset", summary="Reset all memories")
#212	def reset_memory():
#213	"""Completely reset stored memories."""
#214	try:
#215	MEMORY_INSTANCE.reset()
#216	return {"message": "All memories reset"}
#217	except Exception as e:
#218	logging.exception("Error in reset_memory:")
#219	raise HTTPException(status_code=500, detail=str(e))
#220
#221
#222	@app.get("/", summary="Redirect to the OpenAPI documentation", include_in_schema=False)
#223	def home():
#224	"""Redirect to the OpenAPI documentation."""
#225	return RedirectResponse(url="/docs")
#226

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public