mirror of
https://github.com/agentscope-ai/ReMe.git
synced 2026-09-30 01:52:29 +00:00
71e42dbad0
* up * up * up * up * up * up * up * up * up * up * up * feat(config): add daily_dir configuration and background job logging - Added daily_dir setting with default value 'memory' to config - Implemented logging for background job startup events - Enhanced component start logic to handle background backend type - Updated default YAML configuration structure * refactor(file_parser): replace _get_relative_path with to_vault_relative method - Remove redundant working_dir property from base file parser - Add to_vault_relative method to base component for path resolution - Update bare_file_parser to use new to_vault_relative method - Update default_file_parser to use new to_vault_relative method - Update linked_file_parser to use new to_vault_relative method - Make working_path absolute in base_component and steps - Simplify index_changes step by removing redundant base variable - Consolidate path relative logic in single shared method * docs(reme4): update report with detailed architecture sections - Add comprehensive Markdown kernel section covering Obsidian compatibility - Include detailed explanation of YAML front matter and wikilink formats - Document smart slicing mechanism using Markdown AST instead of fixed tokens - Explain graph indexing with bidirectional links and multiple backends - Restructure sections with proper numbering from 4 to 7 - Move Markdown kernel section to appear before self-evolution features - Add detailed explanations of auto-memory, auto-dream, and auto-link processes - Document three-way hybrid search with RRF fusion and progressive expansion - Include engineering value explanations for keyword indexing in Chinese context
89 lines
3 KiB
Python
89 lines
3 KiB
Python
"""Per-request runtime context shared across steps and jobs."""
|
|
|
|
import asyncio
|
|
|
|
from ..enumeration import ChunkEnum
|
|
from ..schema import Response, StreamChunk
|
|
|
|
|
|
class RuntimeContext:
|
|
"""Scratch space for a single execution.
|
|
|
|
Holds the response object, an optional stream queue, and a free-form
|
|
data dict accessed via mapping-style operators.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
response: Response | None = None,
|
|
stream_queue: asyncio.Queue | None = None,
|
|
stop_event: asyncio.Event | None = None,
|
|
**kwargs,
|
|
):
|
|
self.response: Response = response or Response()
|
|
self.stream_queue: asyncio.Queue | None = stream_queue
|
|
self.stop_event: asyncio.Event | None = stop_event
|
|
self.data: dict = kwargs
|
|
|
|
def get(self, key: str, default=None):
|
|
"""Get a value from the data dict."""
|
|
return self.data.get(key, default)
|
|
|
|
def update(self, data: dict) -> "RuntimeContext":
|
|
"""Merge data into the context."""
|
|
self.data.update(data)
|
|
return self
|
|
|
|
def __getitem__(self, key: str):
|
|
return self.data[key]
|
|
|
|
def __setitem__(self, key: str, value):
|
|
self.data[key] = value
|
|
|
|
def __delitem__(self, key: str):
|
|
del self.data[key]
|
|
|
|
def __contains__(self, key: str) -> bool:
|
|
return key in self.data
|
|
|
|
@property
|
|
def stream(self) -> bool:
|
|
"""Whether streaming is enabled."""
|
|
return self.stream_queue is not None
|
|
|
|
@classmethod
|
|
def from_context(cls, context: "RuntimeContext | None" = None, **kwargs) -> "RuntimeContext":
|
|
"""Reuse or create a RuntimeContext."""
|
|
# Reuse the existing context (merging kwargs) or create a new one.
|
|
if context is None:
|
|
return cls(**kwargs)
|
|
context.update(kwargs)
|
|
return context
|
|
|
|
async def _enqueue(self, chunk: StreamChunk) -> None:
|
|
"""Put a chunk on the stream queue."""
|
|
if self.stream_queue is None:
|
|
raise RuntimeError("Stream queue not initialized")
|
|
await self.stream_queue.put(chunk)
|
|
|
|
async def add_stream_string(self, chunk: str, chunk_type: ChunkEnum) -> "RuntimeContext":
|
|
"""Emit a text chunk to the stream queue."""
|
|
# Emit a text chunk to the stream queue.
|
|
await self._enqueue(StreamChunk(chunk_type=chunk_type, chunk=chunk))
|
|
return self
|
|
|
|
async def add_stream_done(self) -> "RuntimeContext":
|
|
"""Emit the terminal DONE marker to close the stream."""
|
|
# Emit the terminal DONE marker to close the stream.
|
|
await self._enqueue(StreamChunk(chunk_type=ChunkEnum.DONE, chunk="", done=True))
|
|
return self
|
|
|
|
def apply_mapping(self, mapping: dict[str, str]) -> "RuntimeContext":
|
|
"""Copy data[source] into data[target] for each mapping pair."""
|
|
# Copy data[source] into data[target] for each {source: target} pair.
|
|
if not mapping:
|
|
return self
|
|
for source, target in mapping.items():
|
|
if source in self.data:
|
|
self.data[target] = self.data[source]
|
|
return self
|