This commit is contained in:
star-nebula 2026-11-04 14:40:25 +08:00 • committed by GitHub
commit e09a75fbd6
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 401 additions and 36 deletions

View file

@ -19,6 +19,25 @@ class BaseFileGraph(BaseComponent):
component_type = ComponentEnum.FILE_GRAPH
@staticmethod
def normalize_path(path: str) -> str:
"""Normalize persisted Windows separators to workspace-relative POSIX."""
return path.replace("\\", "/")
@classmethod
def canonical_node(cls, node: FileNode) -> FileNode:
"""Return a node whose path and links use canonical POSIX separators."""
links = [
link.model_copy(
update={
"source_path": cls.normalize_path(link.source_path),
"target_path": cls.normalize_path(link.target_path),
},
)
for link in node.links
]
return node.model_copy(update={"path": cls.normalize_path(node.path), "links": links})
# -- Lifecycle ---------------------------------------------------------
async def _start(self) -> None:

View file

@ -33,7 +33,7 @@ class LocalFileGraph(BaseFileGraph):
try:
for line in read_jsonl_zst(self._graph_file):
if line.strip():
node = FileNode.model_validate_json(line)
node = self.canonical_node(FileNode.model_validate_json(line))
self._nodes[node.path] = node
self.logger.debug(f"Loaded {len(self._nodes)} nodes from {self._graph_file}")
except Exception as e:
@ -78,7 +78,8 @@ class LocalFileGraph(BaseFileGraph):
# -- Node CRUD ---------------------------------------------------------
async def upsert_nodes(self, nodes: list[FileNode]) -> None:
for node in nodes:
for raw_node in nodes:
node = self.canonical_node(raw_node)
path = node.path
old = self._nodes.get(path)
if old is not None:
@ -92,7 +93,8 @@ class LocalFileGraph(BaseFileGraph):
self._inverse.setdefault(path, set()).update(promoted)
async def delete_nodes(self, paths: list[str]) -> None:
for path in paths:
for raw_path in paths:
path = self.normalize_path(raw_path)
node = self._nodes.pop(path, None)
if node is None:
continue
@ -105,7 +107,7 @@ class LocalFileGraph(BaseFileGraph):
async def get_nodes(self, paths: list[str] | None = None) -> list[FileNode]:
if paths is None:
return list(self._nodes.values())
return [self._nodes[p] for p in paths if p in self._nodes]
return [self._nodes[self.normalize_path(p)] for p in paths if self.normalize_path(p) in self._nodes]
async def rebuild_links(self) -> None:
self._inverse.clear()
@ -124,13 +126,14 @@ class LocalFileGraph(BaseFileGraph):
async def get_outlinks(self, path: str, scope: LinkScopeEnum | str = LinkScopeEnum.REAL) -> list[FileLink]:
scope = self._normalize_scope(scope)
node = self._nodes.get(path)
node = self._nodes.get(self.normalize_path(path))
if node is None:
return []
return [lnk for lnk in node.links if lnk.target_path and self._scope_match(lnk.target_path, scope)]
async def get_inlinks(self, path: str, scope: LinkScopeEnum | str = LinkScopeEnum.REAL) -> list[FileLink]:
scope = self._normalize_scope(scope)
path = self.normalize_path(path)
sources: set[str] = set()
if scope in (LinkScopeEnum.REAL, LinkScopeEnum.ALL):
sources |= self._inverse.get(path, set())

View file

