深入LangGraph：多智能体应用框架架构解析

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

{
  "graphs": {
    "deep_research": "./src/workflow.py:graph",
    "podcast_generation": "./src/podcast/graph/builder.py:workflow", 
    "ppt_generation": "./src/ppt/graph/builder.py:workflow",
    "multi_agent_system": "./src/agents/orchestrator.py:coordinator_graph"
  },
  "python_version": "3.12",
  "env": "./.env",
  "dependencies": ["."],
  "dockerfile_lines": [
    "RUN apt-get update && apt-get install -y git",
    "RUN pip install --no-cache-dir torch torchvision"
  ],
  "deployment": {
    "platform": "langgraph-cloud",
    "scaling": {
      "min_instances": 2,
      "max_instances": 50,
      "auto_scaling_metric": "cpu_utilization",
      "target_utilization": 70
    },
    "monitoring": {
      "enable_tracing": true,
      "log_level": "INFO",
      "metrics_collection": ["latency", "throughput", "error_rate"]
    }
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

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

class State(TypedDict):
    messages: list[str]
    result: str

# 创建状态图
graph = StateGraph(State)

# 添加节点
def process_node(state: State) -> State:
    """处理节点：执行核心业务逻辑"""
    messages = state["messages"]
    result = f"Processed: {', '.join(messages)}"
    return {"result": result}

graph.add_node("process", process_node)

# 设置入口和出口
graph.set_entry_point("process")
graph.set_finish_point("process")

# 编译图
app = graph.compile()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

class Pregel:
    """Pregel执行引擎
    
    基于消息传递的图计算模型，支持：
    - 超步（Super-step）执行
    - 并行节点处理
    - 状态同步机制
    """
    
    def __init__(self, nodes, channels, input_channels, output_channels):
        self.nodes = nodes          # 节点映射
        self.channels = channels    # 通道映射
        self.input_channels = input_channels    # 输入通道
        self.output_channels = output_channels  # 输出通道
        
    async def astream(self, input_data, config=None):
        """异步流式执行"""
        # 初始化状态
        checkpoint = self._prepare_checkpoint(input_data)
        
        # 执行超步循环
        while True:
            # 1. 计划阶段：确定活跃节点
            active_nodes = self._plan_step(checkpoint)
            if not active_nodes:
                break
                
            # 2. 执行阶段：并行执行节点
            results = await self._execute_step(active_nodes, checkpoint)
            
            # 3. 更新阶段：同步状态更新
            checkpoint = self._update_step(results, checkpoint)
            
            yield checkpoint

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

from abc import ABC, abstractmethod

class BaseChannel(ABC):
    """通道基类"""
    
    @abstractmethod
    def update(self, updates: Sequence[Any]) -> None:
        """更新通道值"""
        pass
    
    @abstractmethod
    def get(self) -> Any:
        """获取通道值"""
        pass
    
    @abstractmethod
    def checkpoint(self) -> Any:
        """创建检查点"""
        pass

class LastValue(BaseChannel):
    """最后值通道：保存最后一次更新的值"""
    
    def __init__(self, typ: type):
        self.typ = typ
        self.value = None
    
    def update(self, updates: Sequence[Any]) -> None:
        if updates:
            self.value = updates[-1]
    
    def get(self) -> Any:
        return self.value

class Topic(BaseChannel):
    """主题通道：支持消息累积和去重"""
    
    def __init__(self, typ: type, accumulate: bool = False, unique: bool = False):
        self.typ = typ
        self.accumulate = accumulate
        self.unique = unique
        self.values = []
    
    def update(self, updates: Sequence[Any]) -> None:
        for update in updates:
            if self.unique and update in self.values:
                continue
            self.values.append(update)
    
    def get(self) -> Any:
        return self.values if self.accumulate else (self.values[-1] if self.values else None)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

from abc import ABC, abstractmethod
from typing import Optional, AsyncIterator

class BaseCheckpointSaver(ABC):
    """检查点保存器基类"""
    
    @abstractmethod
    async def aput(
        self, 
        config: RunnableConfig, 
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata
    ) -> RunnableConfig:
        """保存检查点"""
        pass
    
    @abstractmethod
    async def aget(
        self, 
        config: RunnableConfig
    ) -> Optional[CheckpointTuple]:
        """获取检查点"""
        pass
    
    @abstractmethod
    async def alist(
        self,
        config: RunnableConfig,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> AsyncIterator[CheckpointTuple]:
        """列举检查点历史"""
        pass

# PostgreSQL实现示例
class PostgresCheckpointSaver(BaseCheckpointSaver):
    """PostgreSQL检查点保存器"""
    
    def __init__(self, connection_string: str):
        self.conn_string = connection_string
        
    async def aput(self, config, checkpoint, metadata):
        """将检查点保存到PostgreSQL"""
        async with asyncpg.connect(self.conn_string) as conn:
            await conn.execute("""
                INSERT INTO checkpoints (thread_id, checkpoint_id, data, metadata)
                VALUES ($1, $2, $3, $4)
            """, 
            config["configurable"]["thread_id"],
            config["configurable"]["checkpoint_id"],
            pickle.dumps(checkpoint),
            json.dumps(metadata)
            )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

class StateManager:
    """状态管理器：协调状态在节点间的传播"""
    
    def __init__(self, channels: dict[str, BaseChannel]):
        self.channels = channels
        self.pending_updates = defaultdict(list)
    
    def apply_writes(self, writes: list[tuple[str, Any]]):
        """应用写操作到通道"""
        for channel_key, value in writes:
            if channel_key in self.channels:
                self.pending_updates[channel_key].append(value)
    
    def commit_updates(self):
        """提交所有待处理的更新"""
        for channel_key, updates in self.pending_updates.items():
            channel = self.channels[channel_key]
            channel.update(updates)
        self.pending_updates.clear()
    
    def read_state(self, node_inputs: list[str]) -> dict[str, Any]:
        """读取节点所需的状态"""
        state = {}
        for input_key in node_inputs:
            if input_key in self.channels:
                state[input_key] = self.channels[input_key].get()
        return state

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

class ParallelExecutor:
    """并行执行器：管理节点的并行执行"""
    
    async def execute_step(self, active_nodes: list[PregelNode]) -> dict[str, Any]:
        """并行执行一个超步中的所有活跃节点"""
        tasks = []
        
        for node in active_nodes:
            # 为每个节点创建异步任务
            task = asyncio.create_task(
                self._execute_node(node),
                name=f"node-{node.name}"
            )
            tasks.append((node.name, task))
        
        # 等待所有任务完成或超时
        results = {}
        try:
            for node_name, task in tasks:
                result = await asyncio.wait_for(task, timeout=self.timeout)
                results[node_name] = result
        except asyncio.TimeoutError:
            # 取消未完成的任务
            for _, task in tasks:
                if not task.done():
                    task.cancel()
            raise
        
        return results
    
    async def _execute_node(self, node: PregelNode) -> Any:
        """执行单个节点"""
        try:
            # 读取节点输入
            input_state = self.state_manager.read_state(node.input_channels)
            
            # 调用节点函数
            if asyncio.iscoroutinefunction(node.func):
                result = await node.func(input_state)
            else:
                result = node.func(input_state)
            
            return result
            
        except Exception as e:
            # 错误处理和传播
            return PregelTaskError(node.name, e)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

from langgraph.types import Command, Send

def router(state: dict) -> Command[str]:
    """基于状态的路由：返回下一个要执行的节点，并可同时更新状态"""
    if state.get("needs_more_info"):
        return Command(goto="search", update={"query": state["query"]})
    else:
        return Command(goto="answer", update={"finalize": True})

def dispatcher(state: dict) -> list[Send]:
    """并行分发：将一批子任务以Send形式分发到多个节点实例"""
    tasks = state.get("subtasks", [])
    return [
        Send("worker", {"task": t, "task_id": i})
        for i, t in enumerate(tasks)
    ]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75

class ReflectiveAgent:
    """反思式智能体：支持自我修正和迭代优化"""
    
    def __init__(self, model, max_reflections: int = 3):
        self.model = model
        self.max_reflections = max_reflections
        
    def generate_with_reflection(self, state: State) -> State:
        """带反思机制的生成过程"""
        attempt = 0
        current_result = None
        
        while attempt < self.max_reflections:
            if attempt == 0:
                # 初次生成
                current_result = self._generate_initial(state)
            else:
                # 基于前次结果和错误进行反思
                current_result = self._reflect_and_improve(
                    state, current_result, error_feedback
                )
            
            # 验证结果
            validation_result = self._validate_result(current_result)
            if validation_result.success:
                break
            
            error_feedback = validation_result.error_details
            attempt += 1
        
        return {"result": current_result, "iterations": attempt + 1}
    
    def _validate_result(self, result) -> ValidationResult:
        """验证生成结果，支持代码执行、逻辑检查等"""
        try:
            # 执行代码验证（如果是代码生成任务）
            if self._is_code_generation_task():
                exec_result = self._execute_code_safely(result)
                return ValidationResult(success=True, result=exec_result)
            
            # 其他类型的验证逻辑
            return self._generic_validation(result)
            
        except Exception as e:
            return ValidationResult(
                success=False, 
                error_details=str(e),
                stack_trace=traceback.format_exc()
            )

# 反思式代码生成的完整流程
def create_reflective_code_agent():
    """创建反思式代码生成智能体"""
    graph = StateGraph(CodeGenState)
    
    # 添加生成节点
    graph.add_node("generate", code_generation_node)
    
    # 添加验证节点
    graph.add_node("validate", code_validation_node)
    
    # 添加反思节点
    graph.add_node("reflect", reflection_node)
    
    # 设置条件边
    graph.add_conditional_edges(
        "validate",
        lambda state: "reflect" if state.get("has_errors") else END,
        {"reflect": "reflect", END: END}
    )
    
    graph.add_edge("generate", "validate")
    graph.add_edge("reflect", "generate")
    
    return graph.compile()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

class MultiAgentOrchestrator:
    """多智能体编排器：协调多个专业智能体的协作"""
    
    def __init__(self):
        self.agents = {}
        self.coordination_graph = StateGraph(MultiAgentState)
    
    def create_hierarchical_system(self):
        """创建分层多智能体系统"""
        # 管理者智能体：负责任务分解和结果整合
        manager_agent = self._create_manager_agent()
        self.coordination_graph.add_node("manager", manager_agent)
        
        # 专业智能体：负责特定领域任务
        specialist_agents = {
            "data_analyst": self._create_data_analyst(),
            "code_generator": self._create_code_generator(),
            "qa_specialist": self._create_qa_specialist(),
        }
        
        for name, agent in specialist_agents.items():
            self.coordination_graph.add_node(name, agent)
        
        # 建立协作关系
        self.coordination_graph.add_conditional_edges(
            "manager",
            self._route_to_specialist,
            {**specialist_agents, END: END}
        )
        
        # 专业智能体完成后返回管理者
        for specialist in specialist_agents:
            self.coordination_graph.add_edge(specialist, "manager")
        
        return self.coordination_graph.compile()
    
    def _route_to_specialist(self, state: MultiAgentState) -> str:
        """智能路由：根据任务类型选择合适的专业智能体"""
        task_type = self._classify_task(state["current_task"])
        
        routing_map = {
            "data_analysis": "data_analyst",
            "code_generation": "code_generator", 
            "quality_assurance": "qa_specialist",
        }
        
        return routing_map.get(task_type, END)
    
    def _create_manager_agent(self):
        """创建管理者智能体"""
        def manager_node(state: MultiAgentState) -> dict:
            messages = state["messages"]
            current_task = state.get("current_task")
            
            if not current_task:
                # 初始任务分解
                task_breakdown = self._decompose_task(messages[-1].content)
                return {
                    "current_task": task_breakdown[0],
                    "remaining_tasks": task_breakdown[1:],
                    "task_results": [],
                }
            else:
                # 处理专业智能体的结果
                if "specialist_result" in state:
                    results = state.get("task_results", [])
                    results.append(state["specialist_result"])
                    
                    remaining = state.get("remaining_tasks", [])
                    if remaining:
                        # 还有待处理任务
                        return {
                            "current_task": remaining[0],
                            "remaining_tasks": remaining[1:],
                            "task_results": results,
                        }
                    else:
                        # 所有任务完成，整合结果
                        final_result = self._integrate_results(results)
                        return {
                            "messages": [AIMessage(content=final_result)],
                            "task_results": results,
                            "status": "completed",
                        }
        
        return manager_node

# 多智能体状态模式
class MultiAgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    current_task: Optional[str]
    remaining_tasks: list[str]
    task_results: list[dict]
    specialist_result: Optional[dict]
    status: str

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

from langgraph.constants import INTERRUPT

def approval_node(state: State) -> State:
    """需要人工审批的节点"""
    if state["requires_approval"]:
        # 触发中断，等待人工介入
        raise INTERRUPT("Waiting for human approval")
    
    return {"approved": True}

# 使用中断后恢复执行
app = graph.compile(checkpointer=checkpointer)

# 第一次执行到中断点
try:
    result = app.invoke({"data": "sensitive_operation"})
except Interrupt as e:
    print(f"Interrupted: {e.value}")
    
# 人工处理后恢复执行
config = {"configurable": {"thread_id": "thread-1"}}
app.update_state(config, {"approved": True})
result = app.invoke(None, config)  # 从中断点恢复

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

def create_processing_subgraph():
    """创建处理子图"""
    subgraph = StateGraph(ProcessingState)
    subgraph.add_node("validate", validate_node)
    subgraph.add_node("process", process_node)
    subgraph.add_edge("validate", "process")
    return subgraph

# 在主图中使用子图
main_graph = StateGraph(MainState)
processing_graph = create_processing_subgraph()

main_graph.add_node("preprocessing", preprocessing_node)
main_graph.add_node("processing", processing_graph.compile())
main_graph.add_node("postprocessing", postprocessing_node)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

class DynamicGraph:
    """动态图：支持运行时图结构修改"""
    
    def __init__(self):
        self.graph = StateGraph(DynamicState)
        self.compiled_cache = {}
    
    def add_conditional_node(self, state: DynamicState):
        """根据状态动态添加节点"""
        if state["node_type"] == "llm":
            self.graph.add_node("dynamic_llm", llm_node)
        elif state["node_type"] == "tool":
            self.graph.add_node("dynamic_tool", tool_node)
    
    def compile_dynamic(self, state: DynamicState):
        """动态编译图"""
        graph_hash = self._compute_graph_hash(state)
        if graph_hash not in self.compiled_cache:
            self.add_conditional_node(state)
            self.compiled_cache[graph_hash] = self.graph.compile()
        return self.compiled_cache[graph_hash]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

class OptimizedExecutor:
    """优化执行器：实现多种性能优化策略"""
    
    def __init__(self):
        self.node_cache = {}  # 节点结果缓存
        self.execution_pool = ThreadPoolExecutor(max_workers=4)
    
    async def execute_with_cache(self, node: PregelNode, state: dict):
        """带缓存的节点执行"""
        # 生成缓存键
        cache_key = self._generate_cache_key(node, state)
        
        # 检查缓存
        if cache_key in self.node_cache:
            return self.node_cache[cache_key]
        
        # 执行并缓存结果
        result = await self._execute_node(node, state)
        self.node_cache[cache_key] = result
        return result
    
    def _generate_cache_key(self, node: PregelNode, state: dict) -> str:
        """生成缓存键"""
        state_hash = hashlib.md5(
            json.dumps(state, sort_keys=True).encode()
        ).hexdigest()
        return f"{node.name}:{state_hash}"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

class MemoryManager:
    """内存管理器：优化内存使用和垃圾回收"""
    
    def __init__(self, max_checkpoints: int = 100):
        self.max_checkpoints = max_checkpoints
        self.checkpoint_cache = OrderedDict()
    
    def add_checkpoint(self, checkpoint_id: str, checkpoint: Checkpoint):
        """添加检查点并管理内存"""
        # 移除最老的检查点
        if len(self.checkpoint_cache) >= self.max_checkpoints:
            self.checkpoint_cache.popitem(last=False)
        
        self.checkpoint_cache[checkpoint_id] = checkpoint
    
    def cleanup_unused_channels(self, active_channels: set[str]):
        """清理未使用的通道"""
        for channel_key in list(self.channels.keys()):
            if channel_key not in active_channels:
                del self.channels[channel_key]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

class GraphProfiler:
    """图分析器：提供性能监控和调试能力"""
    
    def __init__(self):
        self.execution_stats = defaultdict(list)
        self.error_log = []
    
    async def profile_execution(self, graph, input_data):
        """分析图执行性能"""
        start_time = time.time()
        
        async for step in graph.astream(input_data):
            step_time = time.time() - start_time
            self.execution_stats["step_times"].append(step_time)
            
            # 记录节点执行时间
            for node_name, node_result in step.items():
                if hasattr(node_result, "execution_time"):
                    self.execution_stats[f"node_{node_name}"].append(
                        node_result.execution_time
                    )
        
        return self._generate_performance_report()
    
    def _generate_performance_report(self) -> dict:
        """生成性能报告"""
        return {
            "total_steps": len(self.execution_stats["step_times"]),
            "avg_step_time": statistics.mean(self.execution_stats["step_times"]),
            "slowest_nodes": self._find_slowest_nodes(),
            "memory_usage": self._get_memory_stats(),
        }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

class RobustNode:
    """健壮的节点实现"""
    
    def __init__(self, max_retries: int = 3):
        self.max_retries = max_retries
    
    async def execute(self, state: dict) -> dict:
        """带重试和错误处理的节点执行"""
        for attempt in range(self.max_retries):
            try:
                return await self._core_logic(state)
            except RetryableError as e:
                if attempt == self.max_retries - 1:
                    # 最后一次尝试失败，记录错误并降级处理
                    return self._fallback_logic(state, e)
                await asyncio.sleep(2 ** attempt)  # 指数退避
            except FatalError as e:
                # 不可恢复错误，立即失败
                raise
    
    async def _core_logic(self, state: dict) -> dict:
        """核心业务逻辑"""
        # 实际的节点处理逻辑
        pass
    
    def _fallback_logic(self, state: dict, error: Exception) -> dict:
        """降级处理逻辑"""
        return {"error": str(error), "status": "fallback"}