aidlab/FiDA_Python
7
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat 1.增加 建议词 机制 2.对话生图实现

Browse Source
This commit is contained in:
zcr
2026-02-06 11:55:11 +08:00
parent ec195d17e1
commit 3248c45cd4
12 changed files with 655 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

103
src/server/agent/agents.py
View File

@@ -1,86 +1,117 @@
import os
from google.oauth2 import service_account
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.messages import HumanMessage, SystemMessage, ToolMessage, AIMessage
from langchain_core.runnables import RunnableConfig
from langchain_google_genai import ChatGoogleGenerativeAI
from src.server.agent.state import AgentState
from src.server.agent.tools import generate_2025_report_tool, generate_furniture_sketch
from src.server.agent.config_loader import get_agent_prompt
from src.core.config import settings
from src.server.utils.generate_suggestion import generate_chat_suggestions
creds = service_account.Credentials.from_service_account_file(
settings.GOOGLE_GENAI_USE_VERTEXAI,
scopes=["https://www.googleapis.com/auth/cloud-platform"],
)
# 初始化 Gemini 模型 (使用 Flash 以保证速度)
llm = ChatGoogleGenerativeAI(
model="gemini-2.0-flash", temperature=0.5, credentials=creds,
project="aida-461108", location='us-central1', vertexai=True, api_key=settings.GOOGLE_API_KEY
)
# 辅助函数：根据配置动态获取 LLM
def get_model(config: RunnableConfig):
# 从 configurable 中获取温度，默认为 0.5 (对应你之前的设置)
# 这个 key 必须与你在 chat_stream 路由里定义的 "llm_temperature" 一致
temp = config["configurable"].get("llm_temperature", 0.5)
return ChatGoogleGenerativeAI(
model="gemini-2.0-flash",
temperature=temp,
credentials=creds,
project=settings.GOOGLE_CLOUD_PROJECT,
location=settings.GOOGLE_CLOUD_LOCATION,
vertexai=True,
api_key=settings.GOOGLE_API_KEY
)
# --- 1. Designer Agent (设计顾问) ---
def designer_node(state: AgentState):
async def designer_node(state: AgentState, config: RunnableConfig):
"""负责细化设计需求，提供专业参数"""
model = get_model(config) # 获取带动态温度的模型
messages = state["messages"]
system_text = get_agent_prompt("designer") or """
你是一位资深的家具设计师。你的职责是：
1. 从用户的模糊描述中提取或补充具体的设计参数（尺寸、材质、人体工学数据）。
2. 如果用户想画图，不要直接画，而是先描述清楚细节，然后让 Visualizer 去画。
请以专业的口吻回复。
"""
system_text = get_agent_prompt("designer")
system_prompt = SystemMessage(content=system_text)
response = llm.invoke([system_prompt] + messages)
# 改为异步调用 ainvoke
response = await model.ainvoke([system_prompt] + messages)
return {"messages": [response]}
# --- 2. Researcher Agent (情报专家) ---
def researcher_node(state: AgentState):
async def researcher_node(state: AgentState, config: RunnableConfig):
"""负责调用报告生成工具"""
# 绑定工具给 LLM
model = get_model(config)
tools = [generate_2025_report_tool]
llm_with_tools = llm.bind_tools(tools)
llm_with_tools = model.bind_tools(tools)
messages = state["messages"]
system_text = get_agent_prompt("researcher") or "你是情报专家，负责检索与整理参考资料并生成报告。"
system_text = get_agent_prompt("researcher")
system_prompt = SystemMessage(content=system_text)
response = llm_with_tools.invoke([system_prompt] + messages)
response = await llm_with_tools.ainvoke([system_prompt] + messages)
# 如果模型决定调用工具
if response.tool_calls:
# 这里为了简化，直接在节点内执行工具（LangGraph也可以用 ToolNode）
tool_call = response.tool_calls[0]
if tool_call["name"] == "generate_2025_report_tool":
result = generate_2025_report_tool.invoke(tool_call["args"])
# 这里的工具调用如果也是异步的，建议加 await
result = await generate_2025_report_tool.ainvoke(tool_call["args"])
return {"messages": [response, HumanMessage(content=str(result))]}
return {"messages": [response]}
# --- 3. Visualizer Agent (视觉专家) ---
def visualizer_node(state: AgentState):
async def visualizer_node(state: AgentState, config: RunnableConfig):
"""负责将自然语言转化为绘图 Prompt 并调用绘图工具"""
model = get_model(config)
tools = [generate_furniture_sketch]
llm_with_tools = llm.bind_tools(tools)
llm_with_tools = model.bind_tools(tools)
messages = state["messages"]
system_text = get_agent_prompt("visualizer") or """
你是视觉专家。你的目标是生成高质量的家具草图。
步骤：
1. 根据上下文，编写一个详细的 Stable Diffusion 风格的英文 Prompt。
2. 必须调用 generate_furniture_sketch 工具来生成图片。
"""
system_text = get_agent_prompt("visualizer")
# 强制它尝试调用工具
system_prompt = SystemMessage(content=system_text)
response = llm_with_tools.invoke([system_prompt] + messages)
response = await llm_with_tools.ainvoke([system_prompt] + messages)
if response.tool_calls:
tool_call = response.tool_calls[0]
if tool_call["name"] == "generate_furniture_sketch":
result = generate_furniture_sketch.invoke(tool_call["args"])
# 返回工具结果给 LLM，让它生成最终回复
final_msg = f"已为您生成草图，链接如下：{result}"
return {"messages": [response, HumanMessage(content=final_msg)]}
img_url = await generate_furniture_sketch.ainvoke(tool_call["args"])
return {
"messages": [
response,
ToolMessage(content=img_url, tool_call_id=tool_call["id"]) # 标记这是一个图片结果
]
}
return {"messages": [response]}
# --- 4. Suggester Agent (推荐对话专家) ---
async def suggester_node(state: AgentState, config: RunnableConfig):
"""专门生成追问建议的节点，作为流程终点"""
model = get_model(config)
messages = state["messages"]
# 只需要分析最近的对话
suggestions = await generate_chat_suggestions(messages, model)
# 返回一个特殊消息，前端通过解析 additional_kwargs 获取按钮内容
return {
"messages": [
AIMessage(
content="",
additional_kwargs={"suggestions": suggestions},
name="Suggester"
)
]
}

39
src/server/agent/graph.py
View File

@@ -10,14 +10,15 @@ from pymongo import MongoClient
from src.core.config import settings, MONGO_URI
from src.server.agent.state import AgentState
from src.server.agent.agents import designer_node, researcher_node, visualizer_node
from src.server.agent.agents import designer_node, researcher_node, visualizer_node, suggester_node
from langgraph.checkpoint.mongodb import MongoDBSaver
# --- Supervisor (路由逻辑) ---
# 定义路由的输出结构，强制 LLM 选择一个
class RouteResponse(BaseModel):
next: Literal["Designer", "Researcher", "Visualizer", "FINISH"]
# 将 FINISH 替换或增加 Suggester
next: Literal["Designer", "Researcher", "Visualizer", "Suggester"]
creds = service_account.Credentials.from_service_account_file(
@@ -34,30 +35,23 @@ llm_supervisor = ChatGoogleGenerativeAI(
def supervisor_node(state: AgentState):
messages = state["messages"]
if not messages:
return {"next": "FINISH"}
return {"next": "Suggester"}
last_message = messages[-1]
# --- 改进的拦截逻辑 ---
# 如果最后一条消息是 AI 产生的（且没有调用工具），说明专家已经回复完了用户
# 此时我们才拦截并结束，否则会导致专家没机会说话
# --- 拦截逻辑修改 ---
# 如果专家已经回复完了（AIMessage 且无工具调用），则交给 Suggester 生成按钮
if isinstance(last_message, AIMessage) and not last_message.tool_calls:
return {"next": "FINISH"}
return {"next": "Suggester"}
# 如果最后一条是 HumanMessage，说明用户刚说完，Supervisor 必须派发任务
system_prompt = """
你是家具设计团队的主管(Supervisor)。
请根据用户的意图，选择最合适的专家：
- Designer: 设计建议、参数细化、闲聊、问候。
- Visualizer: 绘图、看草图。
- Researcher: 市场报告、趋势。
只需输出专家名称。
system_prompt = """你是家具设计主管。分配任务给专家：
- Designer: 设计建议、参数细化。
- Visualizer: 绘图需求。
- Researcher: 市场报告。
"""
chain = llm_supervisor.with_structured_output(RouteResponse)
decision = chain.invoke([{"role": "system", "content": system_prompt}] + messages)
return {"next": decision.next}
@@ -68,10 +62,11 @@ workflow.add_node("Supervisor", supervisor_node)
workflow.add_node("Designer", designer_node)
workflow.add_node("Researcher", researcher_node)
workflow.add_node("Visualizer", visualizer_node)
workflow.add_node("Suggester", suggester_node) # 新增节点
workflow.add_edge(START, "Supervisor")
# 这里的逻辑是关键：Supervisor 决定去向
# 修改条件边映射
workflow.add_conditional_edges(
"Supervisor",
lambda state: state["next"],
@@ -79,16 +74,18 @@ workflow.add_conditional_edges(
"Designer": "Designer",
"Researcher": "Researcher",
"Visualizer": "Visualizer",
"FINISH": END
"Suggester": "Suggester" # 原本的 FINISH 现在指向 Suggester
}
)
# 重点修改：专家执行完后，必须回到 Supervisor 进行状态检查
# 如果 Supervisor 发现专家刚说完话，它会触发上面的逻辑返回 FINISH
# 专家执行完依然回到 Supervisor
workflow.add_edge("Designer", "Supervisor")
workflow.add_edge("Researcher", "Supervisor")
workflow.add_edge("Visualizer", "Supervisor")
# 重点：Suggester 是整个流程的终点
workflow.add_edge("Suggester", END)
client = MongoClient(MONGO_URI)
checkpointer = MongoDBSaver(
client=client["furniture_agent_db"],

101
src/server/agent/tools.py
View File

@@ -1,4 +1,30 @@
import base64
import uuid
from google.oauth2 import service_account
from langchain_core.tools import tool
from google import genai
from google.genai.types import GenerateContentConfig, Modality
from PIL import Image
from io import BytesIO
from minio import Minio
from src.core.config import settings
from src.server.utils.new_oss_client import oss_upload_image
# 初始化全局凭证和客户端
creds = service_account.Credentials.from_service_account_file(
settings.GOOGLE_GENAI_USE_VERTEXAI,
scopes=["https://www.googleapis.com/auth/cloud-platform"],
)
minio_client = Minio(settings.MINIO_URL, access_key=settings.MINIO_ACCESS, secret_key=settings.MINIO_SECRET, secure=settings.MINIO_SECURE)
client = genai.Client(
credentials=creds,
project=settings.GOOGLE_CLOUD_PROJECT,
location=settings.GOOGLE_CLOUD_LOCATION,
vertexai=True
)
# --- 模拟你已经开发好的报告生成功能 ---
@@ -13,13 +39,76 @@ def generate_2025_report_tool(topic: str) -> str:
return f"【报告生成成功】已生成关于 {topic} 的 PDF 报告。核心洞察：2025年趋势倾向于生物嗜好设计(Biophilic Design)和可持续软木材质。"
# --- 绘图工具 ---
# --- 2. 绘图工具 (接入 Nano Banana 逻辑) ---
@tool
def generate_furniture_sketch(prompt: str) -> str:
"""
用于生成家具草图。输入必须是详细的英文绘画提示词(Prompt)。
使用 Gemini 图像生成模型根据详细的英文提示词生成家具设计草图。
"""
print(f"\n[系统日志] 正在调用 Gemini/Imagen 绘图 API，Prompt: {prompt}...")
# 在真实场景中，这里调用 Google Imagen API 或 Midjourney API
# 示例返回一个模拟的图片链接
return "https://furniture-design-db.com/generated_sketch_v1.jpg"
print(f"\n[系统日志] 正在调用 Nano Banana (Gemini Image Gen) ...")
try:
response = client.models.generate_content(
model="gemini-2.5-flash-image",
contents=(f"Generate a professional furniture design sketch: {prompt}"),
config=GenerateContentConfig(
response_modalities=[Modality.TEXT, Modality.IMAGE],
),
)
image_bytes = None
for part in response.candidates[0].content.parts:
if part.inline_data:
image_bytes = part.inline_data.data
break
if not image_bytes:
return "未能生成图像数据。"
object_name = f"furniture/sketches/{uuid.uuid4()}.png"
bucket = "fida-test" # 替换为你的 bucket 名称
# 3. 调用你的上传函数
upload_res = oss_upload_image(
oss_client=minio_client,
bucket=bucket,
object_name=object_name,
image_bytes=image_bytes
)
if upload_res:
# 4. 构造访问链接 (如果是私有 bucket，需使用 presigned_get_object)
# 这里简单示例为直接访问地址
image_url = f"{bucket}/{object_name}"
return image_url
else:
return "图片生成成功，但上传至存储服务器失败。"
except Exception as e:
return f"绘图流程异常: {str(e)}"
if __name__ == '__main__':
print(generate_furniture_sketch("椅子"))
# creds = service_account.Credentials.from_service_account_file(
# settings.GOOGLE_GENAI_USE_VERTEXAI,
# scopes=["https://www.googleapis.com/auth/cloud-platform"],
# )
# client = genai.Client(
# credentials=creds,
# project=settings.GOOGLE_CLOUD_PROJECT,
# location=settings.GOOGLE_CLOUD_LOCATION,
# vertexai=True
# )
#
# response = client.models.generate_content(
# model="gemini-2.5-flash-image",
# contents=("Generate an image of the Eiffel tower with fireworks in the background."),
# config=GenerateContentConfig(
# response_modalities=[Modality.TEXT, Modality.IMAGE],
# ),
# )
#
# for part in response.candidates[0].content.parts:
# if part.text:
# print(part.text)
# elif part.inline_data:
# image = Image.open(BytesIO((part.inline_data.data)))
# image.save("example-image-eiffel-tower.png")