Projects

项目背景

1. 泰勒斯提出水是万物的本源

2. 泰勒斯的弟子有阿纳克西曼德

3. 阿纳克西曼德将没有任何形式规定性的阿派朗认定为万物的本源

1. Offline: 我们需要离线对语料库进行图索引的构建（将非结构化语料转化为结构化数据存储到图数据库中）
1. Online: 当GraphRAG系统接收到用户问题时，根据图数据库捕捉到的语料库中不同实体之间的关联关系，我们可以从图数据库中抓取到上面的三句话（具体图数据库索引可能如下图所示）

1. Graph Index的构建过程存在优化的可能，构建的GraphIndex会影响答案的质量。

2. GraphRAG Serving的过程中Token消耗巨大

3. GraphRAG中存在各种各样的图算法，如何Retrieval效果最好呢？（配置空间过大）

现有 Workflow：优雅的解耦，未竟的并行

1. Operator：表示「原子式的操作单元」，负责完成一个明确的子任务，如向量索引构建、向量相似度查询、图数据相关操作等等

2. Workflow：由Operator作为节点构成的链状执行流。项目中预定义好的Workflow和项目Demo用例一一对应（如GraphRAG， Vector-Similarity-Based RAG）

class Operator:
	@abstractmethod
	def run(context: dict[str, Any]) -> dict[str,Any]:
		return {}

打破链式束缚，拥抱全新架构

1. 并行性：Workflow中没有数据依赖关系的不同Task能否支持自动并行

2. 低耦合：Task的具体实现应该和Workflow本身解耦（用人话：一种Task可以作为多种不同的Workflow的节点，同时Task的实现是否需要包含其依赖关系约束？）

3. 数据共享：由于我们希望不同的Task之间的依赖关系解耦，那么我们需要Workflow粒度的数据共享机制用来在不同的Task共享数据（用于参数传递）

4. 提供Python接口

海底捞针，山穷水尽

LlamaIndex

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()

Agno

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 = 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
    ]
)

1. Agent

2. 由多个Agent构成的Team

3. Pure Function

Pydantic-Ai

LangGraph

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"}

总结

CGraph —— Graph with Python Interaface Implement in C++

架构设计

抽象设计是否合理？

1. 新增新的Operator，我们只需要修改对应Node的具体实现即可，不需要修改上层Workflow的逻辑
1. Operator对Node负责，Node对Workflow负责

2. 拥有更多的Workflow复用的机会

3. 通过引入新的Node抽象，我们在重构的过程中不需要修改底层Operator的实现，从某种程度上减轻了重构过程中的心智负担

1. 用户输入的状态（我们称之为WorkflowInput）：这部分由用户的请求构成

2. Workflow的中间状态（我们称之为WorkflowState）：这部分由Workflow中Node执行过程中产生

• WorkflowInput需要在Workflow执行前构建好，在Workflow执行之后就不再需要WorkflowInput了

• WorkflowState在Workflow执行过程中被逐渐写入，我们需要保证Workflow执行前是干净的

class BaseFlow(ABC):
    """
    Base class for flows, defines three interface methods: prepare, build_flow, and post_deal.
    """

    @abstractmethod
    def prepare(self, prepared_input: WkFlowInput, **kwargs):
        """
        根据用户请求初始化Workflow输入状态
        """

    @abstractmethod
    def build_flow(self, **kwargs) -> GPipeline:
        """
        用来构建可以运行在CGraph之上的Workflow对象
        """

    @abstractmethod
    def post_deal(self, **kwargs):
        """
        从中间状态中组装返回给用户的Response
        """

1. 如何将Node层和Operator层尽可能解耦

2. Node本身是可以并行的，而Workflow内部存在共享数据，那么如何解决可能存在的并发问题

class BaseNode(GNode):
		// Workflow中间状态
    context: Optional[WkFlowState] = None
    // Workflow输入状态
    wk_input: Optional[WkFlowInput] = None

    def init(self):
		    // 从Pipeline中获取对应状态
        return init_context(self)

    def node_init(self):
        """
        重写这个方法定制化Node初始化逻辑
        """
        return CStatus()

    def run(self):
        """
        Main logic for node execution, can be overridden by subclasses.
        Returns a CStatus object indicating whether execution succeeded.
        """
        sts = self.node_init()
        if sts.isErr():
            return sts
        if self.context is None:
            return CStatus(-1, "Context not initialized")
            
	      
        self.context.lock()
        try:
            data_json = self.context.to_json()
        finally:
            self.context.unlock()

        res = self.operator_schedule(data_json)
       
        self.context.lock()
        try:
            if res is not None and isinstance(res, dict):
                self.context.assign_from_json(res)
            elif res is not None:
                log.warning("operator_schedule returned non-dict type: %s", type(res))
        finally:
            self.context.unlock()
        return CStatus()

    def operator_schedule(self, data_json) -> Optional[Dict]:
        """
        根据用户请求或者Workflow状态决定调用哪些Operator
        """
        raise NotImplementedError("Subclasses must implement operator_schedule")

调度器的设计

Agentic GraphRAG的实现愿景

反思

结语