上下文工程

from agkit.agents.storage import InMemoryMemory, TDAIMemory

# 默认上下文工程 (推荐)
memory = InMemoryMemory()

# 禁用自动上下文管理
manual_memory = InMemoryMemory(
    enable_context_management=False  # 禁用自动上下文管理
)

# 带上下文工程的自定义配置
memory = InMemoryMemory(
    enable_context_management=True,  # 启用自动上下文管理 (默认)
    thresholds={
        'pre_rot_threshold': 150_000,
        'compaction_trigger': 0.8,
        'summarization_trigger': 0.95,
        'keep_recent_count': 10,
    }
)

memory = InMemoryMemory(
    enable_context_management=False  # 禁用自动管理
)

# 添加事件而不进行自动上下文管理
await memory.add(event1, session_id='session-123')
await memory.add(event2, session_id='session-123')
await memory.add(event3, session_id='session-123')

# 在需要时手动触发上下文管理
await memory.manage_context('session-123')

from agkit.agents.storage import BaseMemory, IMemoryEvent, StructuredSummary
from datetime import datetime
from typing import List

# 方法 1: 在配置中传递摘要器
async def custom_summarizer(events: List[IMemoryEvent]) -> StructuredSummary:
    # 自定义摘要逻辑
    return {
        'count': len(events),
        'time_range': {
            'start': events[0]['message']['timestamp'],
            'end': events[-1]['message']['timestamp']
        },
        'timestamp': datetime.now()
    }

custom_memory = BaseMemory(
    summarizer=custom_summarizer,
    thresholds={
        'pre_rot_threshold': 100_000,
        'compaction_trigger': 0.8,
        'summarization_trigger': 0.95
    }
)

# 方法 2: 扩展 BaseMemory 以进行高级自定义
class CustomMemory(BaseMemory):
    def __init__(self, config=None):
        super().__init__({
            **(config or {}),
            'summarizer': self._custom_summarizer
        })

    # 覆盖单个事件的压缩逻辑
    async def _compact_event(self, event: IMemoryEvent) -> None:
        # 自定义压缩逻辑 - 示例: 压缩代码块
        if '```' in event['message']['content']:
            event['state']['__compaction__'] = {
                'original_content': event['message']['content'],
                'compacted_at': datetime.now().isoformat(),
                'tokens_saved': /* 计算节省的 tokens */,
                'strategy': 'code_compression'
            }
            event['message']['content'] = '[COMPACTED: 代码讨论]'
        else:
            await super()._compact_event(event)

    # 自定义摘要函数
    async def _custom_summarizer(self, events: List[IMemoryEvent]) -> StructuredSummary:
        return {
             'count': len(events),
              'time_range': {
                  'start': events[0]['message']['timestamp'],
                  'end': events[-1]['message']['timestamp']
              },
              'timestamp': datetime.now()
        }

    # 覆盖存储方法 (可选)
    async def _store_summary(self, session_id: str, summary: StructuredSummary) -> None:
        # 自定义存储逻辑
        print(f'为 {session_id} 存储摘要')

    async def _clear_summarized_events(self, session_id: str, keep_recent_count: int) -> None:
        # 自定义清理逻辑
        print(f'为 {session_id} 清理事件')
        await super()._clear_summarized_events(session_id, keep_recent_count)

    # 添加事件后触发上下文管理
    async def add(self, event: IMemoryEvent) -> None:
        await super().add(event)
        session_id = event['state'].get('session_id', 'default')
        await self._manage_context(session_id)

# 初始问题 (正常操作 - 第 1 层)
await memory.add({
    'message': {
        'role': 'user',
        'content': '我们的生产数据库在 30 秒后超时,影响了 1200 个用户'
    },
    'state': {
        'issue': 'db_timeout',
        'severity': 'critical',
        'environment': 'production'
    }
})

# 50 条消息后,上下文达到 120K tokens
# 🟡 系统自动触发第 2 层 (压缩)
compacted_event = {
    'message': {'content': '[COMPACTED: 数据库超时问题报告]'},
    'state': {
        'issue': 'db_timeout',
        'severity': 'critical',
        '__compaction__': {
            'original_content': '我们的生产数据库在 30 秒后超时...',
            'tokens_saved': 23,
            'compacted_at': '2024-01-15T10:30:00Z'
        }
    }
}

# 200 条消息后,上下文达到 142.5K tokens
# 🔴 系统触发第 3 层 (结构化摘要)
summary = {
    'modified_files': ['database.py', 'user_service.py', 'auth_middleware.py'],
    'user_goal': '修复生产环境中的数据库连接超时',
    'last_step': '更新了连接池配置',
    'key_decisions': [
        '将池大小从 10 增加到 50',
        '添加了连接重试逻辑',
        '实现了断路器模式'
    ],
    'critical_context': {
        'error_pattern': '30 秒后连接超时',
        'environment': 'production',
        'affected_users': 1200,
        'urgency': 'critical'
    }
}