3.3 性能优化与并发控制

本节导读：通过本节学习，你将掌握LangGraph的性能优化策略，学会处理并发执行和资源管理，构建高效可靠的智能体系统。

学习目标

• 理解LangGraph的性能瓶颈和优化机会
• 掌握并发控制和异步执行技术
• 学会优化状态管理和检查点机制
• 了解内存管理和资源清理策略
• 构建高性能的生产环境智能体

核心概念

LangGraph的性能优化是一个多维度的问题，涉及执行效率、内存管理、并发控制等多个方面。良好的性能优化能够显著提升智能体的响应速度和资源利用率。

性能优化维度

环境准备 / 前置知识

依赖安装

# 基础性能优化
pip install -U langgraph langchain-openai

# 高级性能特性
pip install -U aioredis asyncpg
​

前置要求

• 熟悉Python异步编程
• 了解状态管理基本概念
• 掌握基本的性能分析方法

分步实战

步骤 1：基础性能优化

import asyncio
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated, Sequence
import time
import psutil

class OptimizedState(TypedDict):
    messages: Annotated[Sequence[dict], add_messages]
    processing_time: float
    memory_usage: float
    result_cache: dict

def optimized_analysis_node(state: OptimizedState):
    """优化后的分析节点"""
    start_time = time.time()
    
    # 检查缓存
    query_hash = hash(state["messages"][-1]["content"])
    if query_hash in state["result_cache"]:
        cached_result = state["result_cache"][query_hash]
        print(f"使用缓存结果，节省了 {time.time() - start_time:.2f} 秒")
        return cached_result
    
    llm = ChatOpenAI(model="gpt-4-turbo")
    
    # 优化提示词
    user_query = state["messages"][-1]["content"]
    optimized_prompt = f"""
    优化提示词结构，减少冗余信息
    查询：{user_query}
    重点：提供简洁准确的回答
    """
    
    response = llm.invoke([
        {"role": "system", "content": "你是数据分析专家，提供简洁精确的回答"},
        {"role": "user", "content": optimized_prompt}
    ])
    
    end_time = time.time()
    processing_time = end_time - start_time
    
    # 监控内存使用
    memory_info = psutil.Process().memory_info()
    memory_usage = memory_info.rss / 1024 / 1024  # MB
    
    result = {
        "messages": [response],
        "processing_time": processing_time,
        "memory_usage": memory_usage,
        "result_cache": {**state["result_cache"], query_hash: response.content}
    }
    
    # 缓存结果
    result["result_cache"] = state["result_cache"]
    result["result_cache"][query_hash] = response.content
    
    return result

# 构建优化图
optimized_graph = StateGraph(OptimizedState)
optimized_graph.add_node("analysis", optimized_analysis_node)
optimized_graph.add_edge(START, "analysis")
optimized_graph.add_edge("analysis", END)

# 使用内存检查点进行性能优化
optimized_checkpointer = MemorySaver()
compiled_optimized_graph = optimized_graph.compile(checkpointer=optimized_checkpointer)

# 性能测试
print("=== 性能优化测试 ===")
initial_state = {
    "messages": [{"role": "user", "content": "分析2024年Q1销售数据趋势"}],
    "processing_time": 0.0,
    "memory_usage": 0.0,
    "result_cache": {}
}

# 第一次执行
start_time = time.time()
result1 = compiled_optimized_graph.invoke(initial_state)
first_execution_time = time.time() - start_time
print(f"第一次执行时间: {first_execution_time:.2f} 秒")

# 第二次执行（使用缓存）
start_time = time.time()
result2 = compiled_optimized_graph.invoke(initial_state)
second_execution_time = time.time() - start_time
print(f"第二次执行时间: {second_execution_time:.2f} 秒")
print(f"性能提升: {((first_execution_time - second_execution_time) / first_execution_time * 100):.1f}%")
​

步骤 2：并发执行优化

from concurrent.futures import ThreadPoolExecutor, as_completed
import asyncio
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI

class ConcurrentState(TypedDict):
    messages: Annotated[Sequence[dict], add_messages]
    tasks: list
    results: dict
    error_count: int

async def concurrent_analysis_node(state: ConcurrentState):
    """并发分析节点"""
    llm = ChatOpenAI(model="gpt-4-turbo")
    
    # 模拟多个并发任务
    queries = [
        "分析用户增长趋势",
        "分析产品销售数据", 
        "分析市场竞争对手",
        "分析客户满意度"
    ]
    
    tasks = []
    results = {}
    
    # 创建并发任务
    for i, query in enumerate(queries):
        task = asyncio.create_task(
            process_single_query(llm, query, i)
        )
        tasks.append(task)
        results[i] = None
    
    # 并发执行
    try:
        completed_tasks = await asyncio.gather(*tasks, return_exceptions=True)
        
        for i, result in enumerate(completed_tasks):
            if isinstance(result, Exception):
                results[i] = f"任务 {i} 失败: {str(result)}"
            else:
                results[i] = result
    except Exception as e:
        return {
            "messages": [{"role": "assistant", "content": f"并发执行失败: {str(e)}"}],
            "tasks": queries,
            "results": results,
            "error_count": len([r for r in results.values() if isinstance(r, str) and "失败" in r])
        }
    
    return {
        "messages": [{"role": "assistant", "content": "并发分析完成"}],
        "tasks": queries,
        "results": results,
        "error_count": 0
    }

