feat(evolve): support direct entity auto-tagging

Add a single structured model call for entity extraction with evidence validation, avoiding the agent tool loop for predictable bulk tagging. Canonicalize persisted Windows paths in file graphs and tag indexes so tags and wikilinks resolve consistently.
This commit is contained in:
star-nebula 2026-11-03 15:00:21 +08:00
parent 99ecdd24d1
commit 09e58d0294
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."""