CogneeWriter

ChatMessage 객체를 CogneeMemoryStore에 장기 메모리로 쓰는 Writer예요.

출처: 문서

본문

항목	내용
파이프라인에서의 일반적인 위치	메모리 보강 파이프라인에서 `Agent` 또는 Chat Generator 이후
필수 init 변수	`memory_store` — `CogneeMemoryStore` 인스턴스
선택적 init 변수	`session_id` — 설정하면 세션-캐시 계층을 대상으로 쓰고, `None`이면 영구 knowledge graph에 씀
필수 run 변수	`messages` — `ChatMessage` 객체 리스트
선택적 run 변수	`user_id` — 쓰기 범위를 지정할 Cognee 사용자 ID. `None`을 넘기면 Cognee 기본 사용자 사용
출력 변수	`messages_written` — 쓰여진 `ChatMessage` 객체 리스트(변경 없이 그대로 통과)
API reference	Cognee
GitHub link	https://github.com/deepset-ai/haystack-core-integrations/tree/main/integrations/cognee
Package name	`cognee-haystack`

Overview

CogneeWriter는 ChatMessage 객체 리스트를 CogneeMemoryStore에 영속화합니다. Haystack 파이프라인에서 Agent 턴 후의 대화 사실이나 사용자 선호를 저장하는 데 사용해요.

메시지는 변경 없이 파이프라인 출력(messages_written)으로 전달되어, 파이프라인 흐름을 끊지 않고 Agent나 generator 뒤에 쉽게 연결할 수 있습니다.

session_id init 파라미터는 대상 Cognee 메모리 계층을 제어해요:

session_id를 생략하거나(None으로 설정) 영구 knowledge graph에 쓰기 — Cognee가 수집 중 LLM 추출을 실행해 풍부한 그래프 완성용 노드를 만들어요.
session_id를 설정해 세션 캐시에 쓰기 — LLM 추출 없이 그 세션에 한정된 빠른 쓰기입니다. 세션 내용은 나중에 CogneeMemoryStore.improve()로 영구 그래프로 승격할 수 있어요.

writer의 session_id가 호출별로 스토어의 session_id를 덮어쓰므로, 하나의 스토어가 서로 다른 메모리 계층을 대상으로 하는 여러 writer를 뒷받침할 수 있습니다.

Installation

Cognee 통합을 설치합니다:

pip install cognee-haystack

LLM API 키를 설정합니다(Cognee가 그래프 추출에 사용):

export LLM_API_KEY="your-llm-api-key"

선택적으로 별도의 임베딩 API 키를 설정할 수 있어요(설정하지 않으면 LLM_API_KEY를 사용):

export EMBEDDING_API_KEY="your-embedding-api-key"

Usage

On its own

from haystack.dataclasses import ChatMessage

from haystack_integrations.components.writers.cognee import CogneeWriter
from haystack_integrations.memory_stores.cognee import CogneeMemoryStore

store = CogneeMemoryStore()
writer = CogneeWriter(memory_store=store)

result = writer.run(
    messages=[ChatMessage.from_user("Alice prefers concise Python examples.")],
    user_id="a1b2c3d4-e5f6-7890-abcd-ef1234567890",
)
print(result["messages_written"])

영구 그래프 대신 세션 캐시에 쓰려면 session_id를 넘기세요:

session_writer = CogneeWriter(memory_store=store, session_id="alice_session_1")
session_writer.run(
    messages=[
        ChatMessage.from_user("Alice is currently debugging a vector store issue.")
    ],
    user_id="a1b2c3d4-e5f6-7890-abcd-ef1234567890",
)

In a Pipeline

이 예시는 Agent의 전체 messages 출력을 CogneeWriter에 연결해서, Cognee가 대화 턴을 영구 그래프에 저장하도록 합니다.

from haystack import Pipeline
from haystack.components.agents import Agent
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage

from haystack_integrations.components.writers.cognee import CogneeWriter
from haystack_integrations.memory_stores.cognee import CogneeMemoryStore

store = CogneeMemoryStore(dataset_name="my_agent_memory")

pipeline = Pipeline()
pipeline.add_component(
    "agent",
    Agent(
        chat_generator=OpenAIChatGenerator(model="gpt-4o-mini"),
        system_prompt=(
            "Answer the user and preserve durable user facts or preferences for future conversations."
        ),
    ),
)
pipeline.add_component("writer", CogneeWriter(memory_store=store))

pipeline.connect("agent.messages", "writer.messages")

result = pipeline.run(
    {
        "agent": {
            "messages": [
                ChatMessage.from_user(
                    "My name is Alice and I prefer concise Python examples.",
                ),
            ],
        },
        "writer": {
            "user_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
        },
    },
)

print(result["writer"]["messages_written"])

더 알아보기 (Learn more)

CogneeWriter — Haystack 공식 문서