Semantica × Agno 集成实战:为多智能体团队接入知识图谱记忆与决策智能
2026/9/15 19:53:56 网站建设 项目流程

Semantica × Agno 集成实战:为多智能体团队接入知识图谱记忆与决策智能

【免费下载链接】semanticaGraph-Native Infrastructure for Context and Accountable AI Systems项目地址: https://gitcode.com/GitHub_Trending/sema/semantica

Semantica 提供面向上下文与可问责 AI 系统的图原生基础设施,而 Agno(agno >= 1.0)是广受欢迎的 Python 多智能体框架。integrations/agno提供了五个即插即用的组件,把 Semantica 的知识图谱(KG)、向量记忆与决策智能直接注入任意 Agno Agent 或 Team。读完本文,你将掌握如何用图结构记忆替换 Agno 的扁平存储、如何让 Agent 在推理循环中主动构建和查询上下文图,以及如何为多智能体团队共享同一张上下文图并按角色隔离写入。

集成总览:五个组件解决什么问题

Agno 原生的记忆与知识库是扁平化的:记忆以文本行存储,知识库以文档向量检索为主,缺少实体之间的关联关系,也难以沉淀"决策"这类结构化信息。Semantica × Agno 集成通过五个组件补齐了这一层:

组件Agno 侧接入点职责
AgnoContextStoreAgentMemory(db=…)实现agno.memory.db.base.MemoryDb,用"向量 + 上下文图"混合记忆替换扁平存储
AgnoKnowledgeGraphAgent(knowledge=…)实现agno.knowledge.base.AgentKnowledge,文档经完整抽取流水线进入可查询的ContextGraph,支持多跳 GraphRAG
AgnoDecisionKitAgent(tools=[…])6 个决策智能工具:记录决策、查找先例、追溯因果链、影响分析、策略校验、历史汇总
AgnoKGToolkitAgent(tools=[…])7 个知识图谱工具:抽取实体、抽取关系、写入图、查询图、查找关联、推理事实、导出子图
AgnoSharedContextTeam 级所有 Agent 共享一个ContextGraph,通过bind_agent()获得按角色隔离的视图,写入按角色打标签

所有类都遵循 Agno 官方扩展点的Toolkit/MemoryDb/AgentKnowledge模式,因此可以像使用 Agno 原生组件一样使用它们,无需改动 Agent 的其余代码。

安装

核心集成与可选后端通过pyproject.toml中定义的 extras 安装(源码中对应声明为agno = ["agno>=1.0.0"],见 pyproject.toml):

# 核心集成 pip install "semantica[agno]" # 搭配图存储后端 pip install "semantica[agno,graph-neo4j]" pip install "semantica[agno,graph-falkordb]" # 全栈:图存储 + 向量存储 pip install "semantica[agno,graph-neo4j,vectorstore-pgvector]"

集成模块版本为0.3.0(见 integrations/agno/init.py)。

组件一:AgnoContextStore —— 图结构化的 Agent 记忆

为什么需要它

Agno 默认的AgentMemory(db=…)是扁平的行存储。AgnoContextStore在进程内实现 Agno 的MemoryDb协议,底层由 Semantica 的AgentContext驱动(源码见 integrations/agno/context_store.py):每次upsert_memory都把文本写入向量索引并同步抽取实体、扩充知识图谱节点;每次read_memories则返回"向量相似 + 一跳图邻居扩展"的混合结果。

基本用法

from agno.agent import Agent from agno.memory import AgentMemory from agno.models.openai import OpenAIChat from semantica.context import ContextGraph from semantica.vector_store import VectorStore from integrations.agno import AgnoContextStore store = AgnoContextStore( vector_store=VectorStore(backend="faiss"), knowledge_graph=ContextGraph(advanced_analytics=True), decision_tracking=True, # 每次 upsert 自动记录一条轻量决策 graph_expansion=True, # read_memories 附带一跳图邻居 session_id="user_session_42", ) agent = Agent( model=OpenAIChat(id="gpt-4o"), memory=AgentMemory(db=store), description="A financially aware assistant with persistent decision intelligence.", )

参数与行为

参数默认值说明
vector_store自动创建VectorStore(backend="faiss")Semantica 向量存储实例
knowledge_graph自动创建ContextGraph()上下文图实例
decision_trackingTrue每次upsert_memory自动记录一条category="memory"的轻量决策条目
graph_expansionTrue召回记忆时用一跳图邻居增强结果
session_id自动生成 UUID图中节点的作用域标识

核心方法

AgnoContextStore完整实现了MemoryDb协议(createtable_existsmemory_existsread_memoriesupsert_memorydelete_memorydrop_tableclear),并在其之上扩展了 Semantica API:

方法说明
upsert_memory()将文本存入AgentContext(向量索引 + 图节点),并对文本做 NER,把抽取出的实体写入知识图谱
read_memories()混合检索:向量相似度 + 图跳扩展;支持按user_id过滤、按时间排序(默认新到旧)与limit截断
delete_memory()/drop_table()/clear()移除缓存并调用AgentContext.forget()
record_decision()记录带推理与结果的结构化决策,返回决策 ID
find_precedents()返回语义相似的历史决策(默认最多 5 条)
retrieve()混合检索入口,供应用层直接调用
get_context_for_prompt()把相关历史决策格式化为多行文本,用于注入系统提示词

其中upsert_memory的底层链路(见 context_store.py)为:AgentContext.store()持久化文本 →NERExtractor.extract_entities()抽取实体 →ContextGraph.add_node()写入图 →(若开启)record_decision()记录决策 → 更新进程内缓存。每条路径都有异常兜底,失败仅记日志、不影响主流程。

get_context_for_prompt(scenario, max_precedents=3)特别适合在每次调用 LLM 前自动召回相关先例——无先例时返回空字符串,不会污染提示词(见 context_store.py)。

组件二:AgnoKnowledgeGraph —— 基于知识图谱的 Agent 知识库

从扁平文档库到可查询图

AgnoKnowledgeGraph实现 Agno 的AgentKnowledge协议,让 Agent 面向结构化的ContextGraph而不是扁平文档库做检索。注入的文档会经过 Semantica 完整抽取流水线(源码见 integrations/agno/knowledge_graph.py):

Ingestion:parse → NER → relation extract → graph build → vector index

Search:vector retrieval → entity lookup → graph hop expansion → context injection

from agno.agent import Agent from agno.models.openai import OpenAIChat from semantica.kg import GraphBuilder from semantica.semantic_extract import NERExtractor, RelationExtractor from integrations.agno import AgnoKnowledgeGraph kg = AgnoKnowledgeGraph( graph_builder=GraphBuilder(), ner_extractor=NERExtractor(), relation_extractor=RelationExtractor(), ) kg.load("regulatory_docs/", recursive=True) # 目录递归注入 kg.load(texts=["Basel IV capital requirements apply from January 2026."]) # 直接注入文本 agent = Agent(model=OpenAIChat(id="gpt-4o"), knowledge=kg, search_knowledge=True)

参数说明

参数默认值说明
graph_builder自动创建GraphBuilder()负责把"文本 + 实体 + 关系"构建成图,其graph_store在初始化时被自动接到内部ContextGraph
ner_extractor自动创建NERExtractor()命名实体识别器
relation_extractor自动创建RelationExtractor()关系抽取器
context_graph自动创建ContextGraph()外部传入的已有图;传入后与向量检索共享同一张图
graph_store_backend"inmemory"图后端:"inmemory""neo4j""falkordb"
graph_store_uriNone所选图后端的连接 URI
num_documents5search()默认返回的文档数
chunk_size1000注入时的最大文本分块字符数(按段落边界切分)

加载方式与检索

load()支持三种来源(见 knowledge_graph.py):

  • path:文件、目录或 glob 模式;目录配合recursive=True递归遍历,逐文件以 UTF-8 读取;
  • texts:原始字符串列表,来源标记为<inline>
  • urls:通过load_urls()抓取,复用semantica.ingest.ssrf.request_with_ssrf_guard做 SSRF 防护——仅允许http/https协议,默认拦截私有/回环/链路本地/云元数据地址,校验 DNS 解析并逐跳复查重定向(见 knowledge_graph.py)。

search(query, num_documents=None, filters=None)执行多跳 GraphRAG:先用AgentContext.retrieve()做向量检索,再对结果中的实体做一跳图邻居扩展,把[Graph context]文本块附加到文档内容后返回 AgnoDocument对象;当向量检索不可用时,自动降级为对进程内文档缓存的关键词打分(见 knowledge_graph.py)。

此外,get_graph_context(entity)返回某个实体一跳子图的文本表示(邻居节点与边类型),适合结构化推理场景:

ctx = kg.get_graph_context("Basel IV") # Entity: Basel IV # --[applies_to]--> EU Banks (type: Organization) # --[covers]--> Capital Requirements

内部注入链路(_ingest_text)为:段落分块 → NER → 关系抽取 →GraphBuilder.build()落图 →AgentContext.store()入向量索引 → 缓存文档(见 knowledge_graph.py)。

组件三:AgnoDecisionKit —— 决策智能工具集

AgnoDecisionKit把 Semantica 的决策智能以 AgnoToolkit的形式暴露给 Agent,让 Agent 在推理循环中主动记录、查询与校验决策(源码见 integrations/agno/decision_kit.py):

from agno.agent import Agent from agno.models.openai import OpenAIChat from semantica.context import AgentContext from integrations.agno import AgnoDecisionKit ctx = AgentContext(decision_tracking=True) agent = Agent( model=OpenAIChat(id="gpt-4o"), tools=[AgnoDecisionKit(context=ctx)], show_tool_calls=True, ) agent.print_response("Should we approve this mortgage application?")

构造参数:contextAgentContext或兼容对象,缺省自动创建)、max_precedents=5find_precedents默认返回数)、causal_depth=3trace_causal_chain默认链深)、enable_policy_check=True(是否注册check_policy工具)。

六个工具

工具说明关键实现细节
record_decision记录决策及其推理、结果与置信度返回{"decision_id": ..., "status": "recorded"}JSON;entities支持逗号分隔列表
find_precedents按场景搜索相似历史决策调用find_precedents_advanced,归一化为可序列化列表返回
trace_causal_chain追溯决策的因果链调用图的trace_decision_causality(decision_id, depth=…),缺该 API 时降级为按类别找先例
analyze_impact用图中心性评估决策的下游影响调用AgentContext.analyze_decision_influence()
check_policy校验决策是否符合策略规则支持>= <= != == > <六种比较运算符;不可求值的规则进入warnings而非violations
get_decision_summary按类别汇总决策历史基于AgentContext.get_context_insights()

check_policy 的规则语法

check_policy(decision_data, policy_rules)policy_rules是 JSON 规则字符串列表,例如:

["confidence >= 0.7", "category != \"test\""]

每条规则形如<field> <op> <value>。源码对边界情况处理得非常细致(见 decision_kit.py):decision_data必须解码为 JSON 对象,否则直接记为违规;引用了缺失字段、字段值为null或格式无法识别的规则,会进入warnings(避免静默判为合规);裸字符串policy_rules='"confidence >= 0.7"'视为单条规则而非逐字符遍历。

组件四:AgnoKGToolkit —— 让 Agent 主动构建与查询知识图谱

AgnoKGToolkit让 Agent 在推理过程中直接操作上下文图:既可以从文本中抽取实体/关系写入图,也可以查询、推理与导出子图(源码见 integrations/agno/kg_toolkit.py):

from agno.agent import Agent from agno.models.openai import OpenAIChat from integrations.agno import AgnoKGToolkit agent = Agent( model=OpenAIChat(id="gpt-4o"), tools=[AgnoKGToolkit()], show_tool_calls=True, )

构造参数:graph_store_backend"inmemory"/"neo4j"/"falkordb")、ner_extractorrelation_extractorreasoner(自动创建semantica.reasoning.Reasoner)、context(可传入已有AgentContextContextGraph复用)。

七个工具

工具说明返回格式
extract_entities从文本抽取命名实体{"entities": [{"name", "type", "confidence"}], "count"}
extract_relations抽取实体间关系,可按实体列表限定{"relations": [{"source", "relation", "target", "confidence"}], "count"}
add_to_graph把实体/关系写入上下文图{"nodes_added", "edges_added"}
query_graph自然语言关键词查询;以MATCH开头走 Cypher(需 Neo4j/FalkorDB 后端){"results", "query_type"}
find_related沿图跳(默认 1 跳)查找相关概念{"entity", "related", "count"}
infer_facts用规则在图上演推新事实{"inferred_facts", "count"}
export_subgraph导出子图为 RDF(json-ld/turtle/ttl/xml/nt{"format", "data"}

其中infer_facts的规则形如"IF Person(?x) THEN Human(?x)",未显式提供事实时自动从图中取前 50 个节点生成Type(label)形式的事实集(见 kg_toolkit.py);export_subgraph底层复用semantica.export.RDFExporter,导出失败时兜底返回图节点 JSON。

组件五:AgnoSharedContext —— 多智能体团队共享上下文

在 AgnoTeam中,多个 Agent 往往需要共享同一份知识与决策历史,同时又不能互相污染写入。AgnoSharedContext用一个ContextGraph贯穿整个团队,每个 Agent 通过bind_agent(role)获得一个按角色命名空间隔离的AgnoContextStore,所有写入都带角色标签(源码见 integrations/agno/shared_context.py):

from agno.agent import Agent from agno.team import Team from agno.models.openai import OpenAIChat from semantica.context import ContextGraph from semantica.vector_store import VectorStore from integrations.agno import AgnoSharedContext, AgnoDecisionKit, AgnoKGToolkit shared = AgnoSharedContext( vector_store=VectorStore(backend="faiss"), knowledge_graph=ContextGraph(advanced_analytics=True), decision_tracking=True, ) research_agent = Agent( name="Researcher", model=OpenAIChat(id="gpt-4o"), memory=shared.bind_agent("researcher"), tools=[AgnoKGToolkit(context=shared)], ) decision_agent = Agent( name="Analyst", model=OpenAIChat(id="gpt-4o"), memory=shared.bind_agent("analyst"), tools=[AgnoDecisionKit(context=shared)], ) team = Team( name="Research & Decision Team", agents=[research_agent, decision_agent], mode="coordinate", )

团队级 API

decision_id = shared.record_decision( category="strategy", scenario="Expand to EU market", reasoning="Strong demand signals from Q1 survey", outcome="approved", confidence=0.87, agent_role="cfo", # 可选:按角色给决策打标签 ) precedents = shared.find_precedents("market expansion") insights = shared.get_shared_insights()

核心机制(见 shared_context.py):

  • 角色隔离bind_agent(role)幂等——同一role多次调用返回同一个_AgentScopedStore实例;其session_id{team_session}::{role},记忆以category=f"memory:{role}"写入,决策以f"{category}:{agent_role}"打标签。
  • 共享可见:每个角色的read_memories()返回"自己的记忆 + 团队共享记忆池"的并集(_shared_memories),保证 Researcher 的记忆对 Analyst 可见。
  • 线程安全:内部用threading.RLock保护bind_agentrecord_decision,支持多 Agent 并发访问同一上下文(对应测试见 tests/integrations/agno/test_shared_context.py,其中包含并发绑定同一角色、角色标签、记忆跨角色可见等用例)。

API 参考与优雅降级

from integrations.agno import ( AgnoContextStore, # MemoryDb implementation AgnoKnowledgeGraph, # AgentKnowledge implementation AgnoDecisionKit, # Decision intelligence Toolkit AgnoKGToolkit, # Knowledge graph Toolkit AgnoSharedContext, # Team-level shared context AGNO_AVAILABLE, # bool: True if agno is installed )

五个类均可在未安装agno的情况下导入并使用:它们携带完整的 Semantica API,并在agno缺席时提供轻量替代类型(如_MemoryRow_Document),仅在真正传给 Agno 的Agent/Team构造器时需要agno环境。每个模块都通过try/except ImportError设置AGNO_AVAILABLE标志,Toolkit注册逻辑(self.register())与_tools列表解耦,保证无论agno是否安装,_tools都可被应用层检视(见 decision_kit.py)。测试套件也专门使用 stub 模块模拟agno来验证这一点(见 tests/integrations/agno/test_knowledge_graph.py)。

延伸阅读

  • Context Module —— 支撑本集成的AgentContextContextGraph
  • Knowledge Graph ——AgnoKnowledgeGraph使用的 KG 构建能力
  • LLMs —— 为 Agno Agent 配置 LLM 提供商
  • Vector Store ——AgnoContextStore的向量后端
  • Agno 决策智能 Notebook 与 Agno GraphRAG 上下文 Notebook、多智能体共享上下文 Notebook —— 完整的可运行示例
  • 集成模块测试用例:tests/integrations/agno/(覆盖 MemoryDb 协议、KG 注入、决策工具、共享上下文与 SSRF 防护)

【免费下载链接】semanticaGraph-Native Infrastructure for Context and Accountable AI Systems项目地址: https://gitcode.com/GitHub_Trending/sema/semantica

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询