def retrieve_memory_context(query: str, owner_id: str) -> List[SystemMessage]:
"""Search Cognis for relevant memories and format as LangChain messages."""
results = cog.search(query=query, owner_id=owner_id, limit=5)
if not results:
return []
formatted = "\n".join(f"- {r.content}" for r in results)
return [SystemMessage(content=f"Relevant memories about this student:\n{formatted}")]
def store_interaction(user_input: str, response: str, owner_id: str, session_id: str):
"""Persist the conversation turn in Cognis."""
cog.add(
messages=[
CognisMessage(role="user", content=user_input),
CognisMessage(role="assistant", content=response),
],
owner_id=owner_id,
session_id=session_id,
)
def chat(user_input: str, chat_history: list, owner_id: str, session_id: str) -> str:
# 1. Retrieve relevant memories
memory_msgs = retrieve_memory_context(user_input, owner_id)
# 2. Generate response via LCEL chain
result = chain.invoke({
"input": user_input,
"memory_context": memory_msgs,
"chat_history": chat_history,
})
# 3. Store interaction in Cognis
store_interaction(user_input, result.content, owner_id, session_id)
return result.content