메모리

from crewai import Memory

memory = Memory()

# 저장 -- LLM이 scope, categories, importance를 추론
memory.remember("We decided to use PostgreSQL for the user database.")

# 검색 -- 복합 점수(의미 + 최신성 + 중요도)로 결과 순위 매기기
matches = memory.recall("What database did we choose?")
for m in matches:
    print(f"[{m.score:.2f}] {m.record.content}")

# 빠르게 변하는 프로젝트를 위한 점수 조정
memory = Memory(recency_weight=0.5, recency_half_life_days=7)

# 삭제
memory.forget(scope="/project/old")

# 자동 구성된 scope 트리 탐색
print(memory.tree())
print(memory.info("/"))

from crewai import Memory

memory = Memory()

# 지식 구축
memory.remember("The API rate limit is 1000 requests per minute.")
memory.remember("Our staging environment uses port 8080.")
memory.remember("The team agreed to use feature flags for all new releases.")

# 나중에 필요한 것을 recall
matches = memory.recall("What are our API limits?", limit=5)
for m in matches:
    print(f"[{m.score:.2f}] {m.record.content}")

# 긴 텍스트에서 원자적 사실 추출
raw = """Meeting notes: We decided to migrate from MySQL to PostgreSQL
next quarter. The budget is $50k. Sarah will lead the migration."""

facts = memory.extract_memories(raw)
# ["Migration from MySQL to PostgreSQL planned for next quarter",
#  "Database migration budget is $50k",
#  "Sarah will lead the database migration"]

for fact in facts:
    memory.remember(fact)

from crewai import Crew, Agent, Task, Process, Memory

# 옵션 1: 기본 메모리
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential,
    memory=True,
    verbose=True,
)

# 옵션 2: 조정된 점수가 있는 사용자 정의 메모리
memory = Memory(
    recency_weight=0.4,
    semantic_weight=0.4,
    importance_weight=0.2,
    recency_half_life_days=14,
)
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    memory=memory,
)

from crewai import Agent, Memory

memory = Memory()

# 연구원은 비공개 scope를 받음 -- /agent/researcher만 볼 수 있음
researcher = Agent(
    role="Researcher",
    goal="Find and analyze information",
    backstory="Expert researcher with attention to detail",
    memory=memory.scope("/agent/researcher"),
)

# 작성자는 crew 공유 메모리 사용 (에이전트 수준 메모리 미설정)
writer = Agent(
    role="Writer",
    goal="Produce clear, well-structured content",
    backstory="Experienced technical writer",
    # memory 미설정 -- crew에 메모리가 활성화되면 crew._memory 사용
)

from crewai.flow.flow import Flow, listen, start

class ResearchFlow(Flow):
    @start()
    def gather_data(self):
        findings = "PostgreSQL handles 10k concurrent connections. MySQL caps at 5k."
        self.remember(findings, scope="/research/databases")
        return findings

    @listen(gather_data)
    def write_report(self, findings):
        # 컨텍스트를 제공하기 위해 과거 연구 recall
        past = self.recall("database performance benchmarks")
        context = "\n".join(f"- {m.record.content}" for m in past)
        return f"Report:\nNew findings: {findings}\nPrevious context:\n{context}"

/
  /company
    /company/engineering
    /company/product
  /project
    /project/alpha
    /project/beta
  /agent
    /agent/researcher
    /agent/writer

memory = Memory()

# LLM이 콘텐츠에서 scope 추론
memory.remember("We chose PostgreSQL for the user database.")
# -> /project/decisions 또는 /engineering/database 아래에 배치될 수 있음

# scope를 명시적으로 지정할 수도 있음
memory.remember("Sprint velocity is 42 points", scope="/team/metrics")

print(memory.tree())
# / (15 records)
#   /project (8 records)
#     /project/alpha (5 records)
#     /project/beta (3 records)
#   /agent (7 records)
#     /agent/researcher (4 records)
#     /agent/writer (3 records)

