内容纲要
一、 Multi-agent architectures 架构
继续深入探讨 Multi-Agent 概念。LangGraph 利用基于图的结构来定义代理并在它们之间建立连接。在此框架中，每个代理都表示为图中的一个节点，并通过边链接到其它代理。每个代理通过接收来自其他代理的输入并将控制权传递给下一个代理来执行其指定的操作。在 LangGraph 框架的设计中，主要通过如下几种方法来建立各个子代理之间的通信连接：
- NetWork（网络）：每个代理都可以与其他每个代理通信。任何代理都可以决定接下来要呼叫哪个其他代理。
- Supervisor（主管）：每个代理都与一个 Supervisor 代理通信。由 Supervisor 代理决定接下来应调用哪个代理。
- Supervisor（tool-calling）：Supervisor 架构的一个特例。每个代理都是一个工具。由 Supervisor 代理通过工具调用的方式来决定调用哪些子代理执行任务，以及要传递给这些代理程序的参数
- Hierarchical（分层）：定义具有 supervisor 嵌套 supervisor 多代理系统。这是 Supervisor 架构的一种泛化，允许更复杂的控制流。
但是，在具体实践各个不同多代理架构下的具体应用方法之前，我们需要结合 LangGraph 构建图的机制去思考一个问题：通过 State 可以让一个图中的所有节点共享全局的信息，那么在多代理架构中，当每一个图变成了一个节点，那么不同图之间的状态，应该怎么传递？
二、父子图状态传递
在上图所示的架构中，每个子代理（例如 Agent 1、Agent 2、Agent 3）由多个内部节点组成，各自通过独立的 State 维护内部节点间的消息传递。举个例子，假设 Agent 1 是一个数据分析师，负责接收并整理关键数据；Agent 2 则负责进行精细化的计算。在这个过程中，如何有效地在 Agent 1 和 Agent 2 之间传递数据成为了一个关键问题。
Agent 1 需要将数据传递给 Agent 2 进行处理，之后 Agent 2 需要将计算结果返回给 Agent 1。但是当前的涉及是不支持在代理间传递状态的，即在 Agent 2 的执行过程中无法访问 Agent 1 中的状态。
因此，如果想让不同 Agent 之间能够做到消息共享，则需要了解 LangGraph 框架中的 Subgraphs（子图）相关的概念和使用方法。
I、父图和子图有相同的键
# 1、配置模型
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
model_name = os.environ["OPENAI_MODEL_NAME"]
llm = ChatOpenAI(model=model_name, api_key=key, base_url=base_url, temperature=0, request_timeout=20)
# 测试大模型
print(f"{"测试大模型是否可用："}\n{'*'*80}\n{llm.invoke("你好，介绍你自己").content}\n{'*'*80}\n\n")
# Step2：父图状态定义 ParentState
from typing import TypedDict
# 定义父图中的状态
class ParentState(TypedDict):
user_input: str # 用来接收用户的输入
final_answer: str # 用来存储大模型针对用户输入的响应
# 父图是顶层主流程，承载两类全局状态：
# user_input：原始用户提问输入；
# final_answer：大模型初次生成的完整回答，会向下传递给子图作为共享状态。
# Step3：父图节点逻辑 parent_node
def parent_node(state: ParentState):
response = llm.invoke(state["user_input"])
return {"final_answer": response}
# 执行流程
# 读取状态里用户输入 user_input；
# 调用大模型生成完整回复；
# 将生成结果写入共享状态键 final_answer，供后续子图读取。
# Step4：子图状态 SubgraphState
# 定义子图中的状态
class SubgraphState(TypedDict):
# 这个 key 是和 父图（ParentState）共享的
final_answer: str
# 这个key 是 子图（subgraph）中独享的
summary_answer: str
# 状态设计规则
# 共享字段 final_answer：和父图状态同名，LangGraph 会自动在父子图之间同步该字段，实现数据跨图传递；
# 私有字段 summary_answer：仅在子图内部存在，用于存储摘要结果，不会向上同步到父图（隔离子图内部临时数据）。
# 子图业务目标
# 节点 1：压缩父图传来的完整回答为 50 字以内摘要；
# 节点 2：结合原文 + 摘要进行打分（1~10 分），并将分数覆盖回共享final_answer，传回父图。
# 四、子图双节点业务逻辑
# 子图节点 1：摘要节点 subgraph_node_1
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, AIMessage
def subgraph_node_1(state: SubgraphState):
system_prompt = """
Please summary the content you receive to 50 words or less
"""
messages = state['final_answer'] # 接收父图传递过来的完整响应
messages = [SystemMessage(content=system_prompt)] + [HumanMessage(content=messages.content)]
response = llm.invoke(messages)
return {"summary_answer": response}
# 子图节点 2：评分节点 subgraph_node_2
def subgraph_node_2(state: SubgraphState):
# final_answer 共享字段可直接读取父图传递的原文
messages = f"""
This is the full content of what you received: {state["final_answer"]} \n
This information is summarized for the full content:{state["summary_answer"]}
Please rate the text and summary information, returning a scale of 1 to 10. Note: Only the score value need return
"""
response = llm.invoke([HumanMessage(content=messages)])
# 更新共享状态键 final_answer，结果会回传给父图
return {"final_answer": response.content}
# Step6：构建并编译子图 Subgraph
from langgraph.graph import START, StateGraph
# 1. 初始化子图构造器，绑定子图状态 SubgraphState
subgraph_builder = StateGraph(SubgraphState)
# 2. 注册子图内两个业务节点
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
# 3. 定义边流转：起始点 → 摘要节点 → 评分节点
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
# 4. 编译生成可执行子图实例
subgraph = subgraph_builder.compile()
# Step7：构建父图，将子图作为节点嵌入父流程
# 1. 初始化父图构造器，绑定顶层状态 ParentState
builder = StateGraph(ParentState)
# 2. 添加父图原生节点
builder.add_node("node_1", parent_node)
# 3. 核心：把编译好的子图整体注册为父图的一个节点 node_2
builder.add_node("node_2", subgraph)
# 4. 父图流程流转：起始 → 生成节点 → 子图节点
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
# 5. 编译完整顶层主图
graph = builder.compile()
# 3、渲染流程图可视化
image_data = graph.get_graph().draw_mermaid_png()
with open("Same_Key_Main_Subgraph_G.png", "wb") as f:
f.write(image_data)
# LangGraph 父子图异步流式输出 astream
# async def stream_output():
# async for chunk in graph.astream(
# {"user_input": "我现在想学习大模型，应该关注哪些技术？"},
# stream_mode="values"
# ):
# print(chunk)
# import asyncio
# asyncio.run(stream_output())
async def stream_output2():
async for chunk in graph.astream(
{"user_input": "如何理解RAG？"},
stream_mode="values",
subgraphs=True
):
print(chunk)
import asyncio
asyncio.run(stream_output2())II、父图和子图没有相同的键
隔离式父子图（无共享状态）完整代码与设计解读
# 1、配置模型
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
model_name = os.environ["OPENAI_MODEL_NAME"]
llm = ChatOpenAI(model=model_name, api_key=key, base_url=base_url, temperature=0, request_timeout=40)
from typing import TypedDict
# 父图顶层状态
class ParentState(TypedDict):
user_input: str # 接收用户原始提问
final_answer: str # 存储大模型初次完整回答
# 父图生成节点
def parent_node_1(state: ParentState):
response = llm.invoke(state["user_input"])
return {"final_answer": response}
# 子图独立状态（完全隔离，无共享 key）
from langchain_core.messages import SystemMessage, HumanMessage
# 子图完全独立状态，无任何与父图同名字段
class SubgraphState(TypedDict):
response_answer: str # 接收父图传入的完整回答（子图私有）
summary_answer: str # 子图生成的摘要
score: str # 子图生成的打分结果
# 子图节点 1：摘要节点 subgraph_node_1
def subgraph_node_1(state: SubgraphState):
system_prompt = """
Please summary the content you receive to 50 words or less
"""
# 读取父图传入、存入子图私有字段 response_answer
messages = state['response_answer']
messages = [SystemMessage(content=system_prompt)] + [HumanMessage(content=messages.content)]
response = llm.invoke(messages)
return {"summary_answer": response}
# 子图节点 2：打分节点 subgraph_node_2
def subgraph_node_2(state: SubgraphState):
messages = f"""
This is the full content of what you received: {state["response_answer"]} \n
This information is summarized for the full content:{state["summary_answer"]}
Please rate the text and summary information, returning a scale of 1 to 10. Note: Only the score value need return
"""
response = llm.invoke([HumanMessage(content=messages)])
return {"score": response.content}
# 子图拥有完全独立状态，无任何与父图共享的 key，内部只做摘要、打分，不感知父图存在。
from langgraph.graph import START, StateGraph
# 1. 绑定子图独立状态 SubgraphState
subgraph_builder = StateGraph(SubgraphState)
# 2. 注册子图两个业务节点
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
# 3. 定义子图串行流转：起始→摘要节点→打分节点
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
# 4. 编译生成可执行子图实例
subgraph = subgraph_builder.compile()
# 核心中转节点 parent_node_2（父子状态转换器）
def parent_node_2(state: ParentState):
# 步骤1：父图状态 → 子图状态（手动映射字段）
response = subgraph.invoke({"response_answer": state["final_answer"]})
# 步骤2：子图输出 → 回写父图顶层状态
return {"final_answer": response["score"]}
# 绑定父图顶层状态 ParentState
builder = StateGraph(ParentState)
# 添加父图原生生成节点
builder.add_node("node_1", parent_node_1)
# 关键：父图节点不直接挂载subgraph，而是挂载中转函数parent_node_2
builder.add_node("node_2", parent_node_2)
# 定义父图流转链路
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
# 编译顶层完整父图
graph = builder.compile()
# image_data = graph.get_graph().draw_mermaid_png()
# with open("Different_Key_Main_Subgraph.png", "wb") as f:
# f.write(image_data)
# LangGraph 父子图异步流式输出 astream
all_chunk = []
async def stream_output():
async for chunk in graph.astream(
{"user_input": "我现在想学习大模型，应该关注哪些技术？"},
stream_mode="values"
):
all_chunk.append(chunk)
# print(chunk)
import asyncio
asyncio.run(stream_output())
print(all_chunk)
print(all_chunk[-1])
print(all_chunk[-1]["final_answer"])
# async def stream_output2():
# async for chunk in graph.astream(
# {"user_input": "如何理解RAG？"},
# stream_mode="values",
# subgraphs=True
# ):
# print(chunk)
# import asyncio
# asyncio.run(stream_output2())
三、基于 Network（网络）的多代理架构（用得少）
Single-Agent 可以在单个图结构运行工具，但即使使用像 gpt-4 这样的强大模型，当工具特别多的时候也会经常出现问题。处理复杂任务的一种方法是通过 “分而治之” 的方法：为每个任务或领域创建一个专门的代理，并将任务路由到正确的 “专家”。这是多代理网络架构的一个核心思想。那对于 Network（网络）代理来说，它的架构是：每个代理都可以与其他代理通信，且任何代理都可以决定接下来要呼叫哪个其他代理。 正如下图所示：
# 1、配置模型
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
model_name = os.environ["OPENAI_MODEL_NAME"]
coder_key = os.environ["MINIMAX_API_KEY"]
coder_base_url = os.environ["MINIMAX_BASE_URL"]
coder_model_name = os.environ["MINIMAX_MODEL_NAME"]
db_llm = ChatOpenAI(model=model_name, api_key=key, base_url=base_url, temperature=0, request_timeout=100)
coder_llm = ChatOpenAI(model=coder_model_name, api_key=coder_key, base_url=coder_base_url, temperature=0, request_timeout=100)
from sqlalchemy import create_engine, Column, Integer, String, Float, ForeignKey
from sqlalchemy.orm import sessionmaker, declarative_base
from faker import Faker
import random
# 创建基类
Base = declarative_base()
# 定义模型
class SalesData(Base):
__tablename__ = 'sales_data'
sales_id = Column(Integer, primary_key=True)
product_id = Column(Integer, ForeignKey('product_information.product_id'))
employee_id = Column(Integer) # 示例简化，未创建员工表
customer_id = Column(Integer, ForeignKey('customer_information.customer_id'))
sale_date = Column(String(50))
quantity = Column(Integer)
amount = Column(Float)
discount = Column(Float)
class CustomerInformation(Base):
__tablename__ = 'customer_information'
customer_id = Column(Integer, primary_key=True)
customer_name = Column(String(50))
contact_info = Column(String(50))
region = Column(String(50))
customer_type = Column(String(50))
class ProductInformation(Base):
__tablename__ = 'product_information'
product_id = Column(Integer, primary_key=True)
product_name = Column(String(50))
category = Column(String(50))
unit_price = Column(Float)
stock_level = Column(Integer)
class CompetitorAnalysis(Base):
__tablename__ = 'competitor_analysis'
competitor_id = Column(Integer, primary_key=True)
competitor_name = Column(String(50))
region = Column(String(50))
market_share = Column(Float)
# 数据库连接和表创建
from dotenv import load_dotenv
load_dotenv()
import os
passwrod = os.environ["MYSQL_PASSWORD"]
acc = os.environ["MYSQL_account"]
DATABASE_URI = f'mysql+pymysql://{acc}:{passwrod}@localhost/langgraph?charset=utf8mb4'
engine = create_engine(DATABASE_URI)
Base.metadata.create_all(engine)
# 接下来，创建模拟数据。我们使用 Python 的 faker 库来生成模拟数据，并构建插入数据的代码。(faker 是一个 Python 库，用于创建假数据，非常适合在开发和测试阶段填充数据库。)
# ! pip install faker
# 插入模拟数据
Session = sessionmaker(bind=engine)
session = Session()
fake = Faker()
# 生成客户信息
for _ in range(50): # 生成50个客户
customer = CustomerInformation(
customer_name=fake.name(),
contact_info=fake.phone_number(),
region=fake.state(), # 地区
customer_type=random.choice(['Retail', 'Wholesale']) # 零售、批发
)
session.add(customer)
# 生成产品信息
for _ in range(20): # 生成20种产品
product = ProductInformation(
product_name=fake.word(),
category=random.choice(['Electronics', 'Clothing', 'Furniture', 'Food', 'Toys']),
unit_price=random.uniform(10.0, 1000.0),
stock_level=random.randint(10, 100) # 库存
)
session.add(product)
# 生成竞争对手信息
for _ in range(10): # 生成10个竞争对手
competitor = CompetitorAnalysis(
competitor_name=fake.company(),
region=fake.state(),
market_share=random.uniform(0.01, 0.2) # 市场占有率
)
session.add(competitor)
# 提交事务（客户、产品、竞品）
session.commit()
# 生成销售数据，假设有100条销售记录
for _ in range(100):
sale = SalesData(
product_id=random.randint(1, 20),
employee_id=random.randint(1, 10), # 员工ID范围
customer_id=random.randint(1, 50),
sale_date=fake.date_between(start_date='-1y', end_date='today').strftime('%Y-%m-%d'),
quantity=random.randint(1, 10),
amount=random.uniform(50.0, 5000.0),
discount=random.uniform(0.0, 0.15)
)
session.add(sale)
session.commit()
# 关闭会话
session.close()
# 准备好数据以后，接下来给 db_agent 配置工具用来执行数据的增删改查操作，具体如下：
from pydantic import BaseModel, Field
from langchain_core.tools import tool
from typing import Union, Optional
# 新增销售记录参数模型
class AddSaleSchema(BaseModel):
product_id: int
employee_id: int
customer_id: int
sale_date: str
quantity: int
amount: float
discount: float
# 删除销售记录参数模型
class DeleteSaleSchema(BaseModel):
sales_id: int
# 更新销售记录参数模型
class UpdateSaleSchema(BaseModel):
sales_id: int
quantity: int
amount: float
# 查询销售记录参数模型
class QuerySalesSchema(BaseModel):
sales_id: int
# 1. 添加销售数据：
@tool(args_schema=AddSaleSchema)
def add_sale(product_id, employee_id, customer_id, sale_date, quantity, amount, discount):
"""Add sale record to the database."""
session = Session()
try:
new_sale = SalesData(
product_id=product_id,
employee_id=employee_id,
customer_id=customer_id,
sale_date=sale_date,
quantity=quantity,
amount=amount,
discount=discount
)
session.add(new_sale)
session.commit()
return {"messages": ["销售记录添加成功。"]}
except Exception as e:
return {"messages": [f"添加失败，错误原因：{e}"]}
finally:
session.close()
# 2. 删除销售数据
@tool(args_schema=DeleteSaleSchema)
def delete_sale(sales_id):
"""Delete sale record from the database."""
session = Session()
try:
sale_to_delete = session.query(SalesData).filter(SalesData.sales_id == sales_id).first()
if sale_to_delete:
session.delete(sale_to_delete)
session.commit()
return {"messages": ["销售记录删除成功。"]}
else:
return {"messages": [f"未找到销售记录ID: {sales_id}"]}
except Exception as e:
return {"messages": [f"删除失败，错误原因: {e}"]}
finally:
session.close()
# 3. 修改销售数据
@tool(args_schema=UpdateSaleSchema)
def update_sale(sales_id, quantity, amount):
"""Update sale record in the database."""
session = Session()
try:
sale_to_update = session.query(SalesData).filter(SalesData.sales_id == sales_id).first()
if sale_to_update:
sale_to_update.quantity = quantity
sale_to_update.amount = amount
session.commit()
return {"messages": ["销售记录更新成功。"]}
else:
return {"messages": [f"未找到销售记录ID: {sales_id}"]}
except Exception as e:
return {"messages": [f"更新失败，错误原因: {e}"]}
finally:
session.close()
# 4. 查询销售数据
@tool(args_schema=QuerySalesSchema)
def query_sales(sales_id):
"""Query sale record from the database."""
session = Session()
try:
sale_data = session.query(SalesData).filter(SalesData.sales_id == sales_id).first()
if sale_data:
return {
"sales_id": sale_data.sales_id,
"product_id": sale_data.product_id,
"employee_id": sale_data.employee_id,
"customer_id": sale_data.customer_id,
"sale_date": sale_data.sale_data,
"quantity": sale_data.quantity,
"amount": sale_data.amount,
"discount": sale_data.discount
}
else:
return {"messages": [f"未找到销售记录ID: {sales_id}。"]}
except Exception as e:
return {"messages": [f"查询失败，错误原因: {e}"]}
finally:
session.close()
# 然后，第二个数据分析师（code_agent）在需要的时候，接收 db_agent 的数据，生成可视化的图表，这里我们给他配置一个本地的 Python 代码解释器。这里我们使用 Python REPL 工具，它是 LangChain 封装的一个工具，作用是先让大模型生成代码，然后再运行该代码来获取答案，且仅返回打印的内容。使用的方式非常简单，代码如下：
from typing import Annotated
from langchain_core.tools import tool
from langchain_experimental.utilities import PythonREPL
import json
repl = PythonREPL()
# 在定义 python_repl 的地方（第 255 行附近），替换为：
@tool
def python_repl(
code: Annotated[str, "The python code to execute to generate your chart."],
):
"""Use this to execute python code. If you want to see the output of a value,
you should print it out with <code>print(...)</code>. This is visible to the user."""
try:
# --- 自动注入中文字体配置 ---
pre_code = """
import matplotlib
matplotlib.rcParams['font.sans-serif'] = ['SimHei', 'Microsoft YaHei', 'DejaVu Sans']
matplotlib.rcParams['axes.unicode_minus'] = False
"""
result = repl.run(pre_code + "\n" + code)
except BaseException as e:
return f"Failed to execute. Error: {repr(e)}"
result_str = f"Successfully executed:\n``<code>python\n{code}\n</code>``\nStdout: {result}"
return (
result_str + "\n\nIf you have completed all tasks, respond with FINAL ANSWER."
)
# 定义工具列表，并使用 ToolNode 进行构建。
from langgraph.prebuilt import ToolNode
# 定义工具列表
tools = [add_sale, delete_sale, update_sale, query_sales, python_repl]
tool_executor = ToolNode(tools)
# Stage3 创建代理 整体说明
# 要通过图状态进行通信，需要将单个代理定义为图节点，在图执行的每个步骤中，代理节点接收图的当前状态，执行代理代码，然后将更新的状态传递给下一个节点。代理节点是共享单个状态架构的。这里我们定义一个辅助函数 create_agent，用来帮助我们创建多代理系统中的每一个子代理。每个子代理可以通过 llm 参数定义其使用的大模型，system_message 用于定义其每个代理独有的背景信息，而 tools 则可以给不同的子代理绑定特有的工具。具体函数代码如下：
from langchain_core.messages import BaseMessage, HumanMessage, ToolMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
def create_agent(llm, tools, system_message: str):
"""Create an agent."""
prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"You are a helpful AI assistant, collaborating with other assistants."
" Use the provided tools to progress towards answering the question."
" If you are unable to fully answer, that's OK, another assistant with different tools "
" will help where you left off. Execute what you can to make progress."
" If you or any of the other assistants have the final answer or deliverable,"
" prefix your response with FINAL ANSWER so the team knows to stop."
" You have access to the following tools: {tool_names}.\n{system_message}",
),
MessagesPlaceholder(variable_name="messages"),
]
)
prompt = prompt.partial(system_message=system_message)
prompt = prompt.partial(tool_names=", ".join([tool.name for tool in tools]))
return prompt | llm.bind_tools(tools)
# Prompt 中文释义
"""你是一个有帮助的人工智能助手，与其他助手合作。使用提供的工具来推进解答问题的过程。
如果你不能完全回答，没关系，另一个拥有不同工具的助手会接着帮忙。
尽你所能执行任务以取得进展。如果你或其他任何助手得到了最终答案或成果，请在你的回答前加上“最终答案”，以便团队知道可以停止。
你可以使用以下工具：{tool_names}。{system_message}。"""
# 数据库管理员 db_agent
db_agent = create_agent(
db_llm,
[add_sale, delete_sale, update_sale, query_sales],
system_message="You should provide accurate data for the code_generator to use."
)
# 数据分析师 code_agent
code_agent = create_agent(
coder_llm,
[python_repl],
system_message="Run python code to display diagrams or output execution results"
)
# 接下来，将 Agent 定义成节点，如下代码所示：
import functools
from langchain_core.messages import AIMessage, ToolMessage
def agent_node(state, agent, name):
# 将父图状态传入代理执行
result = agent.invoke(state)
# 区分工具消息与普通AI回复消息
if isinstance(result, ToolMessage):
pass
else:
# 封装为带代理名称的AIMessage存入全局状态
result = AIMessage(**result.dict(exclude={"type", "name"}), name=name)
return {
"messages": [result],
# 记录发送方，用于工具执行后路由回原代理
"sender": name,
}
# 固定参数生成两个专用节点
db_node = functools.partial(agent_node, agent=db_agent, name="db_manager")
code_node = functools.partial(agent_node, agent=code_agent, name="code_generator")
# Stage4：路由函数 router
"""文档说明
在多代理系统中，路由器的功能是根据从代理收到的消息确定接下来要执行的流程，用于路由调用工具、结束流程或继续下一步，具体取决于消息的内容和代理做出的决策。如下所示的 router 函数，将图的当前状态作为输入，从状态中提取消息，重点关注列表中的最后一条消息。
如果最后一条消息包含 tool_calls → 路由到 call_tool 工具执行节点；
如果消息包含 FINAL ANSWER → 路由至 END 终止流程；
其余情况返回 continue，流转至另一协作代理。"""
from typing import Literal
from langgraph.graph import END
def router(state) -> Literal["call_tool", "continue", "__end__"]:
messages = state["messages"]
last_message = messages[-1]
# 代理需要调用工具
if last_message.tool_calls:
return "call_tool"
# 任务完成，终止图
if "FINAL ANSWER" in last_message.content:
return END
# 切换另一代理协作
return "continue"
# Stage5：定义状态结构 AgentState
# 在初始化 StateGraph 对象之前，我们首先定义 AgentState。AgentState 定义在图中的节点之间传递的对象的结构。在这里，它包括消息列表和要执行任务的代理名称 (sender)。
import operator
from typing import Annotated, Sequence
from typing_extensions import TypedDict
from langchain_core.messages import BaseMessage
class AgentState(TypedDict):
# messages：消息序列，operator.add 实现消息自动追加
messages: Annotated[Sequence[BaseMessage], operator.add]
# sender：记录上一步执行的代理名称
sender: str
# Stage5 续：构建 StateGraph 工作流
# 一切准备就绪后，构建管理多智能体工作流的流程。多代理架构支持任意代理间互相连通，通过条件边定义节点跳转逻辑。
from langgraph.graph import END, StateGraph
# 1. 初始化状态图，绑定自定义状态
workflow = StateGraph(AgentState)
# 2. 注册所有节点
workflow.add_node("db_manager", db_node)
workflow.add_node("code_generator", code_node)
workflow.add_node("call_tool", tool_executor)
# 3. 条件边1：db_manager 执行后走router路由
workflow.add_conditional_edges(
"db_manager",
router,
{"continue": "code_generator", "call_tool": "call_tool", END: END},
)
# 4. 条件边2：code_generator 执行后走router路由
workflow.add_conditional_edges(
"code_generator",
router,
{"continue": "db_manager", "call_tool": "call_tool", END: END},
)
# 5. 条件边3：工具执行完成后，返回调用工具的原代理
workflow.add_conditional_edges(
"call_tool",
lambda x: x["sender"],
{"db_manager": "db_manager", "code_generator": "code_generator"},
)
# 6. 设置图入口节点：初始先走数据库代理
workflow.set_entry_point("db_manager")
# 7. 编译生成可运行图实例
graph = workflow.compile()
# 3、渲染流程图可视化
# image_data = graph.get_graph().draw_mermaid_png()
# with open("Network-Architectures.png", "wb") as f:
# f.write(image_data)
# Stage6 调用测试
# for chunk in graph.stream(
# {"messages": [HumanMessage(content="根据sales_id使用折线图显示前5名销售的销售总额")]},
# {"recursion_limit": 50},
# stream_mode='values'
# ):
# print(chunk)
# 用例 2：删除指定销售数据（单数据库操作）
# for chunk in graph.stream(
# {"messages": [HumanMessage(content="帮我删除销售id 是 20 的这名销售信息")]},
# {"recursion_limit": 20},
# stream_mode='values'
# ):
# print(chunk)
# 用例 3：TOP10 销售额柱状图（美化输出）
for chunk in graph.stream(
{"messages": [HumanMessage(content="帮我根据前10名的销售记录id，生成对应的销售额柱状图")]},
{"recursion_limit": 20},
stream_mode='values'
):
chunk["messages"][-1].pretty_print()
四、 Supervisor 架构介绍与基本构建原理
正如 Magentic-One 的内部结构一样，LangGraph 中的 Supervisor 充当多代理工作流程中的中央控制器，协调各个代理之间的通信和任务分配。它的工作原理是接收一个代理的输出，解释这些消息，然后相应地指导任务流程。它在 LangGraph 中基于图结构中的节点实现，允许随着任务的发展或新代理的集成而动态交互和灵活调整工作流程，从而优化流程的有效性和速度。其结构如下图所示：
I、将Agent定义为Node，进行代码实现
# 1、配置模型
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
model_name = os.environ["OPENAI_MODEL_NAME"]
llm = ChatOpenAI(model=model_name, api_key=key, base_url=base_url, temperature=0, request_timeout=110)
# 接下来创建代理主管。需要利用 LangGraph 的 StateGraph、AgentState 等状态模式来定义 Supervisor 节点的行为和决策逻辑。
from langgraph.graph import StateGraph, MessagesState, START, END
class AgentState(MessagesState):
next: str
# 然后去设置代理主管可以管理的子代理，添加 FINISH 是为了用来标识任务是否已经全部完成，可以返回最终的结果了。这就与 NetWork 网络代理不同了，NetWork 网络代理是每一个子代理节点都可以决定是否直接返回 END，而 supervisor 则是由主管代理节点做一切的决策，这包括是否继续执行，还是结束图的运行状态
members = ["chat", "coder", "sqler"]
options = members + ["FINISH"]
print(options)
# 接下来定义主管节点。主管节点常见的模式是接收状态模式中的相关数据，让大模型根据实时的任务进展自主决定下一步调用哪个代理，并通过结构化输出（例如，强制它返回带有 "next_agent" 字段），以维持图完整的运行状态，直至输出 end，相关代码如下图所示:
from typing import Literal
from typing_extensions import TypedDict
# class Router(TypedDict):
# """Worker to route to next. If no workers needed, route to FINISH"""
# next: Literal[*options]
class Router(TypedDict):
"""Worker to route to next. If no workers needed, route to FINISH"""
next: Literal["chat", "coder", "sqler", "FINISH"]
# Literal 是 Python 的 typing 模块中的一个类型，用于定义一个变量的具体值的类型限制。当使用 Literal 时，实际上是在告诉 Python，变量的值必须是指定的几个值中的一个。而 next: Literal["chat", "coder", "sqler"] 意味着 next 属性只能赋予三个字符串值之一: "chat"、"coder"、"sqler" 或 "FINISH"，分别用来表示使用哪一个子代理来执行任务，或者直接通过 END 结束当前的图。
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, ToolMessage
def supervisor(state: AgentState):
system_prompt = (
"You are a supervisor tasked with managing a conversation between the"
f" following workers: {members}.\n\n"
"Each worker has a specific role:\n"
"- chat: Responds directly to user inputs using natural language.\n"
"- coder: Activated for tasks that require mathematical calculations or specific coding needs.\n"
"- sqler: Used when database queries or explicit SQL generation is needed.\n\n"
"Given the following user request, respond with the worker to act next."
" Each worker will perform a task and respond with their results and status."
" When finished, respond with FINISH."
)
messages = [{"role": "system", "content": system_prompt},] + state["messages"]
response = llm.with_structured_output(Router).invoke(messages)
next_ = response["next"]
if next_ == "FINISH":
next_ = END
return {"next": next_}
members
# 接下来依次子代理，每个代理通过 Node 的形式来定义。关键在于：每个子代理节点在执行完内部逻辑最后，在更新全局状态模式的时候，要通过添加 name = 代理名称 的方式告诉 supervisor 代理，该信息是哪个子代理返回的数据。要与 members 中的定义保持一致。代码如下所示：
def chat(state: AgentState):
messages = state["messages"][-1]
model_response = llm.invoke(messages.content)
final_response = [HumanMessage(content=model_response.content, name="chat")] # 这里要添加名称
return {"messages": final_response}
def coder(state: AgentState):
messages = state["messages"][-1]
model_response = llm.invoke(messages.content)
final_response = [HumanMessage(content=model_response.content, name="coder")] # 这里要添加名称
return {"messages": final_response}
def sqler(state: AgentState):
messages = state["messages"][-1]
model_response = llm.invoke(messages.content)
final_response = [HumanMessage(content=model_response.content, name="sqler")] # 这里要添加名称
return {"messages": final_response}
# 添加所有节点
builder = StateGraph(AgentState)
# builder.add_edge(START, "supervisor")
builder.add_node("supervisor", supervisor)
builder.add_node("chat", chat)
builder.add_node("coder", coder)
builder.add_node("sqler", sqler)
# 子代理指向主管的边
# 然后让每个子代理在完成工作后总是向主管 “汇报”，即需要构建它们之间的边。如下所示：
for member in members:
# 我们希望我们的工人在完成工作后总是向主管“汇报”
builder.add_edge(member, "supervisor")
# 替代 add_conditional_edges，为每个 worker 手动添加实线边
# for member in members:
# builder.add_edge("supervisor", member) # supervisor → worker
# builder.add_edge(member, "supervisor") # worker → supervisor
# 条件路由与编译图
# 然后在图状态中填充 next 字段，路由到具体的某个节点或者结束图的运行，从而指定如何执行接下来的任务。
builder.add_conditional_edges("supervisor", lambda state: state["next"])
# 添加开始和节点
builder.add_edge(START, "supervisor")
# 编译图
graph = builder.compile()
# 3、渲染流程图可视化
image_data = graph.get_graph(xray=True).draw_mermaid_png()
with open("Supervisor-Architectures-g.png", "wb") as f:
f.write(image_data)
# 运行测试代码
# 测试 1：通用自然语言问答
# for chunk in graph.stream({"messages": "你好，什么是机器学习？"}, stream_mode="values"):
# print(chunk)
# 测试 2：代码生成任务
# for chunk in graph.stream({"messages": "你好，帮我生成一个二分查找的Python代码"}, stream_mode="values"):
# print(chunk)
# 测试 3：数据库查询任务（完整存储输出）
# all_chunk = []
# for chunk in graph.stream({"messages": "我想查询数据库中 data 表的所有数据，"}, stream_mode="values"):
# all_chunk.append(chunk)
# all_chunk[-1]['messages'][-1].content
II、实现Supervisor多智能体架构的实例
# 依赖导入 & ORM 基类初始化
from sqlalchemy import create_engine, Column, Integer, String, Float, ForeignKey
from sqlalchemy.orm import sessionmaker, declarative_base
from faker import Faker
import random
# 创建基类
Base = declarative_base()
# 定义模型
class SalesData(Base):
__tablename__ = 'sales_data'
sales_id = Column(Integer, primary_key=True)
product_id = Column(Integer, ForeignKey('product_information.product_id'))
employee_id = Column(Integer) # 示例简化，未创建员工表
customer_id = Column(Integer, ForeignKey('customer_information.customer_id'))
sale_date = Column(String(50))
quantity = Column(Integer)
amount = Column(Float)
discount = Column(Float)
class CustomerInformation(Base):
__tablename__ = 'customer_information'
customer_id = Column(Integer, primary_key=True)
customer_name = Column(String(50))
contact_info = Column(String(50))
region = Column(String(50))
customer_type = Column(String(50))
class ProductInformation(Base):
__tablename__ = 'product_information'
product_id = Column(Integer, primary_key=True)
product_name = Column(String(50))
category = Column(String(50))
unit_price = Column(Float)
stock_level = Column(Integer)
class CompetitorAnalysis(Base):
__tablename__ = 'competitor_analysis'
competitor_id = Column(Integer, primary_key=True)
competitor_name = Column(String(50))
region = Column(String(50))
market_share = Column(Float)
# 2、读取配置信息
from dotenv import load_dotenv
load_dotenv()
import os
passwrod = os.environ["MYSQL_PASSWORD"]
acc = os.environ["MYSQL_account"]
DATABASE_URI = f'mysql+pymysql://{acc}:{passwrod}@localhost/langgraph?charset=utf8mb4'
engine = create_engine(DATABASE_URI)
Session = sessionmaker(bind=engine)
session = Session()
from pydantic import BaseModel, Field
from langchain_core.tools import tool
from typing import Union, Optional
class AddSaleSchema(BaseModel):
product_id: int
employee_id: int
customer_id: int
sale_date: str
quantity: int
amount: float
discount: float
class DeleteSaleSchema(BaseModel):
sales_id: int
class UpdateSaleSchema(BaseModel):
sales_id: int
quantity: int
amount: float
class QuerySalesSchema(BaseModel):
sales_id: int
# 1. 添加销售数据：
@tool(args_schema=AddSaleSchema)
def add_sale(product_id, employee_id, customer_id, sale_date, quantity, amount, discount):
"""Add sale record to the database."""
session = Session()
try:
new_sale = SalesData(
product_id=product_id,
employee_id=employee_id,
customer_id=customer_id,
sale_date=sale_date,
quantity=quantity,
amount=amount,
discount=discount
)
session.add(new_sale)
session.commit()
return {"messages": ["销售记录添加成功。"]}
except Exception as e:
return {"messages": [f"添加失败，错误原因：{e}"]}
finally:
session.close()
# 2. 删除销售数据
@tool(args_schema=DeleteSaleSchema)
def delete_sale(sales_id):
"""Delete sale record from the database."""
session = Session()
try:
sale_to_delete = session.query(SalesData).filter(SalesData.sales_id == sales_id).first()
if sale_to_delete:
session.delete(sale_to_delete)
session.commit()
return {"messages": ["销售记录删除成功。"]}
else:
return {"messages": [f"未找到销售记录ID: {sales_id}"]}
except Exception as e:
return {"messages": [f"删除失败，错误原因: {e}"]}
finally:
session.close()
# 3. 修改销售数据
@tool(args_schema=UpdateSaleSchema)
def update_sale(sales_id, quantity, amount):
"""Update sale record in the database."""
session = Session()
try:
sale_to_update = session.query(SalesData).filter(SalesData.sales_id == sales_id).first()
if sale_to_update:
sale_to_update.quantity = quantity
sale_to_update.amount = amount
session.commit()
return {"messages": ["销售记录更新成功。"]}
else:
return {"messages": [f"未找到销售记录ID: {sales_id}"]}
except Exception as e:
return {"messages": [f"更新失败，错误原因: {e}"]}
finally:
session.close()
# 4. 查询销售数据
@tool(args_schema=QuerySalesSchema)
def query_sales(sales_id):
"""Query sale record from the database."""
session = Session()
try:
sale_data = session.query(SalesData).filter(SalesData.sales_id == sales_id).first()
if sale_data:
return {
"sales_id": sale_data.sales_id,
"product_id": sale_data.product_id,
"employee_id": sale_data.employee_id,
"customer_id": sale_data.customer_id,
"sale_date": sale_data.sale_date,
"quantity": sale_data.quantity,
"amount": sale_data.amount,
"discount": sale_data.discount
}
else:
return {"messages": [f"未找到销售记录ID: {sales_id}。"]}
except Exception as e:
return {"messages": [f"查询失败，错误原因: {e}"]}
finally:
session.close()
from typing import Annotated
from langchain_core.tools import tool
from langchain_experimental.utilities import PythonREPL
import json
repl = PythonREPL()
@tool
def python_repl(
code: Annotated[str, "The python code to execute to generate your chart."],
):
"""Use this to execute python code. If you want to see the output of a value,
you should print it out with <code>print(...)</code>. This is visible to the user."""
try:
result = repl.run(code)
except BaseException as e:
return f"Failed to execute. Error: {repr(e)}"
result_str = f"Successfully executed:\n``<code>python\n{code}\n</code>``\nStdout: {result}"
return result_str
# 使用 create_react_agent 构建成两个 ReAct 代理。
from langchain_core.messages import HumanMessage
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import create_react_agent
# 1、配置模型
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
model_name = os.environ["OPENAI_MODEL_NAME"]
coder_key = os.environ["MINIMAX_API_KEY"]
coder_base_url = os.environ["MINIMAX_BASE_URL"]
coder_model_name = os.environ["MINIMAX_MODEL_NAME"]
db_llm = ChatOpenAI(model=model_name, api_key=key, base_url=base_url, temperature=0, request_timeout=100)
coder_llm = ChatOpenAI(model=coder_model_name, api_key=coder_key, base_url=coder_base_url, temperature=0, request_timeout=100)
db_agent = create_react_agent(
db_llm,
tools=[add_sale, delete_sale, update_sale, query_sales],
prompt="You use to perform database operations while should provide accurate data for the code_gen" # ✅
)
code_agent = create_react_agent(
coder_llm,
tools=[python_repl],
prompt="Run python code to display diagrams or output execution results" # ✅
)
from langgraph.graph import StateGraph, MessagesState, START, END
class AgentState(MessagesState):
next: str
# 然后分别将两个 ReAct Agent 构造成节点，并添加代理名称标识。
def db_node(state: AgentState):
result = db_agent.invoke(state)
return {
"messages": [
HumanMessage(content=result["messages"][-1].content, name="sqler")
]
}
def code_node(state: AgentState):
result = code_agent.invoke(state)
return {
"messages": [HumanMessage(content=result["messages"][-1].content, name="coder")]
}
# 然后去设置代理主管可以管理的子代理，添加 FINISH 是为了用来标识任务是否已经全部完成，可以返回最终的结果了。
members = ["chat", "coder", "sqler"]
options = members + ["FINISH"]
from typing import Literal
from typing_extensions import TypedDict
class Router(TypedDict):
"""Worker to route to next. If no workers needed, route to FINISH"""
next: Literal["chat", "coder", "sqler", "FINISH"]
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, ToolMessage
def supervisor(state: AgentState):
system_prompt = (
"You are a supervisor tasked with managing a conversation between the"
f" following workers: {members}.\n\n"
"Each worker has a specific role:\n"
"- chat: Responds directly to user inputs using natural language.\n"
"Given the following user request, respond with the worker to act next."
" Each worker will perform a task and respond with their results and status."
" When finished, respond with FINISH."
)
messages = [{"role": "system", "content": system_prompt},] + state["messages"]
response = db_llm.with_structured_output(Router).invoke(messages)
next_ = response["next"]
if next_ == "FINISH":
next_ = END
return {"next": next_}
# 接下来正常构建 Chat 子代理，通过 Node 的形式来定义。
def chat(state: AgentState):
messages = state["messages"][-1]
model_response = db_llm.invoke(messages.content)
final_response = [HumanMessage(content=model_response.content, name="chatbot")]
return {"messages": final_response}
builder = StateGraph(AgentState)
builder.add_node("supervisor", supervisor)
builder.add_node("chat", chat)
builder.add_node("coder", code_node)
builder.add_node("sqler", db_node)
for member in members:
# 每个子代理在完成工作后总是向主管“汇报”
builder.add_edge(member, "supervisor")
builder.add_conditional_edges("supervisor", lambda state: state["next"])
builder.add_edge(START, "supervisor")
graph = builder.compile()
# image_data = graph.get_graph(xray=True).draw_mermaid_png()
# with open("Router_Agent_G.png", "wb") as f:
# f.write(image_data)
# 接下来进行问答测试：
for chunk in graph.stream({"messages": "帮我查询前3个销售记录的具体信息"}, stream_mode="values"):
print(chunk)
# for chunk in graph.stream({"messages": "帮我根据前10名的销售记录id，生成对应的销售额柱状图"}, stream_mode="values"):
# print(chunk)
# for chunk in graph.stream({"messages": "你好，请你介绍一下机器学习"}, stream_mode="values"):
# chunk["messages"][-1].pretty_print()
# for chunk in graph.stream({"messages": "帮我删除第33条销售数据"}, stream_mode="values"):
# chunk["messages"][-1].pretty_print()
五、Multi-Agent 实现混合多知识库检索（GraphRAG）
I、配置 Neo4j 图数据库实例
首先，登录该地址注册免费账户：
https://console.neo4j.io/?product=aura-db&tenant=2afe251b-59ae-5517-9598-84fc5d57b0b5
1、首次登陆，需注册（谷歌账号最好）：
2、记住保存username和password，方便后续连接
3、创建一个instance（选择免费的）
4、点击inspect，生成一个远程连接的url，连接图数据库使用
5、左边菜单栏，选择Get started页面，点击连接到实例
6、复制数据库的名字到代码中
II、创建图索引及构建 GraphRAG Agent
在将数据引入 Neo4j 图数据库之前，一般来说我们会将数据（例如 .txt 或 .csv 文件）拆分为可管理的块。可以使用 LangChain 框架的 TextLoader 和 RecursiveCharacterTextSplitter 来完成。用于演示目的，我们此处不进行分块，将完整的文档作为一个 TextUnit 进行图属性的提取。
# 打开文件，并赋予读取模式 'r'
with open('D:\projects\code\class-study\GraphRAG-Conception\comany.txt', 'r', encoding="utf-8") as file:
# 读取文件的全部内容
content = file.read()
print(content)
# 转化成 Document 对象。
from langchain_core.documents import Document
documents = [Document(page_content=content)]
print(documents)
# 准备好数据后，我们可以使用 langchain_experimental.graph_transformers 中的 LLMGraphTransformer 将其摄取到 Neo4j 中。该工具会自动将文档转换为图格式。LLMGraphTransformer 能够以两种完全独立的模式运行：
# Tool-Based 模式（默认）：当使用的大模型支持结构化输出或函数调用时，该模式利用内置的 with_structured_output 来使用工具。工具规范定义了输出格式，确保以结构化、预定义的方式提取实体和关系。
# 基于提示的模式（回退）：在使用的大模型不支持工具或函数调用的情况下，该转换器回退到纯粹提示驱动的方法。该模式使用 few-shot 提示来定义输出格式，指导大模型以基于文本的方式提取实体和关系。然后通过自定义函数解析结果，该函数将大模型的输出转换为 JSON 格式。该 JSON 用于填充节点和关系。
# pip install neo4j
# GraphRAG Setup
from langchain_community.graphs import Neo4jGraph
from langchain_experimental.graph_transformers import LLMGraphTransformer
from langchain_core.documents import Document
from langchain_openai import ChatOpenAI
# 1、配置模型
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
model_name = os.environ["OPENAI_MODEL_NAME"]
graph_llm = ChatOpenAI(model=model_name, api_key=key, base_url=base_url, temperature=0, request_timeout=110)
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
model_name = os.environ["OPENAI_MODEL_NAME"]
NEO4J_URL=os.environ["NEO4J_URL"]
NEO4J_USERNAME=os.environ["NEO4J_USERNAME"]
NEO4J_PASSWORD=os.environ["NEO4J_PASSWORD"]
NEO4J_DATABASE=os.environ["NEO4J_DATABASE"]
# 创建图数据库示例
graph = Neo4jGraph(
url=NEO4J_URL, # 替换为自己的
username=NEO4J_USERNAME, # 替换为自己的
password=NEO4J_PASSWORD, #替换为自己的
database=NEO4J_DATABASE # 替换为自己的
)
# 使用 convert_to_graph_documents 函数处理文档，进行实体和关系的提取。可以自定义，当 allowed_nodes 和 allowed_relationships 不手动填写的时候，它会自动判断并进行提取。
# 图转换器配置
# 方式1：限定允许的节点类型与关系类型（推荐，约束提取范围，减少噪声）
graph_transformer = LLMGraphTransformer(
llm=graph_llm,
)
# 方式2：不做约束，由LLM自由识别实体与关系（上面实例会被覆盖）
graph_transformer = LLMGraphTransformer(llm=graph_llm)
# 执行实体、关系抽取，生成GraphDocument对象
graph_documents = graph_transformer.convert_to_graph_documents(documents)
# 将抽取得到的节点、关系批量写入Neo4j数据库
graph.add_graph_documents(graph_documents)
# 打印调试信息，查看抽取结果
print(f"Graph documents: {len(graph_documents)}")
print(f"Nodes from 1st graph doc:{graph_documents[0].nodes}")
print(f"Relationships from 1st graph doc:{graph_documents[0].relationships}")
III、查看网页端的neo4j图谱的抽取
1、执行完代码后，等2分钟，刷新neo4j网页即可
2、选择query页面，查看知识图谱
IV、用cypher语句查询图数据
sql语句示例：
# 打开文件，并赋予读取模式 'r'
with open('D:\projects\code\class-study\GraphRAG-Conception\comany.txt', 'r', encoding="utf-8") as file:
# 读取文件的全部内容
content = file.read()
print(content)
# 转化成 Document 对象。
from langchain_core.documents import Document
documents = [Document(page_content=content)]
print(documents)
# 准备好数据后，我们可以使用 langchain_experimental.graph_transformers 中的 LLMGraphTransformer 将其摄取到 Neo4j 中。该工具会自动将文档转换为图格式。LLMGraphTransformer 能够以两种完全独立的模式运行：
# Tool-Based 模式（默认）：当使用的大模型支持结构化输出或函数调用时，该模式利用内置的 with_structured_output 来使用工具。工具规范定义了输出格式，确保以结构化、预定义的方式提取实体和关系。
# 基于提示的模式（回退）：在使用的大模型不支持工具或函数调用的情况下，该转换器回退到纯粹提示驱动的方法。该模式使用 few-shot 提示来定义输出格式，指导大模型以基于文本的方式提取实体和关系。然后通过自定义函数解析结果，该函数将大模型的输出转换为 JSON 格式。该 JSON 用于填充节点和关系。
# pip install neo4j
# GraphRAG Setup
from langchain_community.graphs import Neo4jGraph
from langchain_experimental.graph_transformers import LLMGraphTransformer
from langchain_core.documents import Document
from langchain_openai import ChatOpenAI
# 1、配置模型
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
model_name = os.environ["OPENAI_MODEL_NAME"]
graph_llm = ChatOpenAI(model=model_name, api_key=key, base_url=base_url, temperature=0, request_timeout=110)
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
model_name = os.environ["OPENAI_MODEL_NAME"]
NEO4J_URL=os.environ["NEO4J_URL"]
NEO4J_USERNAME=os.environ["NEO4J_USERNAME"]
NEO4J_PASSWORD=os.environ["NEO4J_PASSWORD"]
NEO4J_DATABASE=os.environ["NEO4J_DATABASE"]
# 创建图数据库示例
graph = Neo4jGraph(
url=NEO4J_URL, # 替换为自己的
username=NEO4J_USERNAME, # 替换为自己的
password=NEO4J_PASSWORD, #替换为自己的
database=NEO4J_DATABASE # 替换为自己的
)
from langchain_community.chains.graph_qa.cypher import GraphCypherQAChain
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(temperature=0, model_name="gpt-4o-mini", api_key=key,base_url=base_url)
cypher_chain = GraphCypherQAChain.from_llm(
graph=graph,
cypher_llm=llm,
qa_llm=llm,
validate_cypher=True, # Validate relationship directions
verbose=True,
allow_dangerous_requests=True
)
# 测试提问示例
result = cypher_chain.invoke("九纹龙是谁？")
print(f"{"测试知识图谱效果1："}\n{'*'*80}\n{result}\n{'*'*80}\n\n")
result = cypher_chain.invoke("高俅跟哪些人有关系？")
print(f"{"测试知识图谱效果2："}\n{'*'*80}\n{result}\n{'*'*80}\n\n")
result = cypher_chain.invoke("都有哪些人在我的数据库中？")
print(f"{"测试知识图谱效果3："}\n{'*'*80}\n{result}\n{'*'*80}\n\n")
# 将 GraphRAG 封装为 LangGraph 多智能体节点
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_core.messages import HumanMessage
class AgentState(MessagesState):
next: str
def graph_kg(state: AgentState):
messages = state["messages"][-1]
cypher_chain = GraphCypherQAChain.from_llm(
graph=graph,
cypher_llm=llm,
qa_llm=llm,
validate_cypher=True,
allow_dangerous_requests=True
)
response = cypher_chain.invoke(messages.content)
final_response = [HumanMessage(content=response["result"], name="graph_kg")] # 添加节点名称标识
return {"messages": final_response}
V、 创建传统 RAG Agent
这里我们同样使用免费的在线 Milvus 实例，地址如下：
https://cloud.zilliz.com/login?redirect=/orgs
Zilliz Cloud 是 Milvus 官方托管云服务，提供免费实例，用于存放文本向量，实现传统语义检索 RAG。
1、先注册登录。
1、github账号登录注册
相比 Google，GitHub 在国内的访问稳定性要好得多，授权流程通常不会遇到网络问题。通过GitHub账号这种方式，可以成功创建一个 Zilliz Cloud 账户。
2、邮箱获取验证码
3、选择国内区域：
成功登录后，在创建免费集群时，请务必注意选择部署区域 (Region)。
选择国内节点：在创建集群的配置页面，请选择位于中国的云服务商节点，例如 腾讯云 (Tencent Cloud) 或 亚马逊云科技宁夏区 (AWS Ningxia)。
优势：选择国内节点可以确保你后续在本地运行 RAG 项目代码时，连接数据库的网络延迟最低，体验最流畅。
Zilliz Cloud 提供的免费 Serverless 计划（包含 5GB 存储空间）完全足够进行 RAG 项目的学习和原型开发。
4、创建索引：
5、选择免费套餐
6、保存好用户名和密码，用于后续连接使用
7、拿到URL和ID
8、从数据库（clusters）中建表（collection）
9、数据切片成功后，可以到网页上查看
2、代码实例
# 打开文件，并赋予读取模式 'r'
with open('D:\projects\code\class-study\GraphRAG-Conception\comany.txt', 'r', encoding="utf-8") as file:
# 读取文件的全部内容
content = file.read()
print(content)
# 转化成 Document 对象。
from langchain_core.documents import Document
documents = [Document(page_content=content)]
print(documents)
# 1、配置模型
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
model_name = os.environ["OPENAI_MODEL_NAME"]
graph_llm = ChatOpenAI(model=model_name, api_key=key, base_url=base_url, temperature=0, request_timeout=100)
embedding_key = os.environ["EMBEDDING_KEY"]
embedding_url = os.environ["EMBEDDING_URL"]
embedding_name = os.environ["EMBEDDING_NAME"]
milvus_password = os.environ["MILVUS_PASSWORD"]
milvus_url = os.environ["MILVUS_URL"]
milvus_user = os.environ["MILVUS_USER"]
from langchain_text_splitters import RecursiveCharacterTextSplitter
chunk_size = 800
chunk_overlap = 100
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size, chunk_overlap=chunk_overlap
)
# 执行文档切分
splits = text_splitter.split_documents(documents)
splits
from langchain_openai import OpenAIEmbeddings
embeddings = OpenAIEmbeddings(
model=embedding_name,
base_url=embedding_url,
api_key=embedding_key,
# dimensions=4096, # 如需指定维度，取消注释
)
from pymilvus import connections, utility
# 先连接
connections.connect(
alias="default",
uri= milvus_url,
user= milvus_user,
password= milvus_password,
)
# 如果切块的方式、大小有调整，就必须先删除旧的集合，否则切块内容不会变
# 删除旧集合
if utility.has_collection("company_rag_milvus"):
utility.drop_collection("company_rag_milvus")
print("已删除旧集合")
# 然后再创建
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import WebBaseLoader
from langchain_milvus import Milvus
vectorstore = Milvus.from_documents(
documents=splits,
collection_name="company_rag_milvus",
embedding=embeddings,
connection_args={
"uri": milvus_url,
"user": milvus_user,
"password": milvus_password,
}
)
# 同时可以登录网页端进行确认：
from langchain_core.prompts import PromptTemplate
# from langchain import hub
from langchain_core.output_parsers import StrOutputParser
# 提示
prompt = PromptTemplate(
template="""You are an assistant for question-answering tasks.
Use the following pieces of retrieved context to answer the question. If you don't know the answer, just say that you don't know.
Use three sentences maximum and keep the answer concise:
Question: {question}
Context: {context}
Answer:
""",
input_variables=["question", "context"],
)
# 运行
question = "史进的外号是什么？"
# 构建检索器
retriever = vectorstore.as_retriever(
search_type="mmr",
search_kwargs={"k": 2, "fetch_k": 15}
)
# retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
# 执行检索
docs = retriever.invoke(question)
print(f"{"进行一般对话响应："}\n{'*'*80}\n{docs}\n{'*'*80}\n\n")
# 运行 RAG_chain , 生成最终的回复。
# 构建传统的RAG Chain
rag_chain = prompt | graph_llm | StrOutputParser()
generation = rag_chain.invoke({"context": docs, "question": question})
print(f"{"答案："}\n{'*'*80}\n{generation}\n{'*'*80}\n\n")
# 构建混合知识库检索多代理系统
from langgraph.graph import StateGraph, MessagesState, START, END
# 同时可以登录网页端进行确认：
from langchain_core.prompts import PromptTemplate
# from langchain import hub
from langchain_core.output_parsers import StrOutputParser
class AgentState(MessagesState):
next: str
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, ToolMessage
# 构建传统 RAG 的 Agent 节点:
def vec_kg(state: AgentState):
messages = state["messages"][-1]
question = messages.content
prompt = PromptTemplate(
template="""You are an assistant for question-answering tasks.
Use the following pieces of retrieved context to answer the question. If you don't know the answer, just say that you don't know.
Use three sentences maximum and keep the answer concise:
Question: {question}
Context: {context}
Answer:
""",
input_variables=["question", "document"],
)
# 构建传统的RAG Chain
rag_chain = prompt | graph_llm | StrOutputParser()
# 构建检索器
retriever = vectorstore.as_retriever(search_kwargs={"k": 1})
# 执行检索
docs = retriever.invoke("question")
generation = rag_chain.invoke({"context": docs, "question": question})
final_response = [HumanMessage(content=generation, name="vec_kg")] # 这里要添加名称
return {"messages": final_response}
六、构建混合知识库检索多代理系统
# 依赖导入 & ORM 基类初始化
from sqlalchemy import create_engine, Column, Integer, String, Float, ForeignKey
from sqlalchemy.orm import sessionmaker, declarative_base
from faker import Faker
import random
# 创建基类
Base = declarative_base()
# 定义模型
class SalesData(Base):
__tablename__ = 'sales_data'
sales_id = Column(Integer, primary_key=True)
product_id = Column(Integer, ForeignKey('product_information.product_id'))
employee_id = Column(Integer) # 示例简化，未创建员工表
customer_id = Column(Integer, ForeignKey('customer_information.customer_id'))
sale_date = Column(String(50))
quantity = Column(Integer)
amount = Column(Float)
discount = Column(Float)
class CustomerInformation(Base):
__tablename__ = 'customer_information'
customer_id = Column(Integer, primary_key=True)
customer_name = Column(String(50))
contact_info = Column(String(50))
region = Column(String(50))
customer_type = Column(String(50))
class ProductInformation(Base):
__tablename__ = 'product_information'
product_id = Column(Integer, primary_key=True)
product_name = Column(String(50))
category = Column(String(50))
unit_price = Column(Float)
stock_level = Column(Integer)
class CompetitorAnalysis(Base):
__tablename__ = 'competitor_analysis'
competitor_id = Column(Integer, primary_key=True)
competitor_name = Column(String(50))
region = Column(String(50))
market_share = Column(Float)
# 2、读取配置信息
from dotenv import load_dotenv
load_dotenv()
import os
passwrod = os.environ["MYSQL_PASSWORD"]
acc = os.environ["MYSQL_account"]
DATABASE_URI = f'mysql+pymysql://{acc}:{passwrod}@localhost/langgraph?charset=utf8mb4'
engine = create_engine(DATABASE_URI)
Session = sessionmaker(bind=engine)
session = Session()
from pydantic import BaseModel, Field
from langchain_core.tools import tool
from typing import Union, Optional
class AddSaleSchema(BaseModel):
product_id: int
employee_id: int
customer_id: int
sale_date: str
quantity: int
amount: float
discount: float
class DeleteSaleSchema(BaseModel):
sales_id: int
class UpdateSaleSchema(BaseModel):
sales_id: int
quantity: int
amount: float
class QuerySalesSchema(BaseModel):
sales_id: int
# 1. 添加销售数据：
@tool(args_schema=AddSaleSchema)
def add_sale(product_id, employee_id, customer_id, sale_date, quantity, amount, discount):
"""Add sale record to the database."""
session = Session()
try:
new_sale = SalesData(
product_id=product_id,
employee_id=employee_id,
customer_id=customer_id,
sale_date=sale_date,
quantity=quantity,
amount=amount,
discount=discount
)
session.add(new_sale)
session.commit()
return {"messages": ["销售记录添加成功。"]}
except Exception as e:
return {"messages": [f"添加失败，错误原因：{e}"]}
finally:
session.close()
# 2. 删除销售数据
@tool(args_schema=DeleteSaleSchema)
def delete_sale(sales_id):
"""Delete sale record from the database."""
session = Session()
try:
sale_to_delete = session.query(SalesData).filter(SalesData.sales_id == sales_id).first()
if sale_to_delete:
session.delete(sale_to_delete)
session.commit()
return {"messages": ["销售记录删除成功。"]}
else:
return {"messages": [f"未找到销售记录ID: {sales_id}"]}
except Exception as e:
return {"messages": [f"删除失败，错误原因: {e}"]}
finally:
session.close()
# 3. 修改销售数据
@tool(args_schema=UpdateSaleSchema)
def update_sale(sales_id, quantity, amount):
"""Update sale record in the database."""
session = Session()
try:
sale_to_update = session.query(SalesData).filter(SalesData.sales_id == sales_id).first()
if sale_to_update:
sale_to_update.quantity = quantity
sale_to_update.amount = amount
session.commit()
return {"messages": ["销售记录更新成功。"]}
else:
return {"messages": [f"未找到销售记录ID: {sales_id}"]}
except Exception as e:
return {"messages": [f"更新失败，错误原因: {e}"]}
finally:
session.close()
# 4. 查询销售数据
@tool(args_schema=QuerySalesSchema)
def query_sales(sales_id):
"""Query sale record from the database."""
session = Session()
try:
sale_data = session.query(SalesData).filter(SalesData.sales_id == sales_id).first()
if sale_data:
return {
"sales_id": sale_data.sales_id,
"product_id": sale_data.product_id,
"employee_id": sale_data.employee_id,
"customer_id": sale_data.customer_id,
"sale_date": sale_data.sale_date,
"quantity": sale_data.quantity,
"amount": sale_data.amount,
"discount": sale_data.discount
}
else:
return {"messages": [f"未找到销售记录ID: {sales_id}。"]}
except Exception as e:
return {"messages": [f"查询失败，错误原因: {e}"]}
finally:
session.close()
from typing import Annotated
from langchain_core.tools import tool
from langchain_experimental.utilities import PythonREPL
import json
repl = PythonREPL()
@tool
def python_repl(
code: Annotated[str, "The python code to execute to generate your chart."],
):
"""Use this to execute python code. If you want to see the output of a value,
you should print it out with <code>print(...)</code>. This is visible to the user."""
try:
result = repl.run(code)
except BaseException as e:
return f"Failed to execute. Error: {repr(e)}"
result_str = f"Successfully executed:\n``<code>python\n{code}\n</code>``\nStdout: {result}"
return result_str
# 使用 create_react_agent 构建成两个 ReAct 代理。
from langchain_core.messages import HumanMessage
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import create_react_agent
# 1、配置模型
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
model_name = os.environ["OPENAI_MODEL_NAME"]
coder_key = os.environ["MINIMAX_API_KEY"]
coder_base_url = os.environ["MINIMAX_BASE_URL"]
coder_model_name = os.environ["MINIMAX_MODEL_NAME"]
db_llm = ChatOpenAI(model=model_name, api_key=key, base_url=base_url, temperature=0, request_timeout=100)
coder_llm = ChatOpenAI(model=coder_model_name, api_key=coder_key, base_url=coder_base_url, temperature=0, request_timeout=100)
db_agent = create_react_agent(
db_llm,
tools=[add_sale, delete_sale, update_sale, query_sales],
prompt="You use to perform database operations while should provide accurate data for the code_gen" # ✅
)
code_agent = create_react_agent(
coder_llm,
tools=[python_repl],
prompt="Run python code to display diagrams or output execution results" # ✅
)
# 打开文件，并赋予读取模式 'r'
with open('D:\projects\code\class-study\GraphRAG-Conception\comany.txt', 'r', encoding="utf-8") as file:
# 读取文件的全部内容
content = file.read()
print(content)
# 转化成 Document 对象。
from langchain_core.documents import Document
documents = [Document(page_content=content)]
print(documents)
# 准备好数据后，我们可以使用 langchain_experimental.graph_transformers 中的 LLMGraphTransformer 将其摄取到 Neo4j 中。该工具会自动将文档转换为图格式。LLMGraphTransformer 能够以两种完全独立的模式运行：
# Tool-Based 模式（默认）：当使用的大模型支持结构化输出或函数调用时，该模式利用内置的 with_structured_output 来使用工具。工具规范定义了输出格式，确保以结构化、预定义的方式提取实体和关系。
# 基于提示的模式（回退）：在使用的大模型不支持工具或函数调用的情况下，该转换器回退到纯粹提示驱动的方法。该模式使用 few-shot 提示来定义输出格式，指导大模型以基于文本的方式提取实体和关系。然后通过自定义函数解析结果，该函数将大模型的输出转换为 JSON 格式。该 JSON 用于填充节点和关系。
# pip install neo4j
# GraphRAG Setup
from langchain_community.graphs import Neo4jGraph
from langchain_experimental.graph_transformers import LLMGraphTransformer
from langchain_core.documents import Document
from langchain_openai import ChatOpenAI
# 1、配置模型
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
model_name = os.environ["OPENAI_MODEL_NAME"]
graph_llm = ChatOpenAI(model=model_name, api_key=key, base_url=base_url, temperature=0, request_timeout=110)
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
model_name = os.environ["OPENAI_MODEL_NAME"]
NEO4J_URL=os.environ["NEO4J_URL"]
NEO4J_USERNAME=os.environ["NEO4J_USERNAME"]
NEO4J_PASSWORD=os.environ["NEO4J_PASSWORD"]
NEO4J_DATABASE=os.environ["NEO4J_DATABASE"]
# 创建图数据库示例
graph = Neo4jGraph(
url=NEO4J_URL, # 替换为自己的
username=NEO4J_USERNAME, # 替换为自己的
password=NEO4J_PASSWORD, #替换为自己的
database=NEO4J_DATABASE # 替换为自己的
)
from langchain_community.chains.graph_qa.cypher import GraphCypherQAChain
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(temperature=0, model_name="gpt-4o-mini", api_key=key,base_url=base_url)
cypher_chain = GraphCypherQAChain.from_llm(
graph=graph,
cypher_llm=llm,
qa_llm=llm,
validate_cypher=True, # Validate relationship directions
verbose=True,
allow_dangerous_requests=True
)
# 将 GraphRAG 封装为 LangGraph 多智能体节点
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_core.messages import HumanMessage
# 构建混合知识库检索多代理系统
from langgraph.graph import StateGraph, MessagesState, START, END
# 同时可以登录网页端进行确认：
from langchain_core.prompts import PromptTemplate
# from langchain import hub
from langchain_core.output_parsers import StrOutputParser
class AgentState(MessagesState):
next: str
# 打开文件，并赋予读取模式 'r'
with open('D:\projects\code\class-study\GraphRAG-Conception\comany.txt', 'r', encoding="utf-8") as file:
# 读取文件的全部内容
content = file.read()
print(content)
# 转化成 Document 对象。
from langchain_core.documents import Document
documents = [Document(page_content=content)]
print(documents)
# 1、配置模型
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
model_name = os.environ["OPENAI_MODEL_NAME"]
graph_llm = ChatOpenAI(model=model_name, api_key=key, base_url=base_url, temperature=0, request_timeout=100)
embedding_key = os.environ["EMBEDDING_KEY"]
embedding_url = os.environ["EMBEDDING_URL"]
embedding_name = os.environ["EMBEDDING_NAME"]
milvus_password = os.environ["MILVUS_PASSWORD"]
milvus_url = os.environ["MILVUS_URL"]
milvus_user = os.environ["MILVUS_USER"]
from langchain_text_splitters import RecursiveCharacterTextSplitter
chunk_size = 800
chunk_overlap = 100
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size, chunk_overlap=chunk_overlap
)
# 执行文档切分
splits = text_splitter.split_documents(documents)
splits
from langchain_openai import OpenAIEmbeddings
embeddings = OpenAIEmbeddings(
model=embedding_name,
base_url=embedding_url,
api_key=embedding_key,
# dimensions=4096, # 如需指定维度，取消注释
)
from pymilvus import connections, utility
# 先连接
connections.connect(
alias="default",
uri= milvus_url,
user= milvus_user,
password= milvus_password,
)
# 如果切块的方式、大小有调整，就必须先删除旧的集合，否则切块内容不会变
# 删除旧集合
if utility.has_collection("company_rag_milvus"):
utility.drop_collection("company_rag_milvus")
print("已删除旧集合")
# 然后再创建
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import WebBaseLoader
from langchain_milvus import Milvus
vectorstore = Milvus.from_documents(
documents=splits,
collection_name="company_rag_milvus",
embedding=embeddings,
connection_args={
"uri": milvus_url,
"user": milvus_user,
"password": milvus_password,
}
)
def vec_kg(state: AgentState):
messages = state["messages"][-1]
question = messages.content
prompt = PromptTemplate(
template="""You are an assistant for question-answering tasks.
Use the following pieces of retrieved context to answer the question. If you don't know the answer, just say that you don't know.
Use three sentences maximum and keep the answer concise:
Question: {question}
Context: {context}
Answer:
""",
input_variables=["question", "document"],
)
# 构建传统的RAG Chain
rag_chain = prompt | graph_llm | StrOutputParser()
# 构建检索器
retriever = vectorstore.as_retriever(search_kwargs={"k": 1})
# 执行检索
docs = retriever.invoke("question")
generation = rag_chain.invoke({"context": docs, "question": question})
final_response = [HumanMessage(content=generation, name="vec_kg")] # 这里要添加名称
return {"messages": final_response}
def graph_kg(state: AgentState):
messages = state["messages"][-1]
response = cypher_chain.invoke(messages.content)
final_response = [HumanMessage(content=response["result"], name="graph_kg")] # 这里要添加名称
return {"messages": final_response}
def db_node(state: AgentState):
result = db_agent.invoke(state)
return {
"messages": [
HumanMessage(content=result["messages"][-1].content, name="sqler")
]
}
def code_node(state: AgentState):
result = code_agent.invoke(state)
return {
"messages": [HumanMessage(content=result["messages"][-1].content, name="coder")]
}
def chat(state: AgentState):
messages = state["messages"][-1]
model_response = llm.invoke(messages.content)
final_response = [HumanMessage(content=model_response.content, name="chatbot")]
return {"messages": final_response}
# 新增两个不同的数据库代理节点:
members = ["graph_kg", "vec_kg"]
options = members + ["FINISH"]
from typing import Literal
from typing_extensions import TypedDict
class Router(TypedDict):
"""Worker to route to next. If no workers needed, route to FINISH"""
next: Literal["graph_kg", "vec_kg", "FINISH"]
# Literal 是 Python 的 typing 模块中的一个类型，用于定义一个变量的具体值的类型限制。
# 当使用 Literal 时，实际上是在告诉 Python，变量的值必须是指定的几个值中的一个。
# 而 next: Literal["chat", "coder", "sqler"] 意味着 next 属性只能赋予三个字符串值之一: "chat"、"coder"、"sqler"或"FINISH"，
# 分别用来表示使用哪一个子代理来执行任务，或者直接通过 END 结束当前的图。
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, ToolMessage
def supervisor(state: AgentState):
system_prompt = (
"You are a supervisor tasked with managing a conversation between the"
f" following workers: {members}.\n\n"
"Each worker has a specific role:\n"
"- chat: Responds directly to user inputs using natural language.\n"
"- graph_kg: Stores market and company information, built on a graph-based knowledge base, excels at graph query.\n"
"- vec_kg: Stores market and company information, constructed on a traditional semantic retrieval knowledge base.\n"
"Given the following user request, respond with the worker to act next."
" Each worker will perform a task and respond with their results and status."
" When finished, respond with FINISH."
)
messages = [{"role": "system", "content": system_prompt},] + state["messages"]
response = llm.with_structured_output(Router).invoke(messages)
next_ = response["next"]
if next_ == "FINISH":
next_ = END
return {"next": next_}
builder = StateGraph(AgentState)
# builder.add_edge(START, "supervisor")
builder.add_node("supervisor", supervisor)
builder.add_node("chat", chat)
builder.add_node("coder", db_node)
builder.add_node("sqler", code_node)
builder.add_node("graph_kg", graph_kg)
builder.add_node("vec_kg", vec_kg)
# 然后让每个子代理在完成工作后总是向主管 “汇报”，即需要构建它们之间的边。如下所示:
for member in members:
# 我们希望我们的工人在完成工作后总是向主管“汇报”
builder.add_edge(member, "supervisor")
# 然后在图状态中填充 next 字段，路由到具体的某个节点或者结束图的运行，用来指定如何执行接下来的任务。
builder.add_conditional_edges("supervisor", lambda state: state["next"])
# 添加开始和节点
builder.add_edge(START, "supervisor")
# 编译图
graph = builder.compile()
# 3、渲染流程图可视化
image_data = graph.get_graph(xray=True).draw_mermaid_png()
with open("Hybrid-Knowledge-Agent-g.png", "wb") as f:
f.write(image_data)
# 编译完成后，就可以进行问答了，这里我们测试几轮不同的问题类型：
for chunk in graph.stream({"messages": "史进的外号是什么？"}, stream_mode="values"):
chunk["messages"][-1].pretty_print()