高级上下文工程：智能体的大脑

# 安装 semantica
pip install semantica

# 安装 Semantica 包
!pip install -qU semantica

import sys
import os
import time
from typing import Any, List, Dict, Optional

# 将项目根目录添加到路径以导入 semantica
sys.path.append(os.path.abspath(os.path.join(os.getcwd(), "../../")))

# 核心导入
from semantica.context import AgentContext, ContextGraph, AgentMemory
from semantica.vector_store import VectorStore
from semantica.graph_store import GraphStore

print("Libraries imported successfully.")

try:
    # 初始化 FAISS 向量存储
    # 你也可以使用：backend="weaviate"、backend="qdrant" 等。
    vs = VectorStore(backend="faiss", dimension=768)
    print("VectorStore initialized (Backend: FAISS)")
except ImportError:
    print("FAISS not installed. Using in-memory fallback (not persistent).")
    vs = VectorStore(backend="inmemory", dimension=768)
except Exception as e:
    print(f"VectorStore Error: {e}")
    vs = None

try:
    # 初始化 Neo4j 图存储
    # 确保你的 Docker 容器正在运行！
    gs = GraphStore(
        backend="neo4j",
        uri="bolt://localhost:7687",
        user="neo4j",
        password="password"
    )

    # 测试连接
    if gs.connect():
        print("GraphStore connected (Backend: Neo4j)")
    else:
        raise ConnectionError("Could not connect to Neo4j")

except Exception as e:
    print(f"GraphStore Connection Failed: {e}")
    print("   Switching to in-memory ContextGraph (Non-persistent fallback)")
    gs = ContextGraph() # 回退实现

# 基于向量存储与图谱存储构建智能体上下文
if vs:
    context = AgentContext(
        vector_store=vs,
        knowledge_graph=gs,
        retention_days=90,          # 记住 3 个月的内容
        use_graph_expansion=True,   # 启用 GraphRAG
        max_expansion_hops=2,       # 2 跳推理
        hybrid_alpha=0.6            # 平衡检索
    )
    print("Agent Context is online and ready.")
else:
    print("Cannot proceed without VectorStore.")

user_id = "user_123"
session_id = "session_alpha"

# 存储一条用户偏好
mem_id = context.store(
    content="I am working on a new project called 'Project Apollo' which uses Python and React.",
    conversation_id=session_id,
    user_id=user_id,
    metadata={"type": "user_preference"}
)
print(f"Memory Stored: {mem_id}")

documents = [
    {
        "content": "Project Apollo is a next-gen web framework designed for high scalability.",
        "metadata": {"source": "internal_wiki", "category": "projects"}
    },
    {
        "content": "Python 3.12 introduces significant performance improvements for async workloads.",
        "metadata": {"source": "tech_news", "category": "languages"}
    }
]

# 存储文档并触发图谱构建
stats = context.store(
    documents,
    extract_entities=True,      # 从文本中抽取实体
    extract_relationships=True, # 推断关系
    link_entities=True          # 连接到现有图谱节点
)

print("Knowledge Ingestion Stats:", stats)

# 1. 定义节点
entities = [
    {"id": "alice", "type": "PERSON", "text": "Alice", "properties": {"role": "Admin"}},
    {"id": "project_apollo", "type": "PROJECT", "text": "Project Apollo"},
    {"id": "python", "type": "TECH", "text": "Python"},
    {"id": "react", "type": "TECH", "text": "React"}
]

# 2. 定义边（知识）
relationships = [
    {"source": "alice", "target": "project_apollo", "type": "MANAGES", "weight": 1.0},
    {"source": "project_apollo", "target": "python", "type": "USES_TECH", "weight": 1.0},
    {"source": "project_apollo", "target": "react", "type": "USES_TECH", "weight": 1.0}
]

# 3. 注入图谱
graph_stats = context.build_graph(
    entities=entities,
    relationships=relationships
)

print("Manual Graph Build Complete:", graph_stats)

# 用于打印图谱邻居的辅助函数
def inspect_node(node_id):
    if hasattr(gs, "get_neighbors"):
        neighbors = gs.get_neighbors(node_id)
        print(f"\nNeighbors of '{node_id}':")
        for n in neighbors:
            # 处理不同存储之间不同的返回格式
            rel_type = n.get('relationship') or n.get('type') or 'linked'
            target = n.get('id') or n.get('node_id')
            print(f"   └── [{rel_type}] ──> {target}")
    else:
        print("Graph store does not support neighbor inspection.")

inspect_node("project_apollo")

query = "Who is responsible for the Python web framework project?"
print(f"Asking: '{query}'...\n")

results = context.retrieve(
    query,
    max_results=3,
    use_graph=True,         # 对找到 Alice 至关重要
    expand_graph=True,      # 跳到邻居
    include_entities=True   # 返回结构化实体数据
)

print(f"Retrieved {len(results)} context items:\n")

for i, res in enumerate(results, 1):
    print(f"{i}. [Score: {res['score']:.2f}] {res['content'][:120]}...")

    # 我们找到图谱连接了吗？
    if 'related_entities' in res and res['related_entities']:
        print("   Graph Insights:")
        for ent in res['related_entities'][:3]:
            print(f"      - {ent.get('text', 'Entity')} ({ent.get('type', 'Unknown')})")
    print("")

# 获取最近的聊天历史以用于上下文窗口
history = context.conversation(
    conversation_id=session_id,
    limit=5
)

print(f"Chat History for {session_id}:")
for msg in history:
    print(f" - {msg['content']}")

# 获取系统健康与统计信息
stats = context.stats()
print("System Vital Signs:")
print(f"   - Total Memories: {stats.get('total_items', 0)}")
print(f"   - Graph Nodes:    {stats.get('graph_stats', {}).get('node_count', 'N/A')}")
print(f"   - Graph Edges:    {stats.get('graph_stats', {}).get('edge_count', 'N/A')}")