《基于 FastAPI + LangGraph + LLM 大语言模型的通用 Agent 多智能体系统架构设计与开发实战、产业应用 I 》

禅与计算机程序设计艺术

417人浏览 · 2026-01-18 04:07:14

禅与计算机程序设计艺术 · 2026-01-18 04:07:14 发布

第2章：FastAPI框架核心原理与架构

2.1 FastAPI简介：高性能Python Web框架的崛起

FastAPI是由Sebastián Ramírez于2018年创建的现代、快速（高性能）的Python Web框架，基于标准Python类型提示构建。它结合了Starlette的高性能和Pydantic的类型安全性，为开发API提供了前所未有的速度和便捷性。FastAPI的设计目标是让开发者能够快速构建高性能、易维护的API，同时提供自动生成的交互式文档。

FastAPI的主要特点包括：

极快的性能：基于Starlette和Uvicorn，性能接近Node.js和Go
快速开发：减少约40%的开发时间
更少的错误：减少约40%的人为错误
智能：自动补全支持，减少调试时间
简单：易于学习和使用
简短：减少代码重复
健壮：生产级代码，自动生成交互式文档
标准化：基于并完全兼容OpenAPI和JSON Schema

2.2 FastAPI的核心架构设计理念

FastAPI的架构设计基于几个核心理念：

2.2.1 类型提示优先

FastAPI充分利用Python 3.6+的类型提示功能，将其作为API定义的核心。通过类型提示，FastAPI能够自动验证请求数据、序列化响应数据，并生成交互式文档。这种设计使得API定义更加清晰，同时提供了强大的编辑器支持。

2.2.2 模块化与可扩展性

FastAPI采用模块化设计，允许开发者根据需要添加中间件、路由和依赖项。它的架构使得扩展功能变得简单，同时保持核心框架的轻量级。

2.2.3 异步支持

FastAPI原生支持异步编程，允许开发者编写高性能的异步端点。这对于处理大量并发请求或I/O密集型任务非常重要。

2.2.4 自动文档生成

FastAPI能够根据代码中的类型提示和注释自动生成交互式API文档（Swagger UI和ReDoc），这大大减少了维护文档的工作量。

2.2.5 安全性与性能

FastAPI在设计时充分考虑了安全性和性能，提供了内置的安全功能，同时保持了极高的性能水平。

2.3 路由系统：请求的入口与分发机制

路由是FastAPI的核心组件之一，它负责将HTTP请求映射到相应的处理函数。FastAPI的路由系统基于Starlette的路由系统，但添加了更多的功能和类型安全支持。

2.3.1 基本路由定义

在FastAPI中，路由是通过装饰器定义的。以下是一个简单的路由示例：

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def read_root():
    return {"Hello": "World"}

@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

在这个示例中，@app.get("/")装饰器定义了一个GET请求的路由，路径为"/"，处理函数为read_root。@app.get("/items/{item_id}")定义了一个带有路径参数的路由，item_id是路径参数，q是查询参数。

2.3.2 路径参数

路径参数是URL路径的一部分，用于传递动态值。在FastAPI中，路径参数通过在路径字符串中使用{参数名}的方式定义，并在函数参数中指定类型。FastAPI会自动验证路径参数的类型，并在类型不匹配时返回适当的错误信息。

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

在这个示例中，item_id是一个整数类型的路径参数。如果请求的URL是/items/5，那么item_id的值将是5；如果请求的URL是/items/abc，FastAPI会自动返回422 Unprocessable Entity错误。

2.3.3 查询参数

查询参数是URL中?后面的部分，用于传递可选的参数。在FastAPI中，查询参数通过函数参数定义，默认值为None表示该参数是可选的。

@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

在这个示例中，skip和limit是查询参数，默认值分别为0和10。请求/items/?skip=20&limit=50会将skip设置为20，limit设置为50。

2.3.4 请求体

请求体用于在POST、PUT等请求中发送数据。在FastAPI中，请求体通过Pydantic模型定义，FastAPI会自动解析请求体并验证数据类型。

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    price: float
    is_offer: bool = None

@app.post("/items/")
async def create_item(item: Item):
    return item.dict()

在这个示例中，Item是一个Pydantic模型，定义了请求体的结构。FastAPI会自动解析JSON请求体，并将其转换为Item对象。

