mirror of
https://github.com/fastapiadmin/FastapiAdmin.git
synced 2026-09-20 20:39:55 +00:00
refactor(ai): 移除AI工具模块并内联AI处理逻辑
将AI工具模块中的AIClient类逻辑内联到service中，删除不再使用的工具模块 简化资源查询参数中的path字段处理
This commit is contained in:
zhangtao
@@ -1,8 +1,14 @@
|
||||
from collections.abc import AsyncGenerator
|
||||
from typing import Any
|
||||
from collections.abc import AsyncGenerator
|
||||
from typing import Any
|
||||
|
||||
from langchain_core.messages import HumanMessage, SystemMessage
|
||||
from langchain_openai import ChatOpenAI
|
||||
|
||||
from app.api.v1.module_system.auth.schema import AuthSchema
|
||||
from app.core.exceptions import CustomException
|
||||
from app.config.setting import settings
|
||||
from app.core.logger import log
|
||||
|
||||
from .crud import McpCRUD
|
||||
@@ -13,8 +19,6 @@ from .schema import (
|
||||
McpQueryParam,
|
||||
McpUpdateSchema,
|
||||
)
|
||||
from .tools.ai_util import AIClient
|
||||
|
||||
|
||||
class McpService:
|
||||
"""MCP服务层"""
|
||||
@@ -132,13 +136,78 @@ class McpService:
|
||||
- AsyncGenerator[str, None]: 异步生成器,每次返回一个聊天响应
|
||||
"""
|
||||
# 创建MCP客户端实例
|
||||
mcp_client = AIClient()
|
||||
lll_model = ChatOpenAI(
|
||||
api_key=lambda: settings.OPENAI_API_KEY,
|
||||
model=settings.OPENAI_MODEL,
|
||||
base_url=settings.OPENAI_BASE_URL,
|
||||
temperature=0.7,
|
||||
streaming=True,
|
||||
)
|
||||
|
||||
system_prompt = (
|
||||
"""你是一个有用的AI助手，可以帮助用户回答问题和提供帮助。请用中文回答用户的问题。"""
|
||||
)
|
||||
|
||||
messages = [
|
||||
SystemMessage(content=system_prompt),
|
||||
HumanMessage(content=query.message),
|
||||
]
|
||||
|
||||
try:
|
||||
# 处理消息
|
||||
async for response in mcp_client.process(query.message):
|
||||
yield response
|
||||
# 使用LangChain的流式响应
|
||||
async for chunk in lll_model.astream(messages):
|
||||
yield chunk.text
|
||||
|
||||
except Exception as e:
|
||||
log.debug(f"关闭AIClient时发生异常(预期行为，服务可能正在关闭): {e}")
|
||||
raise CustomException(
|
||||
msg=f"关闭AIClient时发生异常(预期行为，服务可能正在关闭), 异常信息: {e}"
|
||||
)
|
||||
|
||||
status_code = getattr(e, "status_code", None)
|
||||
body = getattr(e, "body", None)
|
||||
message = None
|
||||
error_type = None
|
||||
error_code = None
|
||||
try:
|
||||
if isinstance(body, dict) and "error" in body:
|
||||
err = body.get("error") or {}
|
||||
error_type = err.get("type")
|
||||
error_code = err.get("code")
|
||||
message = err.get("message")
|
||||
except Exception:
|
||||
raise CustomException(f"解析 OpenAI 错误失败: {e!s}")
|
||||
|
||||
text = str(e)
|
||||
msg = message or text
|
||||
|
||||
# 特定错误映射
|
||||
# 欠费/账户状态异常
|
||||
if (
|
||||
(error_code == "Arrearage")
|
||||
or (error_type == "Arrearage")
|
||||
or ("in good standing" in (msg or ""))
|
||||
):
|
||||
raise ValueError("账户欠费或结算异常，访问被拒绝。请检查账号状态或更换有效的 API Key。")
|
||||
# 鉴权失败
|
||||
if status_code == 401 or "invalid api key" in msg.lower():
|
||||
raise ValueError("鉴权失败，API Key 无效或已过期。请检查系统配置中的 API Key。")
|
||||
# 权限不足或被拒绝
|
||||
if status_code == 403 or error_type in {
|
||||
"PermissionDenied",
|
||||
"permission_denied",
|
||||
}:
|
||||
raise ValueError("访问被拒绝，权限不足或账号受限。请检查账户权限设置。")
|
||||
# 配额不足或限流
|
||||
if status_code == 429 or error_type in {
|
||||
"insufficient_quota",
|
||||
"rate_limit_exceeded",
|
||||
}:
|
||||
raise ValueError("请求过于频繁或配额已用尽。请稍后重试或提升账户配额。")
|
||||
# 客户端错误
|
||||
if status_code == 400:
|
||||
raise ValueError(f"请求参数错误或服务拒绝：{message or '请检查输入内容。'}")
|
||||
# 服务端错误
|
||||
if status_code in {500, 502, 503, 504}:
|
||||
raise ValueError("服务暂时不可用，请稍后重试。")
|
||||
|
||||
# 默认兜底
|
||||
raise CustomException(f"处理您的请求时出现错误：{msg}")
|
||||
|
||||
|
||||
@@ -1 +0,0 @@
|
||||
|
||||
@@ -1,186 +0,0 @@
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
from langchain.agents import AgentState, create_agent
|
||||
from langchain.agents.middleware import (
|
||||
ModelRequest,
|
||||
ModelResponse,
|
||||
after_model,
|
||||
before_model,
|
||||
dynamic_prompt,
|
||||
wrap_model_call,
|
||||
)
|
||||
from langchain.agents.structured_output import (
|
||||
MultipleStructuredOutputsError,
|
||||
StructuredOutputValidationError,
|
||||
ToolStrategy,
|
||||
)
|
||||
from langchain.chat_models import init_chat_model
|
||||
from langchain.messages import (
|
||||
AIMessage,
|
||||
HumanMessage,
|
||||
RemoveMessage,
|
||||
SystemMessage,
|
||||
)
|
||||
from langchain.tools import ToolRuntime, tool
|
||||
from langgraph.checkpoint.memory import InMemorySaver
|
||||
from langgraph.graph.message import REMOVE_ALL_MESSAGES
|
||||
from langgraph.runtime import Runtime
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
# =================定义提示词=================
|
||||
SYSTEM_PROMPT = """You are an expert weather forecaster, who speaks in puns.
|
||||
|
||||
You have access to two tools:
|
||||
|
||||
- get_weather_for_location: use this to get the weather for a specific location
|
||||
- get_user_location: use this to get the user's location
|
||||
|
||||
If a user asks you for the weather, make sure you know the location. If you can tell from the question that they mean wherever they are, use the get_user_location tool to find their location.
|
||||
"""
|
||||
|
||||
|
||||
# =================定义工具=================
|
||||
@tool
|
||||
def get_weather_for_location(city: str) -> str:
|
||||
"""Get weather for a given city."""
|
||||
return f"It's always sunny in {city}!"
|
||||
|
||||
|
||||
@dataclass
|
||||
class Context:
|
||||
"""Custom runtime context schema."""
|
||||
|
||||
user_id: str
|
||||
|
||||
|
||||
@tool
|
||||
def get_user_location(runtime: ToolRuntime[Context]) -> str:
|
||||
"""Retrieve user information based on user ID."""
|
||||
user_id = runtime.context.user_id
|
||||
return "Florida" if user_id == "1" else "SF"
|
||||
|
||||
|
||||
# =================定义模型=================
|
||||
model = init_chat_model("claude-sonnet-4-5-20250929", temperature=0.5, timeout=10, max_tokens=1000)
|
||||
|
||||
|
||||
# =================定义响应模型=================
|
||||
class ResponseFormat(BaseModel):
|
||||
"""Response schema for the agent."""
|
||||
|
||||
rating: int | None = Field(description="Rating from 1-5", ge=1, le=5)
|
||||
comment: str = Field(description="Review comment")
|
||||
punny_response: str
|
||||
weather_conditions: str | None = None
|
||||
|
||||
|
||||
# =================定义存储记忆=================
|
||||
checkpointer = InMemorySaver()
|
||||
|
||||
|
||||
# =================定义动态提示词=================
|
||||
@dynamic_prompt
|
||||
def dynamic_system_prompt(request: ModelRequest) -> str:
|
||||
user_name = getattr(request.runtime.context, "user_name", "User")
|
||||
system_prompt = f"You are a helpful assistant. Address the user as {user_name}."
|
||||
return system_prompt
|
||||
|
||||
|
||||
@before_model
|
||||
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
|
||||
"""Keep only the last few messages to fit context window."""
|
||||
messages = state["messages"]
|
||||
|
||||
if len(messages) <= 3:
|
||||
return None # No changes needed
|
||||
|
||||
first_msg = messages[0]
|
||||
recent_messages = messages[-3:] if len(messages) % 2 == 0 else messages[-4:]
|
||||
new_messages = [first_msg, *recent_messages]
|
||||
|
||||
return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES), *new_messages]}
|
||||
|
||||
|
||||
@after_model
|
||||
def validate_response(state: AgentState, runtime: Runtime) -> dict | None:
|
||||
"""Remove messages containing sensitive words."""
|
||||
last_message = state["messages"][-1]
|
||||
if any(word in last_message.content for word in ["password", "secret"]):
|
||||
return {"messages": [RemoveMessage(id=last_message.id or "")]}
|
||||
return None
|
||||
|
||||
|
||||
@wrap_model_call
|
||||
def inject_file_context(
|
||||
request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]
|
||||
) -> ModelResponse:
|
||||
"""Inject context about files user has uploaded this session."""
|
||||
# Read from State: get uploaded files metadata
|
||||
uploaded_files = request.state.get("uploaded_files", [])
|
||||
|
||||
if uploaded_files:
|
||||
# Build context about available files
|
||||
file_descriptions = [
|
||||
f"- {file['name']} ({file['type']}): {file['summary']}" for file in uploaded_files
|
||||
]
|
||||
|
||||
file_context = f"""Files you have access to in this conversation:
|
||||
{chr(10).join(file_descriptions)}
|
||||
|
||||
Reference these files when answering questions."""
|
||||
|
||||
# Inject file context before recent messages
|
||||
messages = [
|
||||
*request.messages,
|
||||
{"role": "user", "content": file_context},
|
||||
]
|
||||
request = request.override(messages=messages)
|
||||
|
||||
|
||||
def custom_error_handler(error: Exception) -> str:
|
||||
if isinstance(error, StructuredOutputValidationError):
|
||||
return "There was an issue with the format. Try again."
|
||||
if isinstance(error, MultipleStructuredOutputsError):
|
||||
return "Multiple structured outputs were returned. Pick the most relevant one."
|
||||
return f"Error: {error!s}"
|
||||
|
||||
|
||||
# =================定义智能体=================
|
||||
agent = create_agent(
|
||||
model=model,
|
||||
system_prompt=SYSTEM_PROMPT,
|
||||
tools=[get_user_location, get_weather_for_location],
|
||||
middleware=[
|
||||
dynamic_system_prompt,
|
||||
trim_messages,
|
||||
validate_response,
|
||||
inject_file_context,
|
||||
],
|
||||
context_schema=Context,
|
||||
response_format=ToolStrategy(
|
||||
schema=ResponseFormat,
|
||||
handle_errors=(ValueError, TypeError, custom_error_handler),
|
||||
),
|
||||
checkpointer=checkpointer,
|
||||
)
|
||||
|
||||
# =================定义线程=================
|
||||
config = {"configurable": {"thread_id": "1"}}
|
||||
|
||||
messages = [
|
||||
SystemMessage("You are a poetry expert"),
|
||||
HumanMessage("Write a haiku about spring"),
|
||||
AIMessage("Cherry blossoms bloom..."),
|
||||
]
|
||||
|
||||
# =================运行智能体=================
|
||||
response = agent.invoke(input=messages, config=config, context=Context(user_id="1"))
|
||||
|
||||
# =================解析响应=================
|
||||
print(response["structured_response"])
|
||||
# ResponseFormat(
|
||||
# punny_response="Florida is still having a 'sun-derful' day! The sunshine is playing 'ray-dio' hits all day long! I'd say it's the perfect weather for some 'solar-bration'! If you were hoping for rain, I'm afraid that idea is all 'washed up' - the forecast remains 'clear-ly' brilliant!",
|
||||
# weather_conditions="It's always sunny in Florida!"
|
||||
# )
|
||||
@@ -1,108 +0,0 @@
|
||||
from collections.abc import AsyncGenerator
|
||||
from typing import Any
|
||||
|
||||
from langchain_core.messages import HumanMessage, SystemMessage
|
||||
from langchain_openai import ChatOpenAI
|
||||
|
||||
from app.config.setting import settings
|
||||
from app.core.exceptions import CustomException
|
||||
from app.core.logger import log
|
||||
|
||||
|
||||
class AIClient:
|
||||
"""
|
||||
AI客户端类，用于与OpenAI API交互。
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
# 使用LangChain的ChatOpenAI类
|
||||
self.model = ChatOpenAI(
|
||||
api_key=lambda: settings.OPENAI_API_KEY,
|
||||
model=settings.OPENAI_MODEL,
|
||||
base_url=settings.OPENAI_BASE_URL,
|
||||
temperature=0.7,
|
||||
streaming=True,
|
||||
)
|
||||
|
||||
async def process(self, query: str) -> AsyncGenerator[str, Any]:
|
||||
"""
|
||||
处理查询并返回流式响应
|
||||
|
||||
参数:
|
||||
- query (str): 用户查询。
|
||||
|
||||
返回:
|
||||
- AsyncGenerator[str, Any]: 流式响应内容。
|
||||
"""
|
||||
system_prompt = (
|
||||
"""你是一个有用的AI助手，可以帮助用户回答问题和提供帮助。请用中文回答用户的问题。"""
|
||||
)
|
||||
|
||||
try:
|
||||
# 使用LangChain的异步流式生成
|
||||
messages = [
|
||||
SystemMessage(content=system_prompt),
|
||||
HumanMessage(content=query),
|
||||
]
|
||||
|
||||
# 使用LangChain的流式响应
|
||||
async for chunk in self.model.astream(messages):
|
||||
yield chunk.text
|
||||
|
||||
except Exception as e:
|
||||
# 记录详细错误，返回友好提示
|
||||
log.error(f"AI处理查询失败: {e!s}")
|
||||
yield self._friendly_error_message(e)
|
||||
|
||||
def _friendly_error_message(self, e: Exception) -> str:
|
||||
"""将 OpenAI 或网络异常转换为友好的中文提示。"""
|
||||
# 尝试获取状态码与错误体
|
||||
status_code = getattr(e, "status_code", None)
|
||||
body = getattr(e, "body", None)
|
||||
message = None
|
||||
error_type = None
|
||||
error_code = None
|
||||
try:
|
||||
if isinstance(body, dict) and "error" in body:
|
||||
err = body.get("error") or {}
|
||||
error_type = err.get("type")
|
||||
error_code = err.get("code")
|
||||
message = err.get("message")
|
||||
except Exception:
|
||||
raise CustomException(f"解析 OpenAI 错误失败: {e!s}")
|
||||
|
||||
text = str(e)
|
||||
msg = message or text
|
||||
|
||||
# 特定错误映射
|
||||
# 欠费/账户状态异常
|
||||
if (
|
||||
(error_code == "Arrearage")
|
||||
or (error_type == "Arrearage")
|
||||
or ("in good standing" in (msg or ""))
|
||||
):
|
||||
return "账户欠费或结算异常，访问被拒绝。请检查账号状态或更换有效的 API Key。"
|
||||
# 鉴权失败
|
||||
if status_code == 401 or "invalid api key" in msg.lower():
|
||||
return "鉴权失败，API Key 无效或已过期。请检查系统配置中的 API Key。"
|
||||
# 权限不足或被拒绝
|
||||
if status_code == 403 or error_type in {
|
||||
"PermissionDenied",
|
||||
"permission_denied",
|
||||
}:
|
||||
return "访问被拒绝，权限不足或账号受限。请检查账户权限设置。"
|
||||
# 配额不足或限流
|
||||
if status_code == 429 or error_type in {
|
||||
"insufficient_quota",
|
||||
"rate_limit_exceeded",
|
||||
}:
|
||||
return "请求过于频繁或配额已用尽。请稍后重试或提升账户配额。"
|
||||
# 客户端错误
|
||||
if status_code == 400:
|
||||
return f"请求参数错误或服务拒绝：{message or '请检查输入内容。'}"
|
||||
# 服务端错误
|
||||
if status_code in {500, 502, 503, 504}:
|
||||
return "服务暂时不可用，请稍后重试。"
|
||||
|
||||
# 默认兜底
|
||||
return f"处理您的请求时出现错误：{msg}"
|
||||
Reference in New Issue
Block a user