LangGraph 集成¶

简介¶

by-framework-langgraph 库提供了 by-framework 与 LangGraph 的深度集成，使得可以使用 LangGraph 定义复杂的 Agent 工作流。

安装¶

pip install by-framework-langgraph

核心组件¶

LangGraphWorker¶

from by_framework_langgraph import LangGraphWorker

class MyLangGraphAgent(LangGraphWorker):
    def get_agent_types(self):
        return ["langgraph_agent"]

    def create_graph(self, state: AgentState) -> CompiledGraph:
        # 返回编译后的 LangGraph
        return compiled_graph

使用示例¶

from by_framework import AgentContext, run_worker
from by_framework_langgraph import LangGraphWorker

from langgraph.graph import StateGraph, END
from typing import TypedDict, List

class AgentState(TypedDict):
    messages: List[str]

def node1(state: AgentState) -> AgentState:
    return {"messages": state["messages"] + ["Hello from node 1"]}

def node2(state: AgentState) -> AgentState:
    return {"messages": state["messages"] + ["Hello from node 2"]}

class MyLangGraphAgent(LangGraphWorker):
    def get_agent_types(self):
        return ["my_langgraph_agent"]

    def create_graph(self, state: AgentState) -> CompiledGraph:
        graph = StateGraph(AgentState)
        graph.add_node("node1", node1)
        graph.add_node("node2", node2)
        graph.set_entry_point("node1")
        graph.add_edge("node1", "node2")
        graph.add_edge("node2", END)
        return graph.compile()

run_worker(
    worker_class=MyLangGraphAgent,
    worker_id="langgraph-worker-01",
)

特性¶

流式输出: 支持 LangGraph 执行过程的流式输出
状态同步: AgentContext 与 LangGraph 状态自动同步
错误处理: 集成框架的错误处理机制
插件支持: 完全支持框架的插件系统