2.3.5 路由的顺序与匹配

FastAPI的路由匹配是按照定义的顺序进行的。当一个请求到达时，FastAPI会按照路由定义的顺序依次匹配，直到找到第一个匹配的路由。因此，定义路由的顺序非常重要，特别是当有多个路由可能匹配同一个请求时。

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

@app.get("/items/latest")
async def read_latest_item():
    return {"item_id": "latest"}

在这个示例中，/items/latest路由必须定义在/items/{item_id}路由之前，否则/items/latest请求会被/items/{item_id}路由匹配，item_id的值为"latest"，而不是预期的read_latest_item函数处理。

2.4 请求处理流程：从接收请求到生成响应

FastAPI的请求处理流程可以分为以下几个步骤：

接收请求：当一个HTTP请求到达时，Uvicorn服务器接收请求并将其传递给FastAPI应用。
路由匹配：FastAPI根据请求的方法和路径匹配相应的路由。
参数解析与验证：FastAPI解析请求中的路径参数、查询参数和请求体，并根据定义的类型进行验证。
依赖注入：FastAPI解析并执行路由处理函数的依赖项。
执行处理函数：FastAPI调用路由处理函数，传递解析后的参数。
生成响应：FastAPI将处理函数的返回值转换为HTTP响应。
发送响应：Uvicorn服务器将HTTP响应发送回客户端。

2.4.1 中间件的作用

中间件是在请求处理流程中执行的函数，它可以在请求到达路由处理函数之前或之后执行。中间件可以用于处理跨域资源共享（CORS）、认证、日志记录等。

from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

在这个示例中，中间件函数add_process_time_header在请求处理前后执行，计算请求处理时间并将其添加到响应头中。

2.4.2 异常处理

FastAPI提供了内置的异常处理机制，可以捕获并处理路由处理函数中抛出的异常。开发者也可以自定义异常处理函数。

from fastapi import FastAPI, HTTPException

app = FastAPI()

items = {"foo": "The Foo Wrestlers"}

@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item_id": item_id, "item": items[item_id]}

在这个示例中，如果请求的item_id不存在于items字典中，处理函数会抛出HTTPException异常，FastAPI会将其转换为适当的HTTP响应。

2.5 响应系统：结构化数据与媒体类型处理

FastAPI的响应系统非常灵活，支持多种响应类型，包括JSON、HTML、文件、流媒体等。

2.5.1 JSON响应

默认情况下，FastAPI会将处理函数的返回值转换为JSON响应。如果返回值是一个Pydantic模型，FastAPI会自动将其转换为字典，然后序列化为JSON。

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    price: float
    is_offer: bool = None

@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: str):
    return {"name": "Foo", "price": 50.2}

在这个示例中，response_model=Item指定了响应模型，FastAPI会自动验证响应数据是否符合模型定义，并将其序列化为JSON。

2.5.2 自定义响应

FastAPI允许开发者返回自定义响应，例如HTML响应、文件响应等。

from fastapi import FastAPI, Response
from fastapi.responses import HTMLResponse, FileResponse

app = FastAPI()

@app.get("/html")
async def read_html():
    html_content = """
    <html>
        <body>
            <h1>Hello World</h1>
        </body>
    </html>
    """
    return HTMLResponse(content=html_content, status_code=200)

@app.get("/file")
async def read_file():
    return FileResponse("path/to/file.txt")

在这个示例中，HTMLResponse用于返回HTML内容，FileResponse用于返回文件。

2.5.3 状态码与响应头

FastAPI允许开发者自定义响应的状态码和响应头。

from fastapi import FastAPI, Response

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: str, response: Response):
    response.headers["X-Custom-Header"] = "Custom Value"
    return {"item_id": item_id}

在这个示例中，通过Response对象可以设置响应头。也可以直接返回JSONResponse对象来设置状态码和响应头：

from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: str):
    return JSONResponse(
        content={"item_id": item_id},
        status_code=200,
        headers={"X-Custom-Header": "Custom Value"}
    )

2.6 依赖注入系统：解耦与复用的艺术

依赖注入是FastAPI的核心功能之一，它允许开发者将代码中的依赖项解耦，提高代码的可维护性和可测试性。

2.6.1 基本依赖注入