async def process_single_query(llm, query, task_id):
    """处理单个查询"""
    # 模拟处理时间
    await asyncio.sleep(1)
    
    response = await llm._agenerate([
        {"role": "system", "content": "你是数据分析专家"},
        {"role": "user", "content": f"请分析: {query}"}
    ])
    
    return {
        "task_id": task_id,
        "query": query,
        "result": response.generations[0][0].text,
        "timestamp": time.time()
    }

# 构建并发图
concurrent_graph = StateGraph(ConcurrentState)
concurrent_graph.add_node("concurrent_analysis", concurrent_analysis_node)
concurrent_graph.add_edge(START, "concurrent_analysis")
concurrent_graph.add_edge("concurrent_analysis", END)

# 使用异步执行
async def run_concurrent_analysis():
    initial_state = {
        "messages": [{"role": "user", "content": "开始并发分析"}],
        "tasks": [],
        "results": {},
        "error_count": 0
    }
    
    result = await concurrent_analysis_node(initial_state)
    return result

# 执行并发分析
print("=== 并发执行优化 ===")
result = asyncio.run(run_concurrent_analysis())
print(f"任务完成状态: {len([r for r in result['results'].values() if r is not None])}/{len(result['tasks'])}")
print(f"错误数量: {result['error_count']}")
​

步骤 3：资源管理和内存优化

import gc
import weakref
from typing import Dict, Any

class ResourceOptimizedState(TypedDict):
    messages: Annotated[Sequence[dict], add_messages]
    active_tasks: Dict[str, Any]
    memory_pool: Dict[str, Any]
    cleanup_timestamps: list

def resource_monitor_node(state: ResourceOptimizedState):
    """资源监控节点"""
    import psutil
    import time
    
    # 获取当前系统资源使用情况
    process = psutil.Process()
    memory_info = process.memory_info()
    cpu_percent = process.cpu_percent()
    
    resource_status = {
        "memory_usage_mb": memory_info.rss / 1024 / 1024,
        "cpu_percent": cpu_percent,
        "active_tasks": len(state["active_tasks"]),
        "timestamp": time.time()
    }
    
    # 检查资源使用阈值
    if memory_info.rss > 1024 * 1024 * 1024:  # 1GB
        print(f"警告: 内存使用过高 ({memory_info.rss / 1024 / 1024:.1f}MB)")
        # 执行资源清理
        cleanup_resources(state)
    
    return {
        "messages": [{"role": "assistant", "content": f"资源状态: {resource_status}"}],
        "active_tasks": state["active_tasks"],
        "memory_pool": state["memory_pool"],
        "cleanup_timestamps": state["cleanup_timestamps"] + [time.time()]
    }

def cleanup_resources(state: ResourceOptimizedState):
    """清理资源"""
    print("执行资源清理...")
    
    # 清理内存池
    state["memory_pool"] = {}
    
    # 清理不再活跃的任务
    active_task_ids = list(state["active_tasks"].keys())
    for task_id in active_task_ids:
        if task_id.endswith("_expired"):
            del state["active_tasks"][task_id]
    
    # 强制垃圾回收
    gc.collect()
    
    print("资源清理完成")

def batch_processing_node(state: ResourceOptimizedState):
    """批量处理节点"""
    batch_size = 10
    current_batch = []
    
    # 模拟批量处理
    for i in range(batch_size):
        task_id = f"batch_task_{int(time.time())}_{i}"
        current_batch.append(task_id)
        
        # 模拟任务创建
        state["active_tasks"][task_id] = {
            "status": "processing",
            "start_time": time.time()
        }
    
    # 批量处理完成后清理
    for task_id in current_batch:
        if task_id in state["active_tasks"]:
            del state["active_tasks"][task_id]
    
    return {
        "messages": [{"role": "assistant", "content": f"批量处理完成: {len(current_batch)} 个任务"}],
        "active_tasks": state["active_tasks"],
        "memory_pool": state["memory_pool"],
        "cleanup_timestamps": state["cleanup_timestamps"]
    }

# 构建资源优化图
resource_graph = StateGraph(ResourceOptimizedState)
resource_graph.add_node("resource_monitor", resource_monitor_node)
resource_graph.add_node("batch_processing", batch_processing_node)

