LangGraph + Neo4j 集成
概述
LangGraph 是一个用于构建有状态、多角色应用的框架，使用大语言模型（LLM）。与固定的代理架构不同，LangGraph 让您可以将自定义工作流定义为图——完全控制代理的推理、路由和执行。它提供循环图结构、内置持久化、人机交互模式以及流式支持。
虽然它可以无缝集成到 LangChain 生态系统，但也可以完全独立使用，无需将 LangChain 作为依赖。
安装
pip install langgraph langchain-neo4j langchain-mcp-adapters核心功能
-
可定制架构：精确定义代理的工作流程——添加验证步骤、并行分支、审批关卡或多代理协作
-
显式的节点和边定义，用于代理逻辑
-
内置持久化和人机交互模式
-
流式支持，实现实时响应
-
可独立运行，也可与 LangChain 集成
扩展点
1. 自定义图架构
LangGraph 的核心价值在于架构灵活性。您定义节点（函数）和边（转移），即可创建任何工作流
from langgraph.graph import END, START, StateGraph
from typing_extensions import Annotated, TypedDict
class MessagesState(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
llm_calls: int
# Build your custom workflow
agent_builder = StateGraph(MessagesState)
# Add nodes - each is a function that transforms state
agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("tool_node", tool_node)
# Define the flow with edges
agent_builder.add_edge(START, "llm_call")
agent_builder.add_conditional_edges("llm_call", should_continue, ["tool_node", END])
agent_builder.add_edge("tool_node", "llm_call")
agent = agent_builder.compile()您可以构建的示例架构
-
ReAct 循环与工具调用（示例在 notebook 中）
-
具有交接的多代理系统
-
带有验证和重试逻辑的流水线
-
人机交互审批工作流
-
并行执行分支并合并结果
2. MCP 集成
使用 langchain-mcp-adapters 连接 MCP 服务器
from langchain_mcp_adapters.client import MultiServerMCPClient
cypher_mcp_config = {
"neo4j-database": {
"transport": "stdio",
"command": "uvx",
"args": ["mcp-neo4j-cypher"],
"env": {
"NEO4J_URI": os.environ["NEO4J_URI"],
"NEO4J_USERNAME": os.environ["NEO4J_USERNAME"],
"NEO4J_PASSWORD": os.environ["NEO4J_PASSWORD"],
"NEO4J_DATABASE": os.environ["NEO4J_DATABASE"]
}
}
}
client = MultiServerMCPClient(cypher_mcp_config)
mcp_tools = await client.get_tools()3. 直接 Neo4j 集成
langchain-neo4j 包提供原生集成
-
Neo4jGraph: 直接连接，可在自定义工具中执行 Cypher 查询
-
Neo4jVector: 向量存储集成，用于通过自定义检索查询进行语义搜索
4. 自定义工具
使用 @tool 装饰器定义自定义 Neo4j 工具，并与 MCP 工具组合
from langchain.tools import tool
from langchain_neo4j import Neo4jGraph
neo4j_graph = Neo4jGraph()
@tool
async def get_investments(company: str) -> str:
"""Returns the investments by a company by name."""
results = neo4j_graph.query("""
MATCH (o:Organization)-[:HAS_INVESTOR]->(i)
WHERE o.name = $company
RETURN i.id as id, i.name as name, head(labels(i)) as type
""", {"company": company})
return json.dumps(results, indent=2)5. Neo4j 检查点保存器
langgraph-checkpoint-neo4j 包提供检查点保存器，用于将代理状态持久化到 Neo4j
-
Neo4jSaver: 同步检查点，用于存储对话历史和代理状态
-
AsyncNeo4jSaver: 异步变体，实现非阻塞的检查点操作
from langgraph.checkpoint.neo4j import AsyncNeo4jSaver
async with await AsyncNeo4jSaver.from_conn_string(
uri=NEO4J_URI,
user=NEO4J_USERNAME,
password=NEO4J_PASSWORD,
database=NEO4J_DATABASE
) as checkpointer:
await checkpointer.setup()
agent = agent_builder.compile(checkpointer=checkpointer)MCP 身份验证
支持的机制
✅ 环境变量（STDIO 传输） - 对于本地 MCP 服务器（如 mcp-neo4j-cypher），凭证通过生成时的 env 参数传递。
✅ HTTP 标头（HTTP/SSE 传输） - 对于远程 MCP 服务器，可通过 headers 参数传递 API 密钥或 Bearer 令牌。
❌ OAuth 2.0（客户端内） - 当前 Python SDK 不支持客户端内的 OAuth2 MCP 认证。
HTTP 传输配置（例如，针对 Google Colab）
cypher_mcp_config = {
"neo4j-database": {
"url": "https://:8000/mcp",
"transport": "streamable_http"
}
}