print(memory.info("/project/alpha"))
# ScopeInfo(path='/project/alpha', record_count=5,
#           categories=['architecture', 'database'],
#           oldest_record=datetime(...), newest_record=datetime(...),
#           child_scopes=[])

memory = Memory()

# 특정 에이전트를 위한 scope 생성
agent_memory = memory.scope("/agent/researcher")

# 모든 것이 /agent/researcher 기준으로 상대적
agent_memory.remember("Found three relevant papers on LLM memory.")
# -> /agent/researcher 아래에 저장

agent_memory.recall("relevant papers")
# -> /agent/researcher 아래에서만 검색

# subscope로 더 좁히기
project_memory = agent_memory.subscope("project-alpha")
# -> /agent/researcher/project-alpha

memory = Memory()
# 각 프로젝트가 자체 분기를 가짐
memory.remember("Using microservices architecture", scope="/project/alpha/architecture")
memory.remember("GraphQL API for client apps", scope="/project/beta/api")

# 모든 프로젝트에서 recall
memory.recall("API design decisions")

# 특정 프로젝트 내에서만
memory.recall("API design", scope="/project/beta")

memory = Memory()

# 연구원은 비공개 발견을 가짐
researcher_memory = memory.scope("/agent/researcher")

# 작성자는 자체 scope와 공유 회사 지식에서 읽을 수 있음
writer_view = memory.slice(
    scopes=["/agent/writer", "/company/knowledge"],
    read_only=True,
)

memory = Memory()

# 각 고객이 격리된 컨텍스트를 가짐
memory.remember("Prefers email communication", scope="/customer/acme-corp")
memory.remember("On enterprise plan, 50 seats", scope="/customer/acme-corp")

# 공유 제품 문서는 모든 에이전트가 접근 가능
memory.remember("Rate limit is 1000 req/min on enterprise plan", scope="/product/docs")

memory = Memory()

# 에이전트는 자체 scope와 회사 지식에서 recall 가능,
# 하지만 회사 지식에 쓸 수 없음
agent_view = memory.slice(
    scopes=["/agent/researcher", "/company/knowledge"],
    read_only=True,
)

matches = agent_view.recall("company security policies", limit=5)
# /agent/researcher와 /company/knowledge 모두에서 검색, 결과 병합 및 순위 매기기

agent_view.remember("new finding")  # PermissionError 발생 (읽기 전용)

view = memory.slice(scopes=["/team/alpha", "/team/beta"], read_only=False)

# 쓸 때 scope를 반드시 지정
view.remember("Cross-team decision", scope="/team/alpha", categories=["decisions"])

composite = semantic_weight * similarity + recency_weight * decay + importance_weight * importance

# 스프린트 회고: 최근 메모리 선호, 짧은 반감기
memory = Memory(
    recency_weight=0.5,
    semantic_weight=0.3,
    importance_weight=0.2,
    recency_half_life_days=7,
)

# 아키텍처 지식 베이스: 중요한 메모리 선호, 긴 반감기
memory = Memory(
    recency_weight=0.1,
    semantic_weight=0.5,
    importance_weight=0.4,
    recency_half_life_days=180,
)

# 2개의 레코드만 저장됨 (세 번째는 첫 번째의 거의 중복)
memory.remember_many([
    "CrewAI supports complex workflows.",
    "Python is a great language.",
    "CrewAI supports complex workflows.",  # 배치 내 중복 제거로 삭제
])

# 즉시 반환 -- 저장은 백그라운드에서 발생
memory.remember_many(["Fact A.", "Fact B.", "Fact C."])

# recall()은 검색 전에 보류 중인 저장을 자동으로 대기
matches = memory.recall("facts")  # 3개 레코드 모두 확인 가능

memory = Memory()
memory.remember_many(["Fact A.", "Fact B."])

# 옵션 1: 보류 중인 저장 대기
memory.drain_writes()

# 옵션 2: 드레인 후 백그라운드 풀 종료
memory.close()

# 메모리에 출처 태그 지정
memory.remember("User prefers dark mode", source="user:alice")
memory.remember("System config updated", source="admin")
memory.remember("Agent found a bug", source="agent:debugger")