在FastAPI中，依赖项是一个可以被调用的对象，它可以返回任何值。依赖项可以是函数、类或任何其他可调用对象。

from fastapi import FastAPI, Depends

app = FastAPI()

async def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons

@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return commons

在这个示例中，common_parameters函数是一个依赖项，它返回一个包含查询参数的字典。read_items和read_users函数通过Depends注入这个依赖项，从而复用了查询参数的处理逻辑。

2.6.2 类作为依赖项

除了函数，类也可以作为依赖项。当一个类作为依赖项时，FastAPI会自动实例化这个类，并将其传递给处理函数。

from fastapi import FastAPI, Depends

app = FastAPI()

class CommonQueryParams:
    def __init__(self, q: str = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
    return {"q": commons.q, "skip": commons.skip, "limit": commons.limit}

在这个示例中，CommonQueryParams类作为依赖项，FastAPI会自动实例化这个类，并将其传递给read_items函数。

2.6.3 依赖项的层级

依赖项可以嵌套，形成层级结构。这允许开发者构建复杂的依赖系统，同时保持代码的清晰和可维护性。

from fastapi import FastAPI, Depends

app = FastAPI()

async def query_extractor(q: str = None):
    return q

async def query_or_cookie_extractor(
    q: str = Depends(query_extractor),
    last_query: str = Cookie(None),
):
    if not q:
        return last_query
    return q

@app.get("/items/")
async def read_query(
    query_or_default: str = Depends(query_or_cookie_extractor),
):
    return {"q_or_cookie": query_or_default}

在这个示例中，query_or_cookie_extractor依赖于query_extractor，形成了一个依赖层级。

2.6.4 全局依赖项

FastAPI允许开发者定义全局依赖项，这些依赖项会应用于所有的路由。

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

app = FastAPI()

security = HTTPBearer()

async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    if credentials.credentials != "secret-token":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )

app = FastAPI(dependencies=[Depends(verify_token)])

@app.get("/items/")
async def read_items():
    return [{"item": "Foo"}, {"item": "Bar"}]

在这个示例中，verify_token依赖项被定义为全局依赖项，所有的路由都会应用这个依赖项。

2.7 中间件机制：请求响应的预处理与后处理

中间件是FastAPI中的一个重要概念，它允许开发者在请求到达路由处理函数之前或之后执行代码。中间件可以用于处理跨域资源共享（CORS）、认证、日志记录、请求修改等。

2.7.1 内置中间件

FastAPI提供了一些内置中间件，例如CORS中间件、GZip中间件等。

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

origins = [
    "http://localhost.tiangolo.com",
    "https://localhost.tiangolo.com",
    "http://localhost",
    "http://localhost:8080",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.get("/")
async def read_root():
    return {"Hello": "World"}

在这个示例中，CORSMiddleware被添加到应用中，用于处理跨域资源共享。

2.7.2 自定义中间件

开发者也可以自定义中间件，以满足特定的需求。

from fastapi import FastAPI, Request
import time

app = FastAPI()

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

@app.get("/")
async def read_root():
    return {"Hello": "World"}

在这个示例中，自定义中间件add_process_time_header计算请求处理时间，并将其添加到响应头中。

2.8 异步编程模型：高性能并发的实现

FastAPI原生支持异步编程，允许开发者编写高性能的异步端点。异步编程对于处理大量并发请求或I/O密集型任务非常重要。

…

第3章：LangGraph框架核心原理与架构

3.1 LangGraph简介：大语言模型应用的编排框架

LangGraph是一个用于构建多智能体系统的Python框架，由LangChain团队开发。它专注于提供一种简单、灵活的方式来编排大语言模型（LLM）应用，特别是涉及多个智能体协作的复杂场景。LangGraph的设计目标是让开发者能够快速构建具有状态管理、事件驱动和插件扩展能力的多智能体系统。

LangGraph的主要特点包括：

状态管理：内置的状态管理系统，用于跟踪智能体对话的上下文和历史
事件驱动：基于事件的架构，允许智能体之间进行灵活的通信和协作
插件系统：支持通过插件扩展智能体的功能
可视化：内置的可视化工具，用于展示智能体的协作流程
兼容性：与LangChain生态系统无缝集成，支持多种LLM平台

3.2 LangGraph的核心设计哲学

LangGraph的架构设计基于几个核心理念：

3.2.1 以智能体为中心

LangGraph将智能体作为系统的核心组件，每个智能体都有自己的状态、能力和职责。智能体之间通过事件进行通信和协作，形成一个分布式的系统。

3.2.2 状态优先

LangGraph内置了强大的状态管理系统，用于跟踪智能体对话的上下文和历史。状态管理是LangGraph的核心功能之一，它允许智能体在对话过程中保持记忆，从而实现更复杂的交互。

3.2.3 事件驱动

LangGraph采用事件驱动的架构，智能体之间通过事件进行通信。当一个智能体执行某个动作时，它会发布一个事件，其他智能体可以监听这些事件并做出相应的反应。

3.2.4 可扩展性

LangGraph的插件系统允许开发者通过插件扩展智能体的功能。插件可以是任何Python对象，它可以添加新的能力、修改智能体的行为或与外部系统集成。

3.2.5 可视化

LangGraph内置了可视化工具，用于展示智能体的协作流程。可视化工具可以帮助开发者理解智能体之间的交互，调试和优化多智能体系统。

3.3 状态管理系统：智能体对话的记忆与上下文

状态管理是LangGraph的核心功能之一，它允许智能体在对话过程中保持记忆，从而实现更复杂的交互。LangGraph的状态管理系统基于LangChain的记忆系统，但添加了更多的功能和灵活性。

3.3.1 基本状态管理

在LangGraph中，状态是一个字典，用于存储智能体对话的上下文和历史。每个智能体都有自己的状态，但也可以访问共享状态。

from langgraph import Graph, State

# 定义状态
state = State(
    {"messages": [], "current_topic": None, "user_info": {}}
)

# 创建图
app = Graph(state)

在这个示例中，我们定义了一个状态，包含messages、current_topic和user_info三个字段。messages用于存储对话历史，current_topic用于存储当前对话主题，user_info用于存储用户信息。

3.3.2 状态更新

智能体可以通过修改状态来更新对话上下文。在LangGraph中，状态更新是通过函数实现的，这些函数接收当前状态并返回新的状态。

def add_message(state, message):
    new_state = state.copy()
    new_state["messages"].append(message)
    return new_state

# 添加节点
app.add_node("add_message", add_message)

在这个示例中，add_message函数接收当前状态和一个消息，然后返回一个新的状态，其中messages字段添加了新的消息。

3.3.3 状态持久化

LangGraph支持状态持久化，允许开发者将状态保存到外部存储（如数据库、文件等），以便在对话结束后恢复状态。

from langgraph.state import SQLiteStateStore

# 创建SQLite状态存储
state_store = SQLiteStateStore("state.db")

# 创建图，使用SQLite状态存储
app = Graph(state, state_store=state_store)

在这个示例中，我们使用SQLiteStateStore将状态保存到SQLite数据库中。LangGraph还支持其他状态存储，如RedisStateStore、FileSystemStateStore等。

3.3.4 状态共享

在多智能体系统中，智能体之间可能需要共享状态。LangGraph允许智能体访问共享状态，从而实现更复杂的协作。

from langgraph import Graph, State, Agent

# 定义共享状态
shared_state = State({"global_counter": 0})

# 创建智能体
agent1 = Agent("agent1", state=State({"local_counter": 0}))
agent2 = Agent("agent2", state=State({"local_counter": 0}))

# 创建图，添加智能体和共享状态
app = Graph(shared_state)
app.add_agent(agent1)
app.add_agent(agent2)

在这个示例中，我们创建了两个智能体和一个共享状态。智能体可以访问自己的本地状态和共享状态。

3.4 事件驱动机制：动态响应与流程控制

3.4.1 事件发布与订阅

在LangGraph中，事件是一个字典，包含事件类型和事件数据。智能体可以通过publish_event方法发布事件，通过subscribe_event方法订阅事件。

from langgraph import Graph, State, Agent

# 创建智能体
class MyAgent(Agent):
    def on_event(self, event):
        if event["type"] == "message_received":
            print(f"Received message: {event['data']['message']}")

# 创建图
app = Graph(State({}))

# 添加智能体
agent = MyAgent("my_agent")
app.add_agent(agent)

# 订阅事件
agent.subscribe_event("message_received", agent.on_event)

# 发布事件
app.publish_event("message_received", {"message": "Hello World"})

在这个示例中，MyAgent类定义了一个on_event方法，用于处理message_received事件。智能体通过subscribe_event方法订阅事件，当事件发布时，on_event方法会被调用。

3.4.2 事件路由

LangGraph支持事件路由，允许开发者根据事件类型和事件数据将事件路由到不同的智能体。

from langgraph import Graph, State, Agent

# 创建智能体
class MessageAgent(Agent):
    def on_message(self, event):
        print(f"Message agent received message: {event['data']['message']}")

class NotificationAgent(Agent):
    def on_notification(self, event):
        print(f"Notification agent received notification: {event['data']['message']}")

# 创建图
app = Graph(State({}))

# 添加智能体
message_agent = MessageAgent("message_agent")
notification_agent = NotificationAgent("notification_agent")
app.add_agent(message_agent)
app.add_agent(notification_agent)

# 路由事件
app.add_event_route("message", message_agent.on_message)
app.add_event_route("notification", notification_agent.on_notification)

# 发布事件
app.publish_event("message", {"message": "Hello World"})
app.publish_event("notification", {"message": "New notification"})

在这个示例中，我们创建了两个智能体：MessageAgent和NotificationAgent。MessageAgent处理message事件，NotificationAgent处理notification事件。通过add_event_route方法，我们将事件路由到相应的智能体。

3.4.3 事件流程

LangGraph允许开发者定义事件流程，即事件的处理顺序。事件流程可以是线性的、分支的或循环的。

from langgraph import Graph, State

# 定义状态
state = State({"messages": [], "step": 0})

# 创建图
app = Graph(state)

# 定义节点
def step1(state):
    new_state = state.copy()
    new_state["messages"].append("Step 1 executed")
    new_state["step"] = 1
    return new_state

def step2(state):
    new_state = state.copy()
    new_state["messages"].append("Step 2 executed")
    new_state["step"] = 2
    return new_state

def step3(state):
    new_state = state.copy()
    new_state["messages"].append("Step 3 executed")
    new_state["step"] = 3
    return new_state

# 添加节点
app.add_node("step1", step1)
app.add_node("step2", step2)
app.add_node("step3", step3)

# 添加边
app.add_edge("start", "step1")
app.add_edge("step1", "step2")
app.add_edge("step2", "step3")
app.add_edge("step3", "end")

# 运行图
result = app.run()
print(result["messages"])

在这个示例中，我们定义了一个线性的事件流程：step1 → step2 → step3。当图运行时，这些节点会按顺序执行。

3.5 插件系统：扩展LangGraph的能力边界

LangGraph的插件系统允许开发者通过插件扩展智能体的功能。插件可以是任何Python对象，它可以添加新的能力、修改智能体的行为或与外部系统集成。

3.5.1 基本插件开发

在LangGraph中，插件是一个类，继承自langgraph.plugin.Plugin。插件可以定义自己的方法、事件处理程序和状态。

from langgraph import Plugin

class MyPlugin(Plugin):
    def __init__(self, name):
        super().__init__(name)
        self.counter = 0

    def increment_counter(self):
        self.counter += 1
        return self.counter

    def on_event(self, event):
        if event["type"] == "reset_counter":
            self.counter = 0

在这个示例中，MyPlugin类定义了一个increment_counter方法和一个on_event方法。increment_counter方法用于增加计数器，on_event方法用于处理reset_counter事件。

3.5.2 插件注册与使用

开发者可以通过add_plugin方法将插件注册到智能体中，然后通过智能体的plugins属性访问插件。

from langgraph import Graph, State, Agent

# 创建智能体
agent = Agent("my_agent")

# 添加插件
plugin = MyPlugin("my_plugin")
agent.add_plugin(plugin)

# 使用插件
counter = agent.plugins["my_plugin"].increment_counter()
print(f"Counter: {counter}")

# 发布事件，触发插件的事件处理程序
agent.publish_event("reset_counter", {})
counter = agent.plugins["my_plugin"].increment_counter()
print(f"Counter after reset: {counter}")

在这个示例中，我们创建了一个智能体，并添加了MyPlugin插件。然后，我们通过agent.plugins["my_plugin"]访问插件，并调用其方法。

3.5.3 插件生命周期

LangGraph的插件系统支持插件生命周期管理，允许开发者在插件注册、注销、启用和禁用时执行特定的操作。

from langgraph import Plugin

class LifecyclePlugin(Plugin):
    def on_register(self, agent):
        print(f"Plugin registered to agent: {agent.name}")

    def on_unregister(self, agent):
        print(f"Plugin unregistered from agent: {agent.name}")

    def on_enable(self):
        print("Plugin enabled")

    def on_disable(self):
        print("Plugin disabled")

在这个示例中，LifecyclePlugin类定义了四个生命周期方法：on_register、on_unregister、on_enable和on_disable。这些方法会在相应的生命周期事件发生时被调用。

3.5.4 插件市场

LangGraph社区维护了一个插件市场，包含各种预构建的插件，如LLM插件、工具插件、存储插件等。开发者可以直接使用这些插件，快速扩展智能体的功能。

from langgraph.plugins import OpenAIPlugin

# 创建OpenAI插件
openai_plugin = OpenAIPlugin("openai_plugin", api_key="your-api-key")

# 添加插件到智能体
agent.add_plugin(openai_plugin)

# 使用插件生成文本
response = agent.plugins["openai_plugin"].generate_text(
    prompt="Hello, how are you?",
    model="gpt-3.5-turbo"
)
print(response)

在这个示例中，我们使用OpenAIPlugin调用OpenAI的API生成文本。LangGraph还支持其他LLM插件，如AnthropicPlugin、GooglePlugin等。

3.6 智能体通信机制：多智能体协作的基础

在多智能体系统中，智能体之间的通信是协作的基础。LangGraph提供了多种通信机制，允许智能体之间进行灵活的通信。

3.6.1 直接通信

智能体之间可以通过直接调用方法进行通信。这种通信方式适用于简单的协作场景。

from langgraph import Graph, State, Agent

# 创建智能体
class AgentA(Agent):
    def __init__(self, name):
        super().__init__(name)
        self.agent_b = None

    def set_agent_b(self, agent_b):
        self.agent_b = agent_b

    def send_message(self, message):
        if self.agent_b:
            self.agent_b.receive_message(message)

class AgentB(Agent):
    def receive_message(self, message):
        print(f"AgentB received message: {message}")

# 创建图
app = Graph(State({}))

# 创建智能体
agent_a = AgentA("agent_a")
agent_b = AgentB("agent_b")
app.add_agent(agent_a)
app.add_agent(agent_b)

# 设置智能体之间的引用
agent_a.set_agent_b(agent_b)

# 发送消息
agent_a.send_message("Hello from AgentA")

在这个示例中，AgentA通过直接调用AgentB的receive_message方法发送消息。

3.6.2 事件通信

事件通信是LangGraph推荐的通信方式，它允许智能体之间进行松耦合的通信。智能体通过发布事件和订阅事件进行通信。

from langgraph import Graph, State, Agent

# 创建智能体
class AgentA(Agent):
    def send_message(self, message):
        self.publish_event("message_sent", {"message": message, "sender": self.name})

class AgentB(Agent):
    def __init__(self, name):
        super().__init__(name)
        self.subscribe_event("message_sent", self.on_message_received)

    def on_message_received(self, event):
        print(f"AgentB received message from {event['data']['sender']}: {event['data']['message']}")

# 创建图
app = Graph(State({}))

# 创建智能体
agent_a = AgentA("agent_a")
agent_b = AgentB("agent_b")
app.add_agent(agent_a)
app.add_agent(agent_b)

# 发送消息
agent_a.send_message("Hello from AgentA")

在这个示例中，AgentA通过发布message_sent事件发送消息，AgentB通过订阅message_sent事件接收消息。

3.6.3 消息队列通信

LangGraph支持消息队列通信，允许智能体之间通过消息队列进行异步通信。这种通信方式适用于高并发、分布式的场景。

from langgraph import Graph, State, Agent
from langgraph.communication import RabbitMQMessageQueue

# 创建消息队列
message_queue = RabbitMQMessageQueue(
    host="localhost",
    port=5672,
    username="guest",
    password="guest"
)

# 创建智能体
class AgentA(Agent):
    def __init__(self, name):
        super().__init__(name)
        self.message_queue = message_queue

    def send_message(self, message):
       ...