mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
modify knowledge writer
This commit is contained in:
@@ -115,14 +115,22 @@ async def add_observations(
|
||||
|
||||
|
||||
@router.get("/entities/{path_id:path}", response_model=EntityResponse)
|
||||
async def get_entity(path_id: PathId, knowledge_service: KnowledgeServiceDep) -> EntityResponse:
|
||||
"""Get a specific entity by ID."""
|
||||
async def get_entity(
|
||||
knowledge_service: KnowledgeServiceDep,
|
||||
path_id: PathId,
|
||||
content: bool = False, # New parameter
|
||||
) -> EntityResponse:
|
||||
"""Get a specific entity by ID.
|
||||
|
||||
Args:
|
||||
path_id: Entity path ID
|
||||
content: If True, include full file content
|
||||
"""
|
||||
try:
|
||||
entity = await knowledge_service.get_entity_by_path_id(path_id)
|
||||
entity_response = EntityResponse.model_validate(entity)
|
||||
|
||||
# if the entity is a note, we add the content via reading from the file
|
||||
if entity_response.entity_type == "note":
|
||||
if content: # Load content if requested
|
||||
content = await knowledge_service.read_entity_content(entity)
|
||||
entity_response.content = content
|
||||
|
||||
@@ -130,7 +138,6 @@ async def get_entity(path_id: PathId, knowledge_service: KnowledgeServiceDep) ->
|
||||
except EntityNotFoundError:
|
||||
raise HTTPException(status_code=404, detail=f"Entity with {path_id} not found")
|
||||
|
||||
|
||||
@router.post("/nodes", response_model=EntityListResponse)
|
||||
async def open_nodes(
|
||||
data: OpenNodesRequest, entity_service: EntityServiceDep
|
||||
|
||||
@@ -4,7 +4,13 @@ from basic_memory.models import Entity as EntityModel
|
||||
|
||||
|
||||
class KnowledgeWriter:
|
||||
"""Formats entities into markdown files."""
|
||||
"""Formats entities into markdown files.
|
||||
|
||||
Content handling:
|
||||
1. If raw content is provided, use it directly
|
||||
2. If structured data exists (observations/relations), generate structured content
|
||||
3. If neither, create basic content from name/summary
|
||||
"""
|
||||
|
||||
async def format_frontmatter(self, entity: EntityModel) -> dict:
|
||||
"""Generate frontmatter metadata for entity."""
|
||||
@@ -18,43 +24,64 @@ class KnowledgeWriter:
|
||||
frontmatter.update(entity.entity_metadata)
|
||||
return frontmatter
|
||||
|
||||
async def format_content(self, entity: EntityModel, content: str) -> str:
|
||||
"""Format entity content as markdown."""
|
||||
sections = [
|
||||
f"# {entity.name}\n",
|
||||
"", # Empty line after name
|
||||
]
|
||||
async def format_content(self, entity: EntityModel, content: str = None) -> str:
|
||||
"""Format entity content as markdown.
|
||||
|
||||
Args:
|
||||
entity: Entity to format
|
||||
content: Optional raw content to use instead of generating structured content
|
||||
|
||||
Returns:
|
||||
Formatted markdown content
|
||||
"""
|
||||
# If raw content provided, use it directly
|
||||
if content is not None:
|
||||
return content
|
||||
|
||||
if entity.summary:
|
||||
sections.extend([entity.summary, ""])
|
||||
# Otherwise, build structured content from entity data
|
||||
sections = []
|
||||
|
||||
# Only add entity name as title if we don't have structured content
|
||||
# This prevents duplicate titles when raw content already has a title
|
||||
if not (entity.observations or entity.outgoing_relations):
|
||||
sections.extend([
|
||||
f"# {entity.name}",
|
||||
"", # Empty line after title
|
||||
])
|
||||
|
||||
if entity.summary:
|
||||
sections.extend([entity.summary, ""])
|
||||
|
||||
# Add observations if present
|
||||
if entity.observations:
|
||||
sections.extend(
|
||||
[
|
||||
"## Observations",
|
||||
"<!-- Format: - [category] Content text #tag1 #tag2 (optional context) -->",
|
||||
"", # Empty line after format comment
|
||||
*[
|
||||
f"- [{obs.category}] {obs.content}"
|
||||
+ (f" ({obs.context})" if obs.context else "")
|
||||
for obs in entity.observations
|
||||
],
|
||||
"",
|
||||
]
|
||||
)
|
||||
sections.extend([
|
||||
"## Observations",
|
||||
"<!-- Format: - [category] Content text #tag1 #tag2 (optional context) -->",
|
||||
"", # Empty line after format comment
|
||||
])
|
||||
|
||||
for obs in entity.observations:
|
||||
line = f"- [{obs.category}] {obs.content}"
|
||||
if obs.context:
|
||||
line += f" ({obs.context})"
|
||||
sections.append(line)
|
||||
sections.append("") # Empty line after observations
|
||||
|
||||
# only outgoing relations are included in entity file
|
||||
# Add relations if present
|
||||
if entity.outgoing_relations:
|
||||
sections.extend(
|
||||
[
|
||||
"## Relations",
|
||||
"<!-- Format: - relation_type [[Entity]] (context) -->"
|
||||
"", # Empty line after format comment
|
||||
]
|
||||
)
|
||||
|
||||
# Outgoing relations (entity is "from")
|
||||
sections.extend([
|
||||
"## Relations",
|
||||
"<!-- Format: - relation_type [[Entity]] (context) -->",
|
||||
"", # Empty line after format comment
|
||||
])
|
||||
|
||||
for rel in entity.outgoing_relations:
|
||||
sections.append(f"- {rel.relation_type} [[{rel.to_entity.name}]] ")
|
||||
line = f"- {rel.relation_type} [[{rel.to_entity.name}]]"
|
||||
if rel.context:
|
||||
line += f" ({rel.context})"
|
||||
sections.append(line)
|
||||
sections.append("") # Empty line after relations
|
||||
|
||||
return "\n".join(sections)
|
||||
# Return joined sections, ensure content isn't empty
|
||||
content = "\n".join(sections).strip()
|
||||
return content if content else f"# {entity.name}"
|
||||
|
||||
@@ -566,6 +566,32 @@ async def test_update_entity_basic(client: AsyncClient):
|
||||
assert updated["entity_metadata"]["status"] == "draft" # Preserved
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_entity_content_parameter(client: AsyncClient):
|
||||
"""Test content parameter controls content loading."""
|
||||
# Create test entity
|
||||
data = {
|
||||
"name": "TestContent",
|
||||
"entity_type": "test",
|
||||
"content": "# Test Content\n\nSome test content."
|
||||
}
|
||||
response = await client.post("/knowledge/entities", json={"entities": [data]})
|
||||
assert response.status_code == 200
|
||||
path_id = response.json()["entities"][0]["path_id"]
|
||||
|
||||
# Get without content
|
||||
response = await client.get(f"/knowledge/entities/{path_id}")
|
||||
assert response.status_code == 200
|
||||
entity = response.json()
|
||||
assert entity["content"] is None
|
||||
|
||||
# Get with content
|
||||
response = await client.get(f"/knowledge/entities/{path_id}?content=true")
|
||||
assert response.status_code == 200
|
||||
entity = response.json()
|
||||
assert "# Test Content" in entity["content"]
|
||||
assert "Some test content" in entity["content"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_entity_content(client: AsyncClient):
|
||||
"""Test updating content for different entity types."""
|
||||
@@ -583,7 +609,7 @@ async def test_update_entity_content(client: AsyncClient):
|
||||
updated = response.json()
|
||||
|
||||
# Verify through get request to check file
|
||||
response = await client.get(f"/knowledge/entities/{updated['path_id']}")
|
||||
response = await client.get(f"/knowledge/entities/{updated['path_id']}?content=true")
|
||||
fetched = response.json()
|
||||
assert "# Updated Note" in fetched["content"]
|
||||
assert "New content" in fetched["content"]
|
||||
|
||||
@@ -77,27 +77,37 @@ async def test_format_frontmatter_with_metadata(
|
||||
assert frontmatter["id"] == "knowledge/test_entity"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_format_content_raw(knowledge_writer: KnowledgeWriter, sample_entity: Entity):
|
||||
"""Test raw content is preserved."""
|
||||
raw_content = "# Test Content\n\nThis is some test content."
|
||||
result = await knowledge_writer.format_content(sample_entity, raw_content)
|
||||
|
||||
assert result == raw_content
|
||||
assert "# test_entity" not in result # Shouldn't add title
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_format_content_basic(knowledge_writer: KnowledgeWriter, sample_entity: Entity):
|
||||
"""Test basic content formatting."""
|
||||
content = ""
|
||||
result = await knowledge_writer.format_content(sample_entity, content)
|
||||
"""Test basic content formatting without raw content."""
|
||||
result = await knowledge_writer.format_content(sample_entity)
|
||||
|
||||
assert "# test_entity" in result
|
||||
assert "Test description" in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_format_content_with_observations(
|
||||
async def test_format_content_structured(
|
||||
knowledge_writer: KnowledgeWriter, entity_with_observations: Entity
|
||||
):
|
||||
"""Test content formatting with observations."""
|
||||
content = ""
|
||||
result = await knowledge_writer.format_content(entity_with_observations, content)
|
||||
"""Test structured content generation."""
|
||||
result = await knowledge_writer.format_content(entity_with_observations)
|
||||
|
||||
# Should only have observation sections, not duplicate title
|
||||
assert "## Observations" in result
|
||||
assert "- [tech] First observation" in result
|
||||
assert "- [design] Second observation (Some context)" in result
|
||||
assert "# test_entity" not in result # No title needed
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -105,29 +115,76 @@ async def test_format_content_with_relations(
|
||||
knowledge_writer: KnowledgeWriter, entity_with_relations: Entity
|
||||
):
|
||||
"""Test content formatting with relations."""
|
||||
content = ""
|
||||
result = await knowledge_writer.format_content(entity_with_relations, content)
|
||||
result = await knowledge_writer.format_content(entity_with_relations)
|
||||
|
||||
assert "## Relations" in result
|
||||
assert "- connects_to [[target_entity]]" in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_format_content_full_entity(
|
||||
async def test_format_content_empty_returns_title(
|
||||
knowledge_writer: KnowledgeWriter, sample_entity: Entity
|
||||
):
|
||||
"""Test that empty content falls back to title."""
|
||||
sample_entity.summary = None # Remove summary
|
||||
result = await knowledge_writer.format_content(sample_entity)
|
||||
|
||||
assert result == "# test_entity"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_format_content_preserves_spacing(
|
||||
knowledge_writer: KnowledgeWriter, entity_with_observations: Entity
|
||||
):
|
||||
"""Test proper markdown spacing is maintained."""
|
||||
result = await knowledge_writer.format_content(entity_with_observations)
|
||||
lines = result.split("\n")
|
||||
|
||||
# Find sections and verify their format structure
|
||||
for i, line in enumerate(lines):
|
||||
if line == "## Observations":
|
||||
# Observations section should have format:
|
||||
# ## Observations
|
||||
# <!-- Format comment -->
|
||||
# <empty line>
|
||||
# - observation entries...
|
||||
assert "<!--" in lines[i+1], "Missing format comment after Observations"
|
||||
assert lines[i+2] == "", "Missing empty line after format comment"
|
||||
assert lines[i+3].startswith("- "), "Should start observations after empty line"
|
||||
|
||||
elif line == "## Relations":
|
||||
# Relations section should have format:
|
||||
# ## Relations
|
||||
# <!-- Format comment -->
|
||||
# <empty line>
|
||||
# - relation entries...
|
||||
assert "<!--" in lines[i+1], "Missing format comment after Relations"
|
||||
assert lines[i+2] == "", "Missing empty line after format comment"
|
||||
if i+3 < len(lines): # If there are relations
|
||||
assert lines[i+3].startswith("- "), "Should start relations after empty line"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_format_content_mixed(
|
||||
knowledge_writer: KnowledgeWriter,
|
||||
entity_with_relations: Entity,
|
||||
entity_with_observations: Entity,
|
||||
):
|
||||
"""Test content formatting with all entity features."""
|
||||
# Combine observations and relations
|
||||
"""Test content with both raw content and structured data."""
|
||||
# Add observations to entity with relations
|
||||
entity_with_relations.observations = entity_with_observations.observations
|
||||
content = ""
|
||||
result = await knowledge_writer.format_content(entity_with_relations, content)
|
||||
|
||||
# Verify all sections present
|
||||
assert "# test_entity" in result
|
||||
assert "Test description" in result
|
||||
|
||||
# Test with raw content
|
||||
raw_content = "# Custom Title\n\nSome content."
|
||||
result = await knowledge_writer.format_content(entity_with_relations, raw_content)
|
||||
|
||||
# Should preserve raw content
|
||||
assert result == raw_content
|
||||
assert "# test_entity" not in result
|
||||
|
||||
# Test without raw content - should generate structured
|
||||
result = await knowledge_writer.format_content(entity_with_relations)
|
||||
|
||||
assert "## Observations" in result
|
||||
assert "- [tech] First observation" in result
|
||||
assert "## Relations" in result
|
||||
assert "- connects_to [[target_entity]]" in result
|
||||
assert "- [tech] First observation" in result
|
||||
assert "- connects_to [[target_entity]]" in result
|
||||
Reference in New Issue
Block a user