# 특정 출처의 메모리만 recall
matches = memory.recall("user preferences", source="user:alice")

# 비공개 메모리 저장
memory.remember("Alice's API key is sk-...", source="user:alice", private=True)

# 이 recall은 비공개 메모리를 볼 수 있음 (source 일치)
matches = memory.recall("API key", source="user:alice")

# 이 recall은 볼 수 없음 (다른 source)
matches = memory.recall("API key", source="user:bob")

# 관리자 액세스: source에 관계없이 모든 비공개 레코드 보기
matches = memory.recall("API key", include_private=True)

# Shallow: 순수 벡터 검색, LLM 없음
matches = memory.recall("What did we decide?", limit=10, depth="shallow")

# Deep (기본값): 긴 쿼리에 대한 LLM 분석을 포함한 지능형 검색
matches = memory.recall(
    "Summarize all architecture decisions from this quarter",
    limit=10,
    depth="deep",
)

memory = Memory(
    confidence_threshold_high=0.9,   # 매우 확신할 때만 합성
    confidence_threshold_low=0.4,    # 더 적극적으로 깊이 탐색
    exploration_budget=2,            # 최대 2라운드 탐색 허용
    query_analysis_threshold=200,    # 이보다 짧은 쿼리는 LLM 건너뛰기
)

from crewai import Memory

# 설정 dict로
memory = Memory(embedder={"provider": "openai", "config": {"model_name": "text-embedding-3-small"}})

# 사전 구축된 callable로
from crewai.rag.embeddings.factory import build_embedder
embedder = build_embedder({"provider": "ollama", "config": {"model_name": "mxbai-embed-large"}})
memory = Memory(embedder=embedder)

from crewai import Crew

crew = Crew(
    agents=[...],
    tasks=[...],
    memory=True,
    embedder={"provider": "openai", "config": {"model_name": "text-embedding-3-small"}},
)

from crewai import Memory, LLM

# 기본값: gpt-4o-mini
memory = Memory()

# 다른 OpenAI 모델 사용
memory = Memory(llm="gpt-4o")

# Anthropic 사용
memory = Memory(llm="anthropic/claude-3-haiku-20240307")

# 완전한 로컬/비공개 분석을 위해 Ollama 사용
memory = Memory(llm="ollama/llama3.2")

# Google Gemini 사용
memory = Memory(llm="gemini/gemini-2.0-flash")

# 사용자 정의 설정이 있는 사전 구성된 LLM 인스턴스 전달
llm = LLM(model="gpt-4o", temperature=0)
memory = Memory(llm=llm)

memory = Memory(
    llm="ollama/llama3.2",
    embedder={"provider": "ollama", "config": {"model_name": "mxbai-embed-large"}},
)

memory.tree()                        # scope 및 레코드 수의 포맷된 트리
memory.tree("/project", max_depth=2) # 하위 트리 뷰
memory.info("/project")              # ScopeInfo: record_count, categories, oldest/newest
memory.list_scopes("/")              # 직계 자식 scope
memory.list_categories()             # 카테고리 이름 및 개수
memory.list_records(scope="/project/alpha", limit=20)  # scope의 레코드, 최신순

from crewai.events import BaseEventListener, MemoryQueryCompletedEvent

class MemoryMonitor(BaseEventListener):
    def setup_listeners(self, crewai_event_bus):
        @crewai_event_bus.on(MemoryQueryCompletedEvent)
        def on_done(source, event):
            if getattr(event, "source_type", None) == "unified_memory":
                print(f"Query '{event.query}' completed in {event.query_time_ms:.0f}ms")

crewai memory                              # TUI 브라우저 열기
crewai memory --storage-path ./my_memory   # 특정 디렉터리 지정

crew.reset_memories(command_type="memory")  # 통합 메모리 초기화
# 또는 Memory 인스턴스에서:
memory.reset()                    # 모든 scope
memory.reset(scope="/project/old")  # 해당 하위 트리만