LangGraph构建多智能体系统的实践与优化-AI智能范式网

LangGraph构建多智能体系统的实践与优化

Maggie H·2026-08-22·阅读 22 GEO检测（免费）

LangGraph多智能体系统图计算

|难度：普通|阅读约 18 分钟

1. 项目概述：LangGraph在多智能体系统中的应用实践

最近在开发一个基于大语言模型的多智能体协作系统时，遇到了一个典型问题：当需要处理包含文件解析、网络搜索和对话生成等多种任务的复杂流程时，传统的线性处理方式显得力不从心。经过多次尝试，最终选择了LangGraph这个基于图结构的框架来构建解决方案。

LangGraph是LangChain生态中的一个重要组件，它通过有向图的方式组织多个智能体（Agent）之间的协作关系。与传统的线性链式调用不同，LangGraph允许开发者定义复杂的条件分支和循环逻辑，使得多智能体系统能够更灵活地处理各种场景。这个框架特别适合需要动态路由、条件判断和并行处理的AI应用场景。

需要模型API调用？免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。

2. 核心架构设计

2.1 系统整体结构

我们的多智能体系统主要包含以下几个核心模块：

智能体模块：处理具体任务的独立单元
图状态管理：维护整个系统的运行状态
路由决策：根据输入类型动态选择处理路径
执行引擎：驱动图结构的运行

项目目录结构如下：

code复制.
├── chains/        # 智能体实现
│   ├── generate.py # 回答生成
│   ├── models.py   # 模型加载
│   └── summary.py  # 关键词提取
├── graph/         # 图结构定义
│   ├── graph.py    
│   └── graph_state.py
├── upload_files/  # 上传文件存储
└── app.py         # 前端界面

2.2 智能体设计与实现

在chains目录下，我们实现了三个核心智能体：

模型加载智能体（models.py）：

python复制from langchain_ollama import ChatOllama, OllamaEmbeddings

def load_model(model_name: str) -> ChatOllama:
    """加载对话模型"""
    return ChatOllama(model=model_name)

def load_embeddings(model_name: str) -> OllamaEmbeddings:
    """加载嵌入模型"""
    return OllamaEmbeddings(model=model_name)

关键词提取智能体（summary.py）：

python复制from langchain.prompts import ChatPromptTemplate

class SummaryChain:
    def __init__(self, model_name):
        self.llm = load_model(model_name)
        self.prompt = ChatPromptTemplate.from_template(
            "从以下对话历史和问题中提取关键词..."
        )
        self.chain = self.prompt | self.llm

回答生成智能体（generate.py）：

python复制class GenerateChain:
    def __init__(self, model_name):
        self.llm = load_model(model_name)
        self.prompt = ChatPromptTemplate.from_template(
            "根据以下文档和对话历史回答问题..."
        )
        self.chain = self.prompt | self.llm

3. 图状态管理与节点设计

3.1 状态定义

在graph_state.py中，我们定义了系统的核心状态结构：

python复制from typing import Literal, TypedDict

class GraphState(TypedDict):
    model_name: str
    type: Literal["websearch", "file", "chat"]
    messages: list
    documents: list = []

3.2 节点实现

在graph.py中，我们实现了以下几个关键节点：

路由节点：

python复制def route_question(state: GraphState) -> str:
    if state['type'] == 'websearch':
        return "extract_keywords"
    elif state['type'] == 'file':
        return "file_process"
    else:
        return "generate"

文件处理节点：

python复制def file_process(state: GraphState, config: RunnableConfig) -> GraphState:
    vector_store = config["configurable"]["vectorstore"]
    for doc in state["documents"]:
        # 文件处理逻辑...
        vector_store.add_documents(processed_docs)
    return state

网络搜索节点：

python复制def web_search(state: GraphState) -> GraphState:
    web_search_tool = TavilySearchResults(k=3)
    docs = web_search_tool.invoke({"query": state["messages"][-1].content})
    # 处理搜索结果...
    return state

4. 图结构构建与执行

4.1 图的构建

在graph.py中，我们通过以下方式构建完整的图结构：

