imbajin commented on code in PR #424:
URL:
https://github.com/apache/incubator-hugegraph-doc/pull/424#discussion_r2480923667
##########
content/cn/blog/hugegraph-ai/agentic_graphrag.md:
##########
@@ -0,0 +1,449 @@
+---
+date: 2025-10-29
+title: "Agentic GraphRAG"
+linkTitle: "Agentic GraphRAG"
+---
+
+# 项目背景
+
+为了应对模型训练数据和现实生活中实际数据之间存在的时效性差异问题，RAG技术应运而生。RAG，顾名思义就是通过向外部数据源获取对应的数据（Retrieval），用于增强（Argument）大模型生成（Generation）回答质量的技术。
+
+最早的RAG采用简单的Retrieval -
Generation架构，我们拿到用户给出的问题，进行一定的预处理（关键词提取等等），得到预处理之后的问题，接着通过Embedding
Model从海量资料中抓取相关的资料作为Prompt交给大模型用于增强模型回答的质量。
+
+但是基于语义相似性匹配进行相关语料的抓取未必能够处理所有情况，因为能够用于增强回答质量的语料不一定和问题本身存在语义相似性。一个常见的例子就是：**告诉我“提出水是万物的本源”的哲学家的徒弟提出的本体论观点。**而我们的语料中并不直接存在这个问题的答案，语料库中可能提到：
+
+1. 泰勒斯提出水是万物的本源
+2. 泰勒斯的弟子有阿纳克西曼德
+3. 阿纳克西曼德将没有任何形式规定性的阿派朗认定为万物的本源
+
+如果单纯从语义相似度匹配出发，我们大概率只能retrieval到第一个句子用于增强大模型的回答。但是缺失语料2和语料3的情况下，如果我们所使用的大模型训练语料中没有哲学相关知识，在缺失这些关键信息的情况下，大模型将无法正确回答这些问题，甚至会出现“幻觉”。
+
+因此GraphRAG技术诞生了，常见的GraphRAG包含两个步骤：
+
+1. Offline: 我们需要离线对语料库进行图索引的构建（将非结构化语料转化为结构化数据存储到图数据库中）
+2. Online:
当GraphRAG系统接收到用户问题时，根据图数据库捕捉到的语料库中不同实体之间的关联关系，我们可以从图数据库中抓取到上面的三句话（具体图数据库索引可能如下图所示）
+
+<div style="text-align: center;">
+ <img src="/blog/images/images-server/agentic-background.png" alt="image"
width="400">
+</div>
+
+但是GraphRAG本身也存在几个问题：
+
+1. 如何构建Graph Index是一门学问，Graph Index会影响到模型回答质量。
+2. GraphRAG索引构建过程Token消耗巨大
+3. GraphRAG中存在各种各样的图算法，如何Retrieval效果最好呢？（配置空间过大）
+
+本次项目主要针对第三个问题展开。我们希望借助大模型的泛化能力使其自动识别用户问题中的意图，然后选择合适配置（比如选择最合适的图算法）从图数据库中读取对应的数据用于增强大模型回答质量——也就是本次项目Agentic
GraphRAG的目的所在。
+
+# **现有 Workflow：优雅的解耦，未竟的并行**
+
+现在的HugeGraph-AI项目中存在两个核心抽象：
+
+1. Operator：表示「原子式的操作单元」，负责完成一个明确的子任务，如向量索引构建、向量相似度查询、图数据相关操作等等
+2. Workflow：由Operator作为节点构成的**链状**执行流。项目中预定义好的Workflow和项目Demo用例一一对应（如GraphRAG，
Vector-Similarity-Based RAG）
+
+由于Operator的实现需要遵循下面的接口:
+
+```python
+class Operator:
+ @abstractmethod
+ def run(context: dict[str, Any]) -> dict[str,Any]:
+ return {}
+```
+
+Operator在实际运行时接受字典类型的context对象作为输入，返回的对象也是一个字典，可以用来作为下一个Operator的输入，这样的设计有一个很高明的地方——他将不同的Operator之间的依赖关系和Operator本身的具体实现解耦了，每个Operator是一个相对独立的存在，如果Operator
A需要依赖Operator B的输出，那么只需要检查context对象中是否存有Operator
B的输出即可。这是一种低耦合的设计。好处是我们能很方便地将不同的Operator自由组合。根据不同的用户输入组装（配置）合适Workflow
Serving用户请求，那不正是我们在项目背景中提到的Agentic GraphRAG的目的所在吗？
+
+```
+👉🏼 理论上现有设计已经可以正常过渡到Agentic GraphRAG，但是现有设计存在诸多悬而未决的问题：
+ 1. 现有调度器仅仅支持链状Workflow，缺失了可能存在的并行空间
+ 2. 现有调度器无法复用被反复使用到的Workflow
+```
+
+# 打破链式束缚，拥抱全新架构
+
+之前的调度器给我们的启发是Operator粒度的解耦是一个不错的设计理念，但是调度器本身能力有限，限制了Workflow的能力。因此我们计划替换项目中的调度器！经过对几种不同的Workflow编排框架进行简单的调研之后，我们认为下面几个特性是我们筛选调度器的标准。（下面我们统一将框架编排对象称为Workflow，Workflow由一系列Task组成）
+
+1. 并行性：Workflow中没有数据依赖关系的不同Task能否支持自动并行
+2.
低耦合：Task的具体实现应该和Workflow本身解耦（用人话：一种Task可以作为多种不同的Workflow的节点，同时Task的实现是否需要包含与其他Task依赖关系约束？）
+3. 数据共享：由于我们希望不同的Task之间的依赖关系解耦，那我们就需要Workflow粒度的数据共享机制用来在不同的Task共享数据（用于参数传递）
+4. 提供Python接口
+
+## AI框架大乱斗
+
+我们首先将目光放到了现在炙手可热的AI
Workflow调度框架。围绕前面提到的几个维度，我们分别调研了下面几种不同的Workflow编排框架——LlamaIndex，Agno，Pydantic-Ai，LangGraph。
+
+### LlamaIndex
+
+对于LlamaIndex，我们用一个常见的例子说明LlamaIndex这个框架的设计理念。
+
+```python
+from workflows import Workflow, Context, step
+from workflows.events import StartEvent, StopEvent, Event
+
+class StepEvent(Event):
+ message: str
+
+class MyWorkflow(Workflow):
+
+ @step
+ async def step_one(self, ctx: Context, ev: StartEvent) -> StepEvent:
+ current_count = await ctx.store.get("count", default=0)
+ current_count += 1
+ await ctx.store.set("count", current_count)
+ print("step one called once")
+ return StepEvent("launch step two")
+
+ @step
+ async def step_two(self, ctx: Context, ev: StepEvent) -> StopEvent:
+ print("step two called once")
+ return StopEvent()
+```
+
+从上面这个简单的例子我们可以看到很多问题。首先明确一个观念：Workflow由两个元素构成：Task，Task之间的依赖关系。只要这两个元素确定之后一个Workflow就确定下来了。我们可以看到LlamaIndex中每个Task(对应代码中用@step注解的函数)的实现和Workflow存在依赖关系。因为每个Task的实现都需要传入Event对象作为参数，但是Event参数其实就是对Task之间依赖关系的一种限定。所以LlamaIndex不具备低耦合的特点。同时我们也发现Task作为Workflow类成员函数本身就违背了我们前面提到的Task需要能够在多种不同Workflow中使用的诉求。但是经过调研，LlamaIndex的数据共享和并行特性支持还算不错。只不过从基于事件驱动模型构建的编程接口在保证了接口易用性的同时也牺牲了编程的灵活性。
+
+### Agno
+
+同样还是从例子入手
+
+```python
+from agno.workflow import Router, Step, Workflow
+
+def route_by_topic(step_input) -> List[Step]:
+ topic = step_input.input.lower()
+
+ if "tech" in topic:
+ return [Step(name="Tech Research", agent=tech_expert)]
+ elif "business" in topic:
+ return [Step(name="Business Research", agent=biz_expert)]
+ else:
+ return [Step(name="General Research", agent=generalist)]
+
+workflow = Workflow(
+ name="Expert Routing",
+ steps=[
+ Router(
+ name="Topic Router",
+ selector=route_by_topic,
+ choices=[tech_step, business_step, general_step]
+ ),
+ Step(name="Synthesis", agent=synthesizer),
+ ]
+)
+
+workflow.print_response("Latest developments in artificial intelligence and
machine learning", markdown=True)
+```
+
+从这个例子我们可以看到Workflow本身和Task之间的绑定关系是通过指定steps参数确定的。理论上来说定义好一种Task之后我们可以将其用于不同的Workflow中，Agno的设计符合我们的低耦合标准。
+
+但是数据共享和任务并行方面的支持就存在一定的限制。
+
+首先是任务并行，例子如下：
+
+```python
+workflow = Workflow(
+ name="Parallel Research Pipeline",
+ steps=[
+ Parallel(
+ Step(name="HackerNews Research", agent=hn_researcher),
+ Step(name="Web Research", agent=web_researcher),
+ Step(name="Academic Research", agent=academic_researcher),
+ name="Research Step"
+ ),
+ Step(name="Synthesis", agent=synthesizer), # Combines the results and
produces a report
+ ]
+)
+```
+
+Agno专门设计了并行接口，我们需要在静态编译时（Python哪有编译时？应该叫写代码的时候哈哈😀）明确哪些任务可以并行。但是Agentic
GraphRAG最终构造的Workflow有可能是在运行时由模型规划出来的，是动态运行时明确的，出于这样的考量，我们认为Agno的并行特性并不符合我们的要求
+
+接下来是数据共享，Agno框架中支持三种不同的Task：
+
+1. Agent
+2. 由多个Agent构成的Team
+3. Pure Function
+
+我们检查了调研时最新版本的Agno源代码，发现Agno支持的状态共享仅限于Agent和Team。那么对于那些适合用Pure
Function实现的Task，我们就需要额外支持数据共享的机制。因此Agno的数据共享机制也不符合我们的要求。
+
+### Pydantic-Ai
+
+我们从官方文档中就看到
+
+<div style="text-align: center;">
+ <img src="/blog/images/images-server/agentic-pydantic.png" alt="image"
width="800">
+</div>
+
+Pydantic-Ai框架竟然不支持Task粒度的自动并行。
+
+和LlamaIndex框架类似采用事件驱动的编程模型，因此Workflow和Task之间不算是完全解耦，但是值得注意的时Pydantic-Ai的Task是可以用到多个不同的Workflow的。
+
+### LangGraph
+
+最后的最后，终于还是遇到了LangGraph，之前一直没有调研LangGraph的原因是因为由团队伙伴认为LangGraph本身太重了。在上一个版本中，即使只是使用LangGraph的部分功能（调度），也需要引入LangGraph的完整依赖，引入LangGraph可能会让项目变“重”。时不时在其他开源项目中看到“xxx比LangGraph快xxx倍”诸如此类的字眼也确实影响到我们的决策判断。所以直到此时此刻才把它提上调研日程。
+
+我们还是来看看LangGraph的例子
+
+```python
+class State(TypedDict):
+ topic: str
+ joke: str
+ improved_joke: str
+
+# Nodes
+def generate_joke(state: State):
+ """First LLM call to generate initial joke"""
+
+ msg = llm.invoke(f"Write a short joke about {state['topic']}")
+ return {"joke": msg.content}
+
+def check_punchline(state: State):
+ """Gate function to check if the joke has a punchline"""
+
+ # Simple check - does the joke contain "?" or "!"
+ if "?" in state["joke"] or "!" in state["joke"]:
+ return "Pass"
+ return "Fail"
+
+def improve_joke(state: State):
+ """Second LLM call to improve the joke"""
+
+ msg = llm.invoke(f"Make this joke funnier by adding wordplay:
{state['joke']}")
+ return {"improved_joke": msg.content}
+
+# Build workflow
+workflow = StateGraph(State)
+
+# Add nodes
+workflow.add_node("generate_joke", generate_joke)
+workflow.add_node("improve_joke", improve_joke)
+
+# Add edges to connect nodes
+workflow.add_edge(START, "generate_joke")
+workflow.add_conditional_edges(
+ "generate_joke", check_punchline, {"Fail": "improve_joke", "Pass": END}
+)
+workflow.add_edge("improve_joke", END)
+
+# Compile
+chain = workflow.compile()
+
+# Invoke
+state = chain.invoke({"topic": "cats"}
+```
+
+这是一个我简化后的官方文档中的例子，我们可以看到基于GraphAPI的LangGraph通过调用workflow.add_edge指定Workflow的依赖关系，将Workflow和Task解耦。同时支持全局State作为Workflow的状态进行Task之间的数据共享。根据官方文档的说法，LangGraph是支持Task自动并行执行的。我们总算是找到了符合所有要求的Workflow编排框架了！
+
+### 总结
+
+| | 并行性 | 低耦合 | 数据共享 | Python Interface |
+| --- | --- | --- | --- | --- |
+| LlamaIndex | 支持 | 不支持 | 支持 | 支持 |
+| Agno | 支持但不符合要求 | 支持 | 支持但不符合要求 | 支持 |
+| Pydantic-Ai | 不支持 | 不支持 | 支持 | 支持 |
+| LangGraph | 支持 | 支持 | 支持 | 支持 |
+
+## CGraph —— Graph with Python Interaface Implement in C++
Review Comment:
```suggestion
```text
👉🏼 给定一个用户请求,我们怎么通过用户的请求推断出最优配置呢?
```
💡 建议为代码块添加语言标识符 `text` 以改善渲染效果
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
