diff --git a/.env.example b/.env.example index c601321..94e4d8f 100644 --- a/.env.example +++ b/.env.example @@ -1,6 +1,38 @@ +# ModelScope 配置 MODELSCOPE_API_KEY= MODELSCOPE_API_BASE=https://api-inference.modelscope.cn/v1 MODELSCOPE_MODEL_ID= + +# 数据库配置 DATABASE_URL=mysql+aiomysql://root:123456@localhost:3306/fastagent + +# MCP 服务配置 MCP_SERVER_URL=http://localhost:8001 + +# 日志配置 LOG_LEVEL=INFO + +# FastAPI 应用配置 +APP_TITLE=FastAgent System +APP_DESCRIPTION=A powerful Agent system built with FastAPI and LangChain +APP_VERSION=1.0.0 + +# 服务器配置 +SERVER_HOST=0.0.0.0 +SERVER_PORT=8000 +SERVER_RELOAD=True + +# Agent 配置 +AGENT_TEMPERATURE=0.3 +AGENT_MAX_ITERATIONS=10 + +# 数据库工具配置 +DB_TOOL_NAME=database_query +DB_TOOL_DESCRIPTION=Execute SQL queries on database. Use this tool whenever you need to query database information. Example: 'SELECT * FROM user' to get all user information. +DB_QUERY_LIMIT=100 +DB_QUERY_MAX_LENGTH=1000 + +# MCP 工具配置 +MCP_TOOL_NAME=mcp_tool +MCP_TOOL_DESCRIPTION=Execute MCP (Model Context Protocol) tools to perform various tasks +MCP_REQUEST_TIMEOUT=30.0 diff --git a/app/agent/base.py b/app/agent/base.py index 775de72..b64b0fc 100644 --- a/app/agent/base.py +++ b/app/agent/base.py @@ -10,7 +10,7 @@ import json class BaseAgent(ABC): - def __init__(self, model_name: str = None, temperature: float = 0.3): + def __init__(self, model_name: str = None, temperature: float = None): model = model_name or settings.modelscope_model_id self.llm = ChatOpenAI( model=model, @@ -59,7 +59,7 @@ def _create_agent(self) -> AgentExecutor: tools=self.tools, verbose=True, return_only_outputs=False, - max_iterations=10, + max_iterations=settings.agent_max_iterations, early_stopping_method="force", handle_parsing_errors=True ) @@ -150,10 +150,17 @@ async def execute(self, task: str, context: Optional[Dict[str, Any]] = None) -> history.result = json.dumps(result) await session.commit() + # 确保 data 对象包含 input 和 output 属性 + data = result or {} + if "input" not in data: + data["input"] = task + if "output" not in data: + data["output"] = data.get("result", "") + return { "success": True, "message": "Task completed successfully", - "data": result or {}, + "data": data, "error": None } except Exception as e: @@ -199,6 +206,9 @@ async def execute(self, task: str, context: Optional[Dict[str, Any]] = None) -> return { "success": False, "message": "Task failed", - "data": {}, + "data": { + "input": task, + "output": "" + }, "error": str(e) } diff --git a/app/api/routes.py b/app/api/routes.py index 35708da..1172bb5 100644 --- a/app/api/routes.py +++ b/app/api/routes.py @@ -17,6 +17,10 @@ async def execute_agent(request: AgentExecuteRequest): return AgentResponse( success=False, message="Unknown agent type", + data={ + "input": request.task, + "output": "" + }, error=f"Unknown agent type: {request.agent_type}" ) @@ -26,12 +30,20 @@ async def execute_agent(request: AgentExecuteRequest): return AgentResponse( success=False, message="Validation error", + data={ + "input": request.task if hasattr(request, 'task') else "", + "output": "" + }, error=str(e) ) except Exception as e: return AgentResponse( success=False, message="Task execution failed", + data={ + "input": request.task if hasattr(request, 'task') else "", + "output": "" + }, error=str(e) ) @@ -45,7 +57,11 @@ async def execute_mcp_tool(request: MCPToolRequest): return AgentResponse( success=True, message="MCP tool executed", - data={"result": result} + data={ + "input": f"Tool: {request.tool_name}, Params: {request.parameters}", + "output": result + }, + error=None ) @router.post("/agent/database/query", response_model=AgentResponse) @@ -58,7 +74,11 @@ async def execute_database_query(request: DatabaseQueryRequest): return AgentResponse( success=True, message="Database query executed", - data={"result": result} + data={ + "input": request.query, + "output": result + }, + error=None ) @router.get("/agent/health") @@ -66,5 +86,9 @@ async def health_check(): return AgentResponse( success=True, message="Agent system is healthy", - data={"status": "running"} - ) + data={ + "input": "Health check", + "output": "Agent system is running" + }, + error=None + ) \ No newline at end of file diff --git a/app/config.py b/app/config.py index 97328d2..36105f3 100644 --- a/app/config.py +++ b/app/config.py @@ -1,14 +1,47 @@ from pydantic_settings import BaseSettings class Settings(BaseSettings): + # ModelScope 配置 modelscope_api_key: str modelscope_api_base: str modelscope_model_id: str + + # 数据库配置 database_url: str + + # MCP 服务配置 mcp_server_url: str - log_level: str = "INFO" + + # 日志配置 + log_level: str + + # FastAPI 应用配置 + app_title: str + app_description: str + app_version: str + + # 服务器配置 + server_host: str + server_port: int + server_reload: bool + + # Agent 配置 + agent_temperature: float + agent_max_iterations: int + + # 数据库工具配置 + db_tool_name: str + db_tool_description: str + db_query_limit: int + db_query_max_length: int + + # MCP 工具配置 + mcp_tool_name: str + mcp_tool_description: str + mcp_request_timeout: float class Config: env_file = ".env" + case_sensitive = False settings = Settings() diff --git a/app/main.py b/app/main.py index 5aaf0a3..48c40df 100644 --- a/app/main.py +++ b/app/main.py @@ -8,9 +8,9 @@ logger = logging.getLogger(__name__) app = FastAPI( - title="FastAgent System", - description="A powerful Agent system built with FastAPI and LangChain", - version="1.0.0" + title=settings.app_title, + description=settings.app_description, + version=settings.app_version ) app.add_middleware( @@ -34,8 +34,8 @@ async def shutdown_event(): @app.get("/") async def root(): return { - "message": "Welcome to FastAgent System", - "version": "1.0.0", + "message": f"Welcome to {settings.app_title}", + "version": settings.app_version, "docs": "/docs" } @@ -43,7 +43,7 @@ async def root(): import uvicorn uvicorn.run( "app.main:app", - host="0.0.0.0", - port=8000, - reload=True + host=settings.server_host, + port=settings.server_port, + reload=settings.server_reload ) diff --git a/app/tools/db_tools.py b/app/tools/db_tools.py index eb266af..708998f 100644 --- a/app/tools/db_tools.py +++ b/app/tools/db_tools.py @@ -1,15 +1,16 @@ from langchain.tools import BaseTool from typing import Type, Optional -from pydantic import BaseModel, Field, field_validator +from pydantic.v1 import BaseModel, Field, field_validator from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import text from app.database.connection import get_db +from app.config import settings import re class DatabaseQueryInput(BaseModel): query: str = Field(description="SQL query to execute") params: Optional[dict] = Field(default=None, description="Query parameters") - limit: Optional[int] = Field(default=100, ge=1, le=1000, description="Maximum number of rows to return") + limit: Optional[int] = Field(default=None, ge=1, le=1000, description="Maximum number of rows to return") @field_validator('query') def validate_sql_query(cls, v): @@ -33,22 +34,23 @@ def validate_sql_query(cls, v): raise ValueError(f"Potential SQL injection detected in query: {v}") # 限制查询长度 - if len(v) > 1000: - raise ValueError("Query too long, maximum length is 1000 characters") + if len(v) > settings.db_query_max_length: + raise ValueError(f"Query too long, maximum length is {settings.db_query_max_length} characters") return v class DatabaseQueryTool(BaseTool): - name = "database_query" - description = "Execute SQL queries on the database. Use this tool whenever you need to query database information. Example: 'SELECT * FROM user' to get all user information." + name = settings.db_tool_name + description = settings.db_tool_description args_schema: Type[BaseModel] = DatabaseQueryInput - async def _arun(self, query: str, params: Optional[dict] = None, limit: int = 100) -> str: + async def _arun(self, query: str, params: Optional[dict] = None, limit: int = None) -> str: + actual_limit = limit if limit is not None else settings.db_query_limit async for session in get_db(): try: result = await session.execute(text(query), params or {}) # 限制返回结果的行数 - rows = result.fetchmany(limit) + rows = result.fetchmany(actual_limit) columns = result.keys() formatted_results = [] @@ -56,13 +58,13 @@ async def _arun(self, query: str, params: Optional[dict] = None, limit: int = 10 formatted_results.append(dict(zip(columns, row))) # 如果结果被截断，添加提示信息 - if len(rows) == limit: - return f"Query executed successfully. Results (limited to {limit} rows): {formatted_results}" + if len(rows) == actual_limit: + return f"Query executed successfully. Results (limited to {actual_limit} rows): {formatted_results}" else: return f"Query executed successfully. Results: {formatted_results}" except Exception as e: return f"Error executing query: {str(e)}" - def _run(self, query: str, params: Optional[dict] = None, limit: int = 100) -> str: + def _run(self, query: str, params: Optional[dict] = None, limit: int = None) -> str: import asyncio return asyncio.run(self._arun(query, params, limit)) diff --git a/app/tools/mcp_tools.py b/app/tools/mcp_tools.py index 2ba68ad..70eda81 100644 --- a/app/tools/mcp_tools.py +++ b/app/tools/mcp_tools.py @@ -1,6 +1,6 @@ from langchain.tools import BaseTool from typing import Type, Optional -from pydantic import BaseModel, Field +from pydantic.v1 import BaseModel, Field import httpx from app.config import settings @@ -9,8 +9,8 @@ class MCPToolInput(BaseModel): parameters: dict = Field(description="Parameters for the MCP tool") class MCPToolWrapper(BaseTool): - name = "mcp_tool" - description = "Execute MCP (Model Context Protocol) tools to perform various tasks" + name = settings.mcp_tool_name + description = settings.mcp_tool_description args_schema: Type[BaseModel] = MCPToolInput async def _arun(self, tool_name: str, parameters: dict) -> str: @@ -19,7 +19,7 @@ async def _arun(self, tool_name: str, parameters: dict) -> str: response = await client.post( f"{settings.mcp_server_url}/tools/{tool_name}", json=parameters, - timeout=30.0 + timeout=settings.mcp_request_timeout ) response.raise_for_status() return f"MCP tool '{tool_name}' executed successfully: {response.json()}"