{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# 课程13 - 带有Cognee知识图谱的代理记忆\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 设置\n", "\n", "本笔记本演示如何使用 [**Cognee**](https://www.cognee.ai/) 知识图谱和 **Microsoft Agent Framework** (MAF) 构建具有持久内存的智能编码助手。\n", "\n", "Cognee 将非结构化文本转化为结构化、可查询的知识图谱，背后由向量嵌入支持——为您的代理提供丰富的、关系感知的长期记忆。\n", "\n", "### 您将学习到\n", "1. 构建知识图谱: 将开发者档案和最佳实践转化为结构化、可查询的知识。\n", "2. **集成 Cognee 与 MAF**: 使用 `@tool` 函数让 MAF 代理查询 Cognee 的知识图谱。\n", "3. 会话感知对话: 维护同一会话中多个问题的上下文。\n", "4. 长期记忆: 在会话之间持久保存重要知识，并在新对话中检索。\n", "\n", "### 先决条件\n", "- Python 3.9+\n", "- 本地运行的 Redis (`docker run -d -p 6379:6379 redis`) 用于会话管理\n", "- 一个 LLM API 密钥（例如 OpenAI）——在 `.env` 中设置 `LLM_API_KEY`\n", "- `.env` 中设置 `CACHING=true`（Cognee 会话所需）\n", "- 一个部署了聊天模型的 Microsoft Foundry 项目\n", "- `.env` 中设置 `AZURE_AI_PROJECT_ENDPOINT` 和 `AZURE_AI_MODEL_DEPLOYMENT_NAME`\n", "- 已通过 Azure CLI 验证身份（`az login`）\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "%pip install agent-framework azure-ai-projects azure-identity \"cognee[redis]==0.4.0\" -q" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import os\n", "from pathlib import Path\n", "from typing import Annotated\n", "\n", "from dotenv import load_dotenv\n", "\n", "load_dotenv()\n", "\n", "os.environ[\"LLM_API_KEY\"] = os.getenv(\"LLM_API_KEY\", \"\")\n", "os.environ[\"CACHING\"] = os.getenv(\"CACHING\", \"true\")\n", "\n", "import cognee\n", "from cognee.modules.search.types import SearchType\n", "\n", "from agent_framework import tool\n", "from agent_framework.foundry import FoundryChatClient\n", "from azure.identity import AzureCliCredential\n", "\n", "print(f\"Cognee version: {cognee.__version__}\")\n", "print(f\"CACHING: {os.environ.get('CACHING')}\")\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "provider = FoundryChatClient(\n", " project_endpoint=os.environ[\"AZURE_AI_PROJECT_ENDPOINT\"],\n", " model=os.environ[\"AZURE_AI_MODEL_DEPLOYMENT_NAME\"],\n", " credential=AzureCliCredential(),\n", ")\n", "\n", "print(\"✅ FoundryChatClient created\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 代理记忆类型\n", "\n", "本笔记本探讨了主课程第13课笔记本中的相同三种记忆类型，但使用 Cognee 作为长期记忆后端：\n", "\n", "| 记忆类型 | 机制 | 生命周期 |\n", "|---|---|---|\n", "| 工作记忆 | `agent.create_session()` (MAF) | 单次对话 |\n", "| 短期记忆 | Cognee 会话缓存 (Redis) | 单次会话 |\n", "| 长期记忆 | Cognee 知识图谱 + 向量 | 永久 |\n", "\n", "### Cognee 的记忆架构\n", "```\n", "┌──────────────────────────┐\n", "│ Raw Data │ (developer profiles, docs, conversations)\n", "└───────────┬──────────────┘\n", " │ cognee.add() + cognee.cognify()\n", " ▼\n", "┌──────────────────────────────────────────┐\n", "│ Knowledge Graph + Vector Embeddings │\n", "└───────────┬──────────────────────────────┘\n", " │ cognee.search()\n", " ▼\n", "┌──────────────────┐ ┌────────────────┐\n", "│ MAF Agent │──────▶│ @tool funcs │\n", "│ (AgentSession) │ │ wrapping │\n", "│ │ │ cognee.search │\n", "└──────────────────┘ └────────────────┘\n", "```\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 准备 Cognee 存储\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "DATA_ROOT = Path('.data_storage').resolve()\n", "SYSTEM_ROOT = Path('.cognee_system').resolve()\n", "\n", "DATA_ROOT.mkdir(parents=True, exist_ok=True)\n", "SYSTEM_ROOT.mkdir(parents=True, exist_ok=True)\n", "\n", "cognee.config.data_root_directory(str(DATA_ROOT))\n", "cognee.config.system_root_directory(str(SYSTEM_ROOT))\n", "\n", "await cognee.prune.prune_data()\n", "await cognee.prune.prune_system(metadata=True)\n", "print(\"✅ Cognee storage configured and reset\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 第一部分 — 构建知识库\n", "\n", "我们摄取三种类型的数据，以创建一个全面的编程助手知识库：\n", "\n", "1. 开发者个人资料 — 个人专业知识和技术背景\n", "2. **Python最佳实践** — Python之禅及实用指南\n", "3. 历史对话 — 开发者与AI助手之间的过去问答记录\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "developer_intro = (\n", " \"Hi, I'm an AI/Backend engineer. \"\n", " \"I build FastAPI services with Pydantic, heavy asyncio/aiohttp pipelines, \"\n", " \"and production testing via pytest-asyncio. \"\n", " \"I've shipped low-latency APIs on AWS, Azure, and GoogleCloud.\"\n", ")\n", "\n", "python_zen_principles = \"\"\"\n", "# The Zen of Python: Practical Guide\n", "\n", "## Key Principles With Guidance\n", "\n", "### 1. Beautiful is better than ugly\n", "Prefer descriptive names, clear structure, and consistent formatting.\n", "\n", "### 2. Explicit is better than implicit\n", "Be clear about behavior, imports, and types.\n", "\n", "### 3. Simple is better than complex\n", "Choose straightforward solutions first.\n", "\n", "### 4. Flat is better than nested\n", "Use early returns to reduce indentation.\n", "\n", "## Modern Python Tie-ins\n", "- Type hints reinforce explicitness\n", "- Context managers enforce safe resource handling\n", "- Dataclasses improve readability for data containers\n", "\"\"\"\n", "\n", "human_agent_conversations = \"\"\"\n", "\"conversations\": [\n", " {\n", " \"topic\": \"async/await patterns\",\n", " \"user_query\": \"I'm building a web scraper that needs to handle thousands of URLs concurrently. What's the best way to structure this with asyncio?\",\n", " \"assistant_response\": \"Use asyncio with aiohttp, a semaphore to cap concurrency, TCPConnector for connection pooling, and context managers for session lifecycle.\"\n", " },\n", " {\n", " \"topic\": \"dataclass vs pydantic\",\n", " \"user_query\": \"When should I use dataclasses vs Pydantic models?\",\n", " \"assistant_response\": \"For API input/output, prefer Pydantic: runtime validation, type coercion, JSON serialization. Integrates cleanly with FastAPI.\"\n", " },\n", " {\n", " \"topic\": \"testing patterns\",\n", " \"user_query\": \"What's the best approach for pytest with async functions?\",\n", " \"assistant_response\": \"Use pytest-asyncio, async fixtures, and an isolated test database or mocks to reliably test async code.\"\n", " },\n", " {\n", " \"topic\": \"error handling and logging\",\n", " \"user_query\": \"What's the best approach for production-ready error management?\",\n", " \"assistant_response\": \"Centralized error handling with custom exceptions, structured logging, and FastAPI middleware.\"\n", " }\n", " ]\n", "\"\"\"\n", "\n", "print(\"✅ Data sources prepared\")" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "await cognee.add(developer_intro, node_set=[\"developer_data\"])\n", "await cognee.add(human_agent_conversations, node_set=[\"developer_data\"])\n", "await cognee.add(python_zen_principles, node_set=[\"principles_data\"])\n", "\n", "await cognee.cognify()\n", "print(\"✅ Knowledge graph built\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 可视化知识图谱\n", "\n", "Cognee 可以渲染它提取的实体和关系的交互式 HTML 可视化。\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from cognee import visualize_graph\n", "\n", "await visualize_graph('./cognee_graph.html')\n", "print(\"📊 Graph saved to cognee_graph.html — open it in a browser to explore.\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 用 Memify 丰富记忆\n", "\n", "`memify()` 分析知识图并生成智能规则 —— 识别模式、最佳实践以及概念之间的关系。\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "await cognee.memify()\n", "print(\"✅ Memory enriched with memify\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 第二部分 — 使用 Cognee 工具的 MAF 代理\n", "\n", "现在我们创建一个可以通过 `@tool` 函数查询 Cognee 知识图谱的 MAF 代理。这样代理就能利用图结构感知的语义搜索的全部优势，同时通过会话保持对话上下文。\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "@tool(approval_mode=\"never_require\")\n", "async def search_knowledge(\n", " query: Annotated[str, \"Natural-language question to search the knowledge graph\"],\n", ") -> str:\n", " \"\"\"Search the Cognee knowledge graph for relevant developer knowledge, best practices, and past conversations.\"\"\"\n", " results = await cognee.search(\n", " query_text=query,\n", " query_type=SearchType.GRAPH_COMPLETION,\n", " )\n", " if not results:\n", " return \"No relevant knowledge found.\"\n", " return str(results)\n", "\n", "\n", "@tool(approval_mode=\"never_require\")\n", "async def search_principles(\n", " query: Annotated[str, \"Question about Python principles or best practices\"],\n", ") -> str:\n", " \"\"\"Search only the Python principles subset of the knowledge graph.\"\"\"\n", " from cognee.modules.engine.models.node_set import NodeSet\n", " results = await cognee.search(\n", " query_text=query,\n", " query_type=SearchType.GRAPH_COMPLETION,\n", " node_type=NodeSet,\n", " node_name=[\"principles_data\"],\n", " )\n", " if not results:\n", " return \"No relevant principles found.\"\n", " return str(results)\n", "\n", "\n", "print(\"✅ Cognee tools defined: search_knowledge, search_principles\")" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "coding_agent = provider.as_agent(\n", " name=\"CodingAssistant\",\n", " instructions=(\n", " \"You are an expert coding assistant with access to a knowledge graph \"\n", " \"containing developer profiles, Python best practices, and past conversations.\\n\\n\"\n", " \"WORKFLOW:\\n\"\n", " \"1. Use search_knowledge() to find relevant information from the full knowledge graph.\\n\"\n", " \"2. Use search_principles() when the question is specifically about Python best practices.\\n\"\n", " \"3. Combine retrieved knowledge with your own expertise to give comprehensive answers.\\n\"\n", " \"4. Reference the developer's known tech stack (FastAPI, asyncio, Pydantic) when relevant.\"\n", " ),\n", ")\n", "\n", "print(\"✅ CodingAssistant agent created\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 使用会话的工作记忆\n", "\n", "`AgentSession`（通过 `agent.create_session()` 创建）在会话中提供工作记忆。代理可以回顾之前的消息，同时查询 Cognee 的长期知识图谱。\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "session = coding_agent.create_session()\n", "\n", "response = await coding_agent.run(\n", " \"How does my AsyncWebScraper implementation align with Python's design principles?\",\n", " session=session,\n", ")\n", "print(\"🤖 Agent:\", response)" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "response = await coding_agent.run(\n", " \"Based on what you just said, when should I pick dataclasses versus Pydantic for this work?\",\n", " session=session,\n", ")\n", "print(\"🤖 Agent:\", response)\n", "print(\"\\n💡 The agent combined working memory (previous answer) with Cognee's knowledge graph.\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 新会话 — 长期记忆持续存在\n", "\n", "开始一个新的会话会清除工作记忆，但Cognee知识图谱仍然可用。代理可以在一个全新的对话中检索相同的长期知识。\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "session_2 = coding_agent.create_session()\n", "\n", "response = await coding_agent.run(\n", " \"What logging guidance should I follow for incident reviews?\",\n", " session=session_2,\n", ")\n", "print(\"🤖 Agent:\", response)\n", "print(\"\\n💡 New session, but the agent still has access to the full Cognee knowledge graph.\")" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "response = await coding_agent.run(\n", " \"How should variables be named according to Python best practices?\",\n", " session=session_2,\n", ")\n", "print(\"🤖 Agent:\", response)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 摘要\n", "\n", "在本笔记本中，您构建了一个结合了MAF工作记忆（`agent.create_session()`）与Cognee长期知识图谱的编码助手。\n", "\n", "### 您学到了什么\n", "1. 知识图谱构建：Cognee摄取非结构化文本并构建图谱 + 向量记忆。\n", "2. **利用memify丰富图谱**：在现有图谱基础上派生事实和更丰富的关系。\n", "3. **MAF + Cognee集成**：`@tool`函数让MAF代理自然查询Cognee的图谱。\n", "4. **工作记忆 + 长期记忆**：`AgentSession`（通过`agent.create_session()`）提供会话上下文，Cognee提供持久知识。\n", "5. **使用NodeSets的过滤搜索**：定位知识图中特定子集（例如，仅原则）。\n", "\n", "### 主要收获\n", "- **Cognee** 将原始文本转化为结构化、具备关系意识的记忆 —— 比单纯的向量存储更强大。\n", "- **`@tool`函数** 清晰桥接MAF代理与外部知识系统。\n", "- **`AgentSession`**（通过`agent.create_session()`）使每个对话上下文与长期知识分离保存。\n", "- 相同的知识图谱服务于多个会话和代理。\n", "\n", "### 现实应用\n", "- 开发者助手：代码审查、事件分析、架构助理\n", "- 面向客户的助手：基于产品文档、FAQ及CRM记录的支持代理\n", "- 内部专家助手：基于政策、法律或安全指南的推理助手\n", "- 统一数据层：将结构化与非结构化数据合成一个可查询图谱\n", "\n", "### 后续步骤\n", "- 在Cognee中尝试时间感知功能\n", "- 为特定领域图谱质量定义OWL本体\n", "- 添加用户反馈机制，提升检索效果\n", "- 扩展到共享同一Cognee记忆层的多代理系统\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n\n\n**免责声明**：\n本文件由 AI 翻译服务 [Co-op Translator](https://github.com/Azure/co-op-translator) 翻译完成。尽管我们力求准确，但请注意，自动翻译可能包含错误或不准确之处。原始语言版文件应视为权威来源。对于重要信息，建议使用专业人工翻译。我们对因使用本翻译而产生的任何误解或误释不承担责任。\n\n" ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.12.0" } }, "nbformat": 4, "nbformat_minor": 5 }