@ -69,7 +69,8 @@ class NxFileGraph(BaseFileGraph):
# -- Node CRUD ---------------------------------------------------------
async def upsert_nodes(self, nodes: list[FileNode]) -> None:
for node in nodes:
for raw_node in nodes:
node = self.canonical_node(raw_node)
path = node.path
if self._graph.has_node(path):
self._graph.remove_edges_from(list(self._graph.out_edges(path, keys=True)))
@ -77,7 +78,8 @@ class NxFileGraph(BaseFileGraph):
self._graph.add_edges_from(self._edges_from(path, node))
async def delete_nodes(self, paths: list[str]) -> None:
for path in paths:
for raw_path in paths:
path = self.normalize_path(raw_path)
if not self._graph.has_node(path):
continue
self._graph.remove_edges_from(list(self._graph.out_edges(path, keys=True)))
@ -89,14 +91,19 @@ class NxFileGraph(BaseFileGraph):
view = self._graph.nodes
if paths is None:
return [d["node"] for _, d in view(data=True) if "node" in d]
return [view[p]["node"] for p in paths if p in view and "node" in view[p]]
normalized = [self.normalize_path(p) for p in paths]
return [view[p]["node"] for p in normalized if p in view and "node" in view[p]]
async def rebuild_links(self) -> None:
self._graph.remove_edges_from(list(self._graph.edges(keys=True)))
virtual = [n for n, d in self._graph.nodes(data=True) if "node" not in d]
self._graph.remove_nodes_from(virtual)
for path, data in list(self._graph.nodes(data=True)):
self._graph.add_edges_from(self._edges_from(path, data["node"]))
canonical = {
self.normalize_path(data["node"].path): self.canonical_node(data["node"])
for _, data in self._graph.nodes(data=True)
if "node" in data
}
self._graph.clear()
for node in canonical.values():
self._graph.add_node(node.path, node=node)
self._graph.add_edges_from(self._edges_from(node.path, node))
async def clear(self):
self._graph.clear()
@ -105,6 +112,7 @@ class NxFileGraph(BaseFileGraph):
# -- Link access -------------------------------------------------------
async def get_outlinks(self, path: str, scope: LinkScopeEnum = LinkScopeEnum.REAL) -> list[FileLink]:
path = self.normalize_path(path)
view = self._graph.nodes
if path not in view or "node" not in view[path]:
return []
@ -115,6 +123,7 @@ class NxFileGraph(BaseFileGraph):
]
async def get_inlinks(self, path: str, scope: LinkScopeEnum = LinkScopeEnum.REAL) -> list[FileLink]:
path = self.normalize_path(path)
view = self._graph.nodes
if path not in view or not self._scope_match(path, scope):
return []

View file

@ -1,7 +1,7 @@
"""In-memory tag index derived from ``FileNode.front_matter``."""
import asyncio
from pathlib import PurePosixPath
from pathlib import PurePosixPath, PureWindowsPath
from .base_tag_index import BaseTagIndex, TagListItem, TagListResult, TagOrder, TagOrderBy
from ..component_registry import R
@ -63,12 +63,18 @@ class LocalTagIndex(BaseTagIndex):
@staticmethod
def _validate_path(path: str) -> str:
if not isinstance(path, str) or not path or "\\" in path:
if not isinstance(path, str) or not path:
raise ValueError(f"Invalid workspace-relative tag-index path: {path!r}")
pure = PurePosixPath(path)
if pure.is_absolute() or path != pure.as_posix() or any(part in ("", ".", "..") for part in pure.parts):
normalized = path.replace("\\", "/")
pure = PurePosixPath(normalized)
if (
pure.is_absolute()
or PureWindowsPath(path).anchor
or normalized != pure.as_posix()
or any(part in ("", ".", "..") for part in pure.parts)
):
raise ValueError(f"Invalid workspace-relative tag-index path: {path!r}")
return path
return normalized
def _prepare_nodes(self, nodes: list[FileNode]) -> list[tuple[str, tuple[str, ...]]]:
prepared: list[tuple[str, tuple[str, ...]]] = []

View file

@ -1,8 +1,10 @@
"""Generate entity-oriented memory tags for added or modified Markdown files."""
from dataclasses import dataclass
import json
from typing import Literal
from agentscope.message import SystemMsg, UserMsg
import frontmatter
from ._evolve import agent_reply_result_text
@ -53,9 +55,18 @@ def normalize_memory_tags(
class AutoTagStep(BaseStep):
"""Update memory tags for Markdown files described by the common ``changes`` contract."""
def __init__(self, **kwargs):
def __init__(
self,
direct_llm: bool = False,
direct_excerpt_chars: int = 4500,
direct_candidate_tags: int = 80,
**kwargs,
):
super().__init__(**kwargs)
self.tools = ["read", "list_tags", "frontmatter_read", "frontmatter_update"]
self.direct_llm = direct_llm
self.direct_excerpt_chars = max(500, direct_excerpt_chars)
self.direct_candidate_tags = max(0, direct_candidate_tags)
def _targets(self) -> tuple[list[_TagTarget], list[dict[str, str]]]:
"""Validate, normalize, and de-duplicate added/modified Markdown changes."""
@ -101,6 +112,140 @@ class AutoTagStep(BaseStep):
targets[path] = _TagTarget(change=normalized_change, path=path)
return list(targets.values()), ignored
@staticmethod
def _document_excerpt(content: str, max_chars: int) -> str:
"""Keep the title-adjacent text and heading outline within a local model budget."""
if len(content) <= max_chars:
return content
headings = [
line.strip()
for line in content.splitlines()
if line.lstrip().startswith("#") and line.strip()
]
heading_text = "\n".join(headings)
if len(heading_text) > max_chars // 3:
heading_text = heading_text[: max_chars // 3]
body_budget = max(0, max_chars - len(heading_text) - 64)
head = content[: body_budget // 2]
tail = content[-(body_budget - len(head)) :] if body_budget > len(head) else ""
return f"{head}\n\n[Headings]\n{heading_text}\n\n[Tail]\n{tail}"
@staticmethod
def _evidence_key(text: str) -> str:
"""Normalize text for conservative evidence checks."""
return "".join(char.casefold() for char in text if char.isalnum())
async def _direct_tags(self, target: _TagTarget) -> list[str]:
"""Generate tags in one structured call without an agent tool loop."""
wrapper = self.agent_wrapper
model = getattr(getattr(wrapper, "as_llm", None), "model", None)
if model is None:
raise RuntimeError("Direct auto-tag mode requires an agent wrapper with a bound LLM")
file_store = self.file_store
tag_index = file_store.require_tag_index()
tag_key = tag_index.tag_key
path = self.workspace_path / target.path
post = frontmatter.loads(path.read_text(encoding="utf-8"))
metadata = post.metadata or {}
candidates: list[str] = []
if self.direct_candidate_tags:
page = await tag_index.list_tags(
order_by="file_count",
order="desc",
page_size=self.direct_candidate_tags,
)
candidates = [item[0] for item in page["items"]]
excerpt = self._document_excerpt(post.content, self.direct_excerpt_chars)
metadata_hint = {
key: metadata[key]
for key in ("name", "description", "kind")
if key in metadata
}
messages = [
SystemMsg(
name="system",
content=self.prompt_format(
"direct_system_prompt",
tag_key=tag_key,
max_tags_per_file=tag_index.max_tags_per_file,
existing_tags=", ".join(candidates) or "(none)",
),
),
UserMsg(
name="user",
content=self.prompt_format(
"direct_user_message",
path=target.path,
metadata_json=json.dumps(metadata_hint, ensure_ascii=False, indent=2),
excerpt=excerpt,
),
),
]
schema = {
"type": "object",
"properties": {
"entity_type": {
"type": "string",
"enum": [
"none",
"person",
"organization",
"product",
"place",
"asset",
"named_project",
],
},
"evidence": {
"type": "string",
"description": "Exact short quote from the document body or description supporting the entity",
},
tag_key: {
"type": "array",
"items": {"type": "string"},
"maxItems": tag_index.max_tags_per_file,
},
},
"required": ["entity_type", "evidence", tag_key],
"additionalProperties": False,
}
response = await model.generate_structured_output(
messages=messages,
structured_model=schema,
)
content = response.content
if hasattr(content, "model_dump"):
content = content.model_dump()
if not isinstance(content, dict):
raise RuntimeError("Direct auto-tag model did not return an object")
normalized = normalize_memory_tags(
content.get(tag_key),
max_tags_per_file=tag_index.max_tags_per_file,
max_tag_length=tag_index.max_tag_length,
)
if str(content.get("entity_type") or "").strip().lower() == "none":
return []
evidence = str(content.get("evidence") or "").strip()
body_without_headings = "\n".join(
line for line in post.content.splitlines() if not line.lstrip().startswith("#")
)
searchable = "\n".join(
value
for value in (str(metadata.get("description") or ""), body_without_headings)
if value
)
evidence_key = self._evidence_key(evidence)
searchable_key = self._evidence_key(searchable)
if not evidence_key or evidence_key not in searchable_key:
self.logger.warning(f"[{self.name}] rejected unsupported tag evidence path={target.path}")
return []
supported = [tag for tag in normalized if self._evidence_key(tag) in evidence_key]
if len(supported) != len(normalized):
self.logger.warning(f"[{self.name}] rejected tags without evidence path={target.path}")
return supported
async def _process_target(self, target: _TagTarget) -> str:
file_store = self.file_store
tag_index = file_store.require_tag_index()
@ -110,26 +255,33 @@ class AutoTagStep(BaseStep):
"_allowed_paths": [target.path],
"_allowed_frontmatter_keys": [tag_key],
}
result = await self.agent_wrapper.reply(
self.prompt_format("user_message", path=target.path, change=target.change, tag_key=tag_key),
system_prompt=self.prompt_format(
"system_prompt",
tag_key=tag_key,
max_tags_per_file=tag_index.max_tags_per_file,
),
job_tools=self.tools,
injected_job_kwargs=tool_context,
)
if self.direct_llm:
generated_tags = await self._direct_tags(target)
result_text = f"Generated tags: {json.dumps(generated_tags, ensure_ascii=False)}"
else:
result = await self.agent_wrapper.reply(
self.prompt_format("user_message", path=target.path, change=target.change, tag_key=tag_key),
system_prompt=self.prompt_format(
"system_prompt",
tag_key=tag_key,
max_tags_per_file=tag_index.max_tags_per_file,
),
job_tools=self.tools,
injected_job_kwargs=tool_context,
)
path = self.workspace_path / target.path
metadata = dict(frontmatter.loads(path.read_text(encoding="utf-8")).metadata or {})
if tag_key not in metadata:
if self.direct_llm:
normalized = generated_tags
elif tag_key not in metadata:
raise RuntimeError(f"Auto-tag agent did not write the required {tag_key!r} field")
normalized = normalize_memory_tags(
metadata.get(tag_key),
max_tags_per_file=tag_index.max_tags_per_file,
max_tag_length=tag_index.max_tag_length,
)
else:
normalized = normalize_memory_tags(
metadata.get(tag_key),
max_tags_per_file=tag_index.max_tags_per_file,
max_tag_length=tag_index.max_tag_length,
)
if metadata.get(tag_key) != normalized:
response = await self.run_job(
"frontmatter_update",
@ -139,7 +291,7 @@ class AutoTagStep(BaseStep):
)
if not response.success:
raise RuntimeError(str(response.answer))
return agent_reply_result_text(result)
return result_text if self.direct_llm else agent_reply_result_text(result)
async def execute(self):
assert self.context is not None

View file

@ -33,6 +33,66 @@ system_prompt_zh: |
只选择实体,不要使用宽泛主题、行为、属性、日期或普通关键词。每个实体使用一个简洁、规范的现实名称。
多个实体必须保存为独立的 YAML 列表项,例如 `[宁德时代, 黄金]`;不要把它们拼成一个逗号分隔的字符串。
不要为已有实体创造别名或近似拼写。只能更新 `{tag_key}`,绝不能修改正文或其他 frontmatter 字段。
direct_system_prompt: |
You assign entity-oriented memory tags to one Markdown note in a single pass.
The `{tag_key}` field must name the central real-world entity: a person, organization, product,
place, asset, or named project. Do not tag broad topics, capabilities, tasks, workflows, dates,
attributes, or generic keywords. If no named entity is central, return an empty list.
Also return `entity_type` and a short exact `evidence` quote from the document body or description.
The evidence must explicitly contain every returned entity name. If the evidence only proves a
capability, request, workflow, error, title, or filename, return `entity_type="none"`, `evidence=""`,
and an empty list.
Prefer an existing tag when it names the same entity. Tags must not contain spaces; replace spaces
with underscores. Return at most {max_tags_per_file} tags.
Counterexamples:
- "A request to list controllable desktop applications" -> []
- "Notes about code review in general" -> []
- "OpenAI releases a new API" -> ["OpenAI"]
- "Sam Altman discusses agent safety" -> ["sam_altman"]
Existing workspace tags, ordered by usage:
{existing_tags}
direct_system_prompt_zh: |
你需要在一次调用内为一份 Markdown 记忆生成以实体为中心的记忆标签。
`{tag_key}` 字段只能填写文档真正围绕的现实实体:人物、组织、产品、地点、资产或具名项目。
不要标记宽泛主题、能力、任务、流程、日期、属性或普通关键词。如果文档没有核心具名实体,返回空列表。
同时返回 `entity_type` 和文档正文或 description 中一段简短的原文 `evidence`。evidence 必须明确
包含每一个返回的实体;如果证据只能说明能力、请求、流程、错误、标题或文件名,必须返回
`entity_type="none"`、`evidence=""` 和空列表。
同一实体已有标签时必须优先复用。标签不能包含空格,空格必须替换为下划线。最多返回
{max_tags_per_file} 个标签。
反例:
- “列出可控制的桌面应用” -> []
- “关于代码审查的一般经验” -> []
- “OpenAI 发布新 API” -> ["OpenAI"]
- “Sam Altman 讨论智能体安全” -> ["sam_altman"]
工作区已有标签(按使用频率排序):
{existing_tags}
direct_user_message: |
Path: {path}
Frontmatter metadata:
{metadata_json}
Document excerpt:
{excerpt}
direct_user_message_zh: |
路径:{path}
Frontmatter 元数据:
{metadata_json}
文档摘录:
{excerpt}
user_message: |
Change: {change}
Target path: {path}

View file

@ -3,6 +3,7 @@
# pylint: disable=missing-function-docstring
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock
import frontmatter
@ -45,6 +46,12 @@ class _TaggingWrapper(BaseAgentWrapper):
return {"result": f"tagged {path}", "last_message": {}}
class _DirectTaggingWrapper(_TaggingWrapper):
def __init__(self, workspace: Path, model) -> None:
super().__init__(workspace)
self.as_llm = SimpleNamespace(model=model)
def _write_note(path: Path) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("---\nname: note\ndescription: useful note\n---\nbody\n", encoding="utf-8")
@ -226,3 +233,68 @@ async def test_auto_tag_uses_configured_key_and_normalizes_agent_output(tmp_path
"宁德时代",
]
assert wrapper.calls[0][1]["injected_job_kwargs"]["file_store"] == "archive"
@pytest.mark.asyncio
async def test_auto_tag_direct_llm_uses_one_structured_call_and_normalizes_output(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
note = tmp_path / "memory/note.md"
note.parent.mkdir(parents=True, exist_ok=True)
note.write_text(
"---\n"
"name: note\n"
"description: OpenAI and Sam Altman are central to this note.\n"
"---\n"
"OpenAI and Sam Altman are central to this note.\n",
encoding="utf-8",
)
class _StructuredTagModel:
def __init__(self) -> None:
self.calls: list[dict] = []
async def generate_structured_output(self, **kwargs):
self.calls.append(kwargs)
return SimpleNamespace(
content={
"entity_type": "organization",
"evidence": "OpenAI and Sam Altman are central to this note.",
"keywords": ["OpenAI", "openai", "Sam Altman", "++"],
},
)
model = _StructuredTagModel()
store = _file_store(name="archive", tag_key="keywords", max_tags_per_file=2, max_tag_length=12)
wrapper = _DirectTaggingWrapper(tmp_path, model)
step = AutoTagStep(
agent_wrapper=wrapper,
direct_llm=True,
direct_candidate_tags=0,
file_store=store,
)
async def update_frontmatter(name, /, **kwargs):
assert name == "frontmatter_update"
assert kwargs["_allowed_frontmatter_keys"] == ["keywords"]
assert kwargs["file_store"] == "archive"
post = frontmatter.loads(note.read_text(encoding="utf-8"))
post.metadata.update(kwargs["metadata"])
note.write_text(frontmatter.dumps(post), encoding="utf-8")
return Response(answer="updated")
monkeypatch.setattr(step, "run_job", update_frontmatter)
response = await step(RuntimeContext(changes=[{"change": "modified", "path": "memory/note.md"}]))
assert response.metadata["auto_tag"]["succeeded"] == 1
assert frontmatter.loads(note.read_text(encoding="utf-8")).metadata["keywords"] == [
"OpenAI",
"Sam_Altman",
]
assert len(model.calls) == 1
assert model.calls[0]["structured_model"]["required"] == [
"entity_type",
"evidence",
"keywords",
]
assert "Frontmatter metadata" in str(model.calls[0]["messages"][1].content)

View file

@ -101,6 +101,38 @@ async def test_upsert_replaces_old_links(backend_cls):
await graph.close()
@pytest.mark.parametrize("backend_cls", BACKENDS)
@pytest.mark.asyncio
async def test_windows_separator_paths_are_canonicalized(backend_cls):
"""Persisted Windows paths and POSIX wikilink targets form one graph."""
with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
graph = backend_cls()
await graph.start()
await graph.upsert_nodes(
[
make_node(r"daily\2026-10-27\a.md", [(r"daily\2026-10-27\b.md", None)]),
make_node(r"daily\2026-10-27\b.md"),
],
)
assert {node.path for node in await graph.get_nodes()} == {
"daily/2026-10-27/a.md",
"daily/2026-10-27/b.md",
}
assert {
link.target_path
for link in await graph.get_outlinks(r"daily\2026-10-27\a.md", scope=LinkScopeEnum.ALL)
} == {"daily/2026-10-27/b.md"}
assert {
link.source_path
for link in await graph.get_inlinks(r"daily\2026-10-27\b.md", scope=LinkScopeEnum.ALL)
} == {"daily/2026-10-27/a.md"}
assert await graph.get_outlinks(r"daily\2026-10-27\a.md", scope=LinkScopeEnum.VIRTUAL) == []
await graph.close()
@pytest.mark.parametrize("backend_cls", BACKENDS)
@pytest.mark.asyncio
async def test_outlinks_include_virtual_targets(backend_cls):

View file

@ -56,6 +56,18 @@ async def test_tag_normalization_and_bidirectional_mutations() -> None:
await index.close()
@pytest.mark.asyncio
async def test_tag_index_normalizes_windows_separators() -> None:
"""Migrate persisted Windows paths to the canonical POSIX key."""
index = LocalTagIndex()
await index.rebuild([_node(r"daily\2026-10-27\note.md", ["ReMe"])])
assert await index.tags_for_path(r"daily\2026-10-27\note.md") == ["reme"]
assert await index.paths_for_tags(["reme"]) == ["daily/2026-10-27/note.md"]
assert index.path_to_tags == {"daily/2026-10-27/note.md": ("reme",)}
@pytest.mark.asyncio
async def test_rebuild_is_atomic_and_supports_all_or_any_queries() -> None:
"""Publish complete rebuilds atomically and support intersection and union lookup."""