Agno is a lightweight framework for building AI agents with tools and structured outputs. By wrapping Cognis methods as Agno tools, you can build agents that autonomously decide when to store, retrieve, and manage their own memories.What you’ll build: A personal assistant that uses Cognis-backed tools to remember user preferences, search past conversations, and manage its own memory — all driven by the agent’s own reasoning.Integration pattern: Tool-based — Cognis operations exposed as @tool decorators. The agent decides when and what to remember.
from agno.agent import Agentfrom agno.models.openai import OpenAIChatfrom agno.tools import toolfrom lyzr import Cognis, CognisMessagecog = Cognis()@tooldef search_memory(query: str) -> str: """Search user's memories for relevant information.""" results = cog.search(query=query, owner_id="user_123", limit=5) if not results: return "No relevant memories found." return "\n".join(f"- {r.content}" for r in results)@tooldef add_memory(user_message: str, assistant_response: str) -> str: """Store important information from the conversation.""" cog.add( messages=[ CognisMessage(role="user", content=user_message), CognisMessage(role="assistant", content=assistant_response), ], owner_id="user_123", ) return "Memory stored successfully."agent = Agent( model=OpenAIChat(id="gpt-4o"), tools=[search_memory, add_memory], instructions=["Store important user info with add_memory", "Search memories before answering personal questions"],)response = agent.run("My name is Sarah and I love hiking.")
from agno.agent import Agentfrom agno.models.openai import OpenAIChatfrom agno.tools import toolfrom cognis import Cognism = Cognis(owner_id="user_123")@tooldef search_memory(query: str) -> str: """Search user's memories for relevant information.""" resp = m.search(query, limit=5) if not resp["results"]: return "No relevant memories found." return "\n".join(f"- {r['content']}" for r in resp["results"])@tooldef add_memory(user_message: str, assistant_response: str) -> str: """Store important information from the conversation.""" m.add([ {"role": "user", "content": user_message}, {"role": "assistant", "content": assistant_response}, ]) return "Memory stored successfully."agent = Agent( model=OpenAIChat(id="gpt-4o"), tools=[search_memory, add_memory], instructions=["Store important user info with add_memory", "Search memories before answering personal questions"],)response = agent.run("My name is Sarah and I love hiking.")# Don't forget: m.close() when done
import osfrom agno.agent import Agentfrom agno.models.openai import OpenAIChatfrom agno.tools import toolfrom lyzr import Cognis, CognisMessagecog = Cognis(api_key=os.getenv("LYZR_API_KEY"))OWNER_ID = "user_123"AGENT_ID = "assistant_agno"@tooldef search_memory(query: str) -> str: """Search the user's memories for relevant information. Use this before answering personal questions or when the user references something from a past conversation.""" results = cog.search(query=query, owner_id=OWNER_ID, limit=5) if not results: return "No relevant memories found." formatted = "\n".join(f"- {r.content}" for r in results) return f"Found {len(results)} relevant memories:\n{formatted}"@tooldef add_memory(user_message: str, assistant_response: str) -> str: """Store important information from the conversation into long-term memory. Use this when the user shares personal preferences, important facts, or information they'd want remembered.""" cog.add( messages=[ CognisMessage(role="user", content=user_message), CognisMessage(role="assistant", content=assistant_response), ], owner_id=OWNER_ID, agent_id=AGENT_ID, ) return "Memory stored successfully."@tooldef get_all_memories() -> str: """Retrieve all stored memories for the current user. Use this when the user asks what you remember about them.""" memory_list = cog.get(owner_id=OWNER_ID, limit=20) if not memory_list.memories: return "No memories stored yet." formatted = "\n".join( f"- [{m.id[:8]}] {m.content}" for m in memory_list.memories ) return f"All memories ({len(memory_list.memories)}):\n{formatted}"@tooldef delete_memory(memory_id: str) -> str: """Delete a specific memory by its ID. Use this when the user asks to forget something.""" success = cog.delete(memory_id=memory_id, owner_id=OWNER_ID) if success: return f"Memory {memory_id} deleted successfully." return f"Failed to delete memory {memory_id}."
from agno.agent import Agentfrom agno.models.openai import OpenAIChatfrom agno.tools import toolfrom cognis import Cognism = Cognis(owner_id="user_123", agent_id="assistant_agno")@tooldef search_memory(query: str) -> str: """Search the user's memories for relevant information. Use this before answering personal questions or when the user references something from a past conversation.""" resp = m.search(query, limit=5) if not resp["results"]: return "No relevant memories found." formatted = "\n".join(f"- {r['content']}" for r in resp["results"]) return f"Found {resp['count']} relevant memories:\n{formatted}"@tooldef add_memory(user_message: str, assistant_response: str) -> str: """Store important information from the conversation into long-term memory. Use this when the user shares personal preferences, important facts, or information they'd want remembered.""" m.add([ {"role": "user", "content": user_message}, {"role": "assistant", "content": assistant_response}, ]) return "Memory stored successfully."@tooldef get_all_memories() -> str: """Retrieve all stored memories for the current user. Use this when the user asks what you remember about them.""" resp = m.get_all(limit=20) if not resp["memories"]: return "No memories stored yet." formatted = "\n".join( f"- [{mem['memory_id'][:8]}] {mem['content']}" for mem in resp["memories"] ) return f"All memories ({resp['total']}):\n{formatted}"@tooldef delete_memory(memory_id: str) -> str: """Delete a specific memory by its ID. Use this when the user asks to forget something.""" result = m.delete(memory_id) if result["success"]: return f"Memory {memory_id} deleted successfully." return f"Failed to delete memory {memory_id}."
agent = Agent( model=OpenAIChat(id="gpt-4o"), tools=[search_memory, add_memory, get_all_memories, delete_memory], instructions=[ "You are a helpful personal assistant with persistent memory.", "When the user shares important personal information (preferences, " "facts about themselves, plans), use add_memory to store it.", "Before answering personal questions or when the user references " "past conversations, use search_memory to check for relevant memories.", "If the user asks what you remember, use get_all_memories.", "If the user asks to forget something, use delete_memory.", ], show_tool_calls=True, markdown=True,)
conversations = [ "Hi! My name is Sarah and I'm a software engineer. I love hiking and Italian food.", "What's a good restaurant recommendation for me?", "I'm planning a trip to Colorado next month for hiking.", "What do you remember about me?", "Actually, I prefer Thai food now instead of Italian.",]for user_msg in conversations: print(f"User: {user_msg}") response = agent.run(user_msg) print(f"Assistant: {response.content}\n")
The agent will:
Turn 1: Call add_memory to store Sarah’s name, job, and preferences
Turn 2: Call search_memory for food preferences, then recommend Italian restaurants
Turn 3: Call add_memory to store the Colorado trip plan
Turn 4: Call get_all_memories to show everything it remembers
Turn 5: Call add_memory with the updated food preference
Cross-session search (cross_session=True) is a hosted-only feature. Open-source Cognis searches the global (owner_id, agent_id) scope by default, which already spans sessions.
@tooldef search_all_sessions(query: str) -> str: """Search across all conversation sessions.""" results = cog.search(query=query, owner_id=OWNER_ID, cross_session=True, limit=10) if not results: return "No memories found across any session." return "\n".join(f"- {r.content}" for r in results)
Summaries are a hosted-only feature (store_summary, get_current_summary, search_summaries).
@tooldef get_session_summary(session_id: str) -> str: """Get a summary of a past conversation session.""" result = cog.get_current_summary(owner_id=OWNER_ID, session_id=session_id) return result.get("content", "No summary available for this session.")