python复制def create_graph() -> CompiledStateGraph:
    workflow = StateGraph(GraphState)
    
    # 添加节点
    workflow.add_node("websearch", web_search)
    workflow.add_node("extract_keywords", extract_keywords)
    workflow.add_node("file_process", file_process)
    workflow.add_node("generate", generate)
    
    # 设置条件入口
    workflow.set_conditional_entry_point(
        route_question,
        {
            "extract_keywords": "extract_keywords",
            "generate": "generate",
            "file_process": "file_process",
        },
    )
    
    # 添加边和条件边
    workflow.add_edge("file_process", "extract_keywords")
    workflow.add_conditional_edges(
        "extract_keywords",
        decide_to_generate,
        {"websearch": "websearch", "generate": "generate"},
    )
    workflow.add_edge("websearch", "generate")
    workflow.add_edge("generate", END)
    
    return workflow.compile()

4.2 图的执行

系统支持两种执行方式：

命令行交互模式（main.py）：

python复制def main():
    graph = create_graph()
    while True:
        user_input = input("User: ")
        state = GraphState(
            model_name="qwen2.5:7b",
            type="chat",
            messages=[HumanMessage(user_input)]
        )
        for answer in stream_graph_updates(graph, state, config):
            print(answer, end="")

Web界面模式（app.py）：

python复制import streamlit as st

def app():
    if "graph" not in st.session_state:
        st.session_state.graph = create_graph()
    
    question = st.chat_input('输入问题')
    if question:
        with st.chat_message("user"):
            st.markdown(question)
        with st.chat_message("assistant"):
            st.write_stream(stream_graph_updates(
                st.session_state.graph,
                state,
                st.session_state.config
            ))

5. 实战经验与优化建议

5.1 性能优化技巧

模型选择：

对于通用对话，使用7B参数的模型即可
对于代码生成等专业任务，建议使用专用模型

缓存策略：

python复制from langchain.cache import InMemoryCache
langchain.llm_cache = InMemoryCache()

并行处理：
对于独立的任务节点，可以使用LangGraph的并行执行特性：

python复制workflow.add_parallel_nodes(["node1", "node2", "node3"])

5.2 常见问题排查

路由错误：

检查state['type']的值是否符合预期
验证route_question函数的返回结果

文件处理失败：

确保文件路径正确
检查文件格式是否支持

网络搜索超时：

增加超时设置
添加重试机制

5.3 扩展建议

添加新节点：

python复制def new_node(state: GraphState) -> GraphState:
    # 实现新功能
    return state

workflow.add_node("new_node", new_node)

集成其他工具：

python复制from langchain_community.tools import Tool

new_tool = Tool(
    name="new_tool",
    func=lambda x: x,
    description="..."
)

监控与日志：

python复制import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def logged_node(state: GraphState) -> GraphState:
    logger.info(f"Processing state: {state}")
    return state

6. 部署与生产化建议

6.1 容器化部署

建议使用Docker进行部署，示例Dockerfile：

dockerfile复制FROM python:3.9
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["streamlit", "run", "app.py"]

6.2 性能监控

集成Prometheus监控：

python复制from prometheus_client import start_http_server, Counter

REQUEST_COUNT = Counter('requests_total', 'Total API Requests')

def monitored_node(state: GraphState) -> GraphState:
    REQUEST_COUNT.inc()
    return state

6.3 安全建议

文件上传安全检查：

python复制import magic

def is_safe_file(file_path):
    file_type = magic.from_file(file_path, mime=True)
    return file_type in ["application/pdf", "text/plain"]

API访问控制：

python复制from fastapi import Depends, HTTPException

async def verify_token(token: str):
    if token != "valid_token":
        raise HTTPException(status_code=403)

在实际项目中，我们发现LangGraph的这种图结构设计特别适合处理需要动态路由的复杂业务流程。通过将不同的处理逻辑封装成独立的节点，并通过图的方式组织它们之间的关系，系统获得了很好的灵活性和可维护性。特别是在需要添加新功能时，只需要实现新的节点并在图中适当的位置插入即可，不会影响现有功能的正常运行。