resource_graph.add_edge(START, "resource_monitor")
resource_graph.add_edge("resource_monitor", "batch_processing")
resource_graph.add_edge("batch_processing", END)

# 运行资源优化测试
print("=== 资源管理优化测试 ===")
initial_state = {
    "messages": [{"role": "user", "content": "开始资源管理测试"}],
    "active_tasks": {},
    "memory_pool": {},
    "cleanup_timestamps": []
}

result = resource_monitor_node(initial_state)
print(f"资源监控完成: {result['messages'][0]['content']}")

result2 = batch_processing_node(result)
print(f"批量处理完成: {result2['messages'][0]['content']}")
​

常见问题 FAQ

Q1：如何处理内存泄漏问题？

A：内存泄漏是LangGraph应用中常见的问题，处理方法包括：

1. 定期清理：定期清理不再使用的数据和对象
2. 弱引用：使用弱引用避免循环引用
3. 内存监控：监控内存使用情况，及时发现异常
4. 批量处理：采用批量处理减少内存压力

# 使用弱引用避免内存泄漏
import weakref

class NodeCache:
    def __init__(self):
        self._cache = weakref.WeakValueDictionary()
    
    def add_node(self, node_id, node):
        self._cache[node_id] = node
    
    def get_node(self, node_id):
        return self._cache.get(node_id)

# 定期清理
def cleanup_cache(cache):
    current_time = time.time()
    expired_keys = [key for key, value in cache.items() 
                   if hasattr(value, 'timestamp') and current_time - value.timestamp > 3600]
    
    for key in expired_keys:
        del cache[key]
​

Q2：如何优化检查点性能？

A：检查点性能优化包括：

1. 存储选择：选择合适的存储后端（Redis、数据库等）
2. 序列化优化：优化数据序列化格式
3. 增量更新：只更新变化的部分状态
4. 异步写入：使用异步写入提高性能

# 异步检查点
async def async_checkpoint_write(checkpointer, state):
    try:
        await asyncio.to_thread(checkpointer.put, state)
        return True
    except Exception as e:
        print(f"检查点写入失败: {e}")
        return False

# 增量状态更新
def incremental_state_update(old_state, new_state):
    """只更新变化的部分"""
    updated_state = old_state.copy()
    
    # 只更新有变化的字段
    for key, value in new_state.items():
        if key not in old_state or old_state[key] != value:
            updated_state[key] = value
    
    return updated_state
​

Q3：如何处理高并发场景？

A：高并发场景的处理策略：

1. 连接池：使用数据库连接池
2. 限流：实施请求限流防止系统过载
3. 负载均衡：将请求分发到多个实例
4. 异步处理：使用异步I/O提高并发性能

# 异步连接池
import asyncpg
from contextlib import asynccontextmanager

class ConnectionPool:
    def __init__(self, database_url, max_connections=10):
        self.pool = None
        self.database_url = database_url
        self.max_connections = max_connections
    
    async def initialize(self):
        self.pool = await asyncpg.create_pool(
            self.database_url,
            max_size=self.max_connections,
            min_size=1
        )
    
    @asynccontextmanager
    async def get_connection(self):
        async with self.pool.acquire() as connection:
            yield connection
    
    async def close(self):
        if self.pool:
            await self.pool.close()

# 限流器
class RateLimiter:
    def __init__(self, max_requests_per_second):
        self.max_requests = max_requests_per_second
        self.requests = []
    
    async def acquire(self):
        now = time.time()
        # 清理过期的请求记录
        self.requests = [req_time for req_time in self.requests 
                       if now - req_time < 1.0]
        
        if len(self.requests) >= self.max_requests:
            # 等待
            sleep_time = 1.0 - (now - self.requests[0])
            await asyncio.sleep(sleep_time)
            self.requests = []
        
        self.requests.append(now)
        return True
​

最佳实践与避坑

最佳实践

1. 内存监控：定期监控内存使用情况
2. 缓存策略：合理使用缓存减少重复计算
3. 异步处理：使用异步提高并发性能
4. 资源清理：及时清理不再使用的资源
5. 错误处理：完善的错误处理和恢复机制

常见坑点

1. 内存泄漏：未正确清理对象导致内存泄漏
2. 并发竞争：多线程访问共享数据时的竞争条件
3. 检查点性能：检查点写入成为性能瓶颈
4. 资源耗尽：未限制并发数量导致资源耗尽

本节小结

通过本节学习，我们掌握了LangGraph的性能优化技术，包括基础性能优化、并发执行优化、资源管理和内存优化等。良好的性能优化能够显著提升智能体的响应速度和资源利用率，是构建生产级智能体系统的重要保障。

下一节我们将开始第4章工具集成的内容学习。

延伸阅读

• Python异步编程
• 内存管理最佳实践
• 并发编程模式

关键词：性能优化, 并发控制, 资源管理, 内存优化, 异步执行
难度：高级
预计阅读：35分钟