diff --git a/.env.example b/.env.example deleted file mode 100644 index f6f27ba..0000000 --- a/.env.example +++ /dev/null @@ -1,6 +0,0 @@ -# OpenAI API Configuration -OPENAI_API_KEY=your_openai_api_key_here - -# Other API Keys (optional) -# ANTHROPIC_API_KEY=your_anthropic_api_key_here -# GROQ_API_KEY=your_groq_api_key_here \ No newline at end of file diff --git a/.gitignore b/.gitignore index d278616..f7d85eb 100644 --- a/.gitignore +++ b/.gitignore @@ -14,4 +14,7 @@ __pycache__/ # 项目临时文件 .DS_Store -*.log \ No newline at end of file +*.log + +# 本地配置 +config/config.ini diff --git a/README.md b/README.md index 1aff3e0..879641a 100644 --- a/README.md +++ b/README.md @@ -9,30 +9,51 @@ - 💬 **多轮对话**: 内置对话状态管理和上下文维护 - 📊 **工作流管理**: 多种工作流类型，支持会话和工具使用 - ⚙️ **配置管理**: 统一的环境变量和配置管理 +- 🧩 **FastAPI 接入**: 提供 HTTP 接口对外服务 +- 🧭 **Nacos 注册**: 支持服务注册与心跳 - 🧪 **测试支持**: 包含基础测试和示例代码 ## 项目结构 ``` more_dots/ -├── agents/ # 代理模块 -│ ├── base_agent.py # 基础代理类 -│ ├── conversation_agent.py # 对话代理 -│ └── tool_agent.py # 工具使用代理 -├── tools/ # 工具模块 -│ ├── calculator.py # 计算器工具 -│ └── web_search.py # 网络搜索工具（占位符） -├── workflows/ # 工作流管理 -│ └── workflow_manager.py # 工作流管理器 -├── examples/ # 使用示例 -│ └── basic_usage.py # 基础用法示例 -├── tests/ # 测试文件 -│ └── test_basic.py # 基础测试 -├── config.py # 配置文件 +├── agent/ # Agent 核心逻辑层 +│ ├── graph.py # LangGraph 图结构定义 +│ ├── nodes.py # 节点执行逻辑 +│ ├── state.py # Agent 状态定义 +│ ├── conversation.py # 对话代理 +│ └── tool.py # 工具代理 +├── api/ # API 接口层 +│ ├── endpoints.py # FastAPI 路由定义 +│ └── dependencies.py # API 依赖注入 +├── services/ # 服务层 +│ ├── llm_factory.py # LLM 实例工厂 +│ └── nacos_service.py # Nacos 集成 +├── schemas/ # 数据模型层 +│ ├── agent_input.py # 输入模型 +│ └── agent_output.py # 输出模型 +├── config/ # 配置层 +│ └── settings.py # 配置读取 +│ └── prompts.yaml # 提示词配置 +│ └── ragflow_templates/ # RAGFlow 模板（表名 -> 模板列表） +│ └── table_metadata_prompts/ # 表模型元数据提示词 +├── tools/ # 工具模块 +│ ├── calculator.py # 计算器工具 +│ └── web_search.py # 网络搜索工具（占位符） +├── workflows/ # 工作流管理 +│ └── workflow_manager.py +├── examples/ # 使用示例 +│ └── basic_usage.py +├── tests/ # 测试文件 +│ └── test_basic.py ├── requirements.txt # 依赖包列表 -├── .env.example # 环境变量示例 -├── main.py # 主程序入口 -└── README.md # 项目说明 +├── config/ +│ ├── config.ini.example # 配置文件示例 +│ ├── config.ini # 本地配置（需自行创建） +│ └── prompts.yaml # 提示词配置 +├── server.py # FastAPI 服务入口 +├── main.py # CLI 入口 +└── README.md # 项目说明 ``` ## 快速开始 @@ -43,14 +64,25 @@ more_dots/ pip install -r requirements.txt ``` -### 2. 配置环境变量 +### 2. 配置 config.ini ```bash -# 复制环境变量文件 -cp .env.example .env +# 复制配置文件 +cp config/config.ini.example config/config.ini -# 编辑 .env 文件，设置你的 OpenAI API 密钥 -OPENAI_API_KEY=your_openai_api_key_here +# 编辑 config/config.ini，设置你的 API Key，并可添加多个模型配置 +[General] +DEFAULT_MODEL_SECTION = gpt-4o +MAX_RETRIES = 3 +TIMEOUT = 30 + +[gpt-4o] +MODEL_NAME = gpt-4o +OPENAI_API_KEY = your_openai_api_key_here + +[gpt-3.5-turbo] +MODEL_NAME = gpt-3.5-turbo +OPENAI_API_KEY = your_openai_api_key_here ``` ### 3. 运行示例 @@ -59,8 +91,14 @@ OPENAI_API_KEY=your_openai_api_key_here # 运行基础示例 python examples/basic_usage.py -# 运行交互式 CLI +# 运行交互式 CLI（默认模型） python main.py + +# 运行交互式 CLI（指定模型配置段） +python main.py gpt-3.5-turbo + +# 运行 FastAPI 服务 +python server.py ``` ## 使用指南 @@ -70,9 +108,55 @@ python main.py ```python from workflows.workflow_manager import WorkflowManager, WorkflowType -# 创建工作流管理器 +# 创建工作流管理器（默认模型） manager = WorkflowManager() +# 创建工作流管理器（指定模型配置段） +manager_alt = WorkflowManager(default_model_section="gpt-3.5-turbo") + +### FastAPI 接口 + +启动服务后，可使用以下接口： + +- `GET /health`：健康检查 +- `GET /nacos/status`：查看 Nacos 注册状态 +- `POST /api/workflows`：执行工作流 + +示例请求体： + +```json +{ + "input": "你好，帮我算 1 + 2", + "session_id": null, + "workflow_type": "conversation" +} +``` + +### Nacos 配置 + +在 `config/config.ini` 中开启 Nacos： + +```ini +[nacos] +enabled = true +server = localhost:8848 +namespace = public +group_name = DEFAULT_GROUP +cluster_name = DEFAULT +heartbeat_interval = 5 +``` + +### RAGFlow 模板同步 + +模板文件位于 `config/ragflow_templates`，每个 JSON 对应一个表名与模板列表。 +同步脚本： + +```bash +python scripts/sync_ragflow_templates.py +``` + +请在 `config/config.ini` 中配置 `ragflow.upload` 上传接口。 + # 使用对话工作流 result = manager.execute_workflow( WorkflowType.CONVERSATION, @@ -161,7 +245,6 @@ logging.basicConfig(level=logging.DEBUG) - `langchain`: LangChain 主包 - `langgraph`: LangGraph 图工作流 - `langchain-openai`: OpenAI 集成 -- `python-dotenv`: 环境变量管理 - `pydantic`: 数据验证 ## 许可证 diff --git a/_trial_temp/_trial_marker b/_trial_temp/_trial_marker new file mode 100644 index 0000000..e69de29 diff --git a/agent/__init__.py b/agent/__init__.py new file mode 100644 index 0000000..64affb5 --- /dev/null +++ b/agent/__init__.py @@ -0,0 +1,6 @@ +from .state import AgentState +from .graph import BaseAgent +from .conversation import ConversationAgent +from .tool import ToolAgent + +__all__ = ["AgentState", "BaseAgent", "ConversationAgent", "ToolAgent"] diff --git a/agent/conversation.py b/agent/conversation.py new file mode 100644 index 0000000..0424d09 --- /dev/null +++ b/agent/conversation.py @@ -0,0 +1,106 @@ +from typing import Dict, Any, List, Optional +from langchain_core.messages import BaseMessage, HumanMessage, AIMessage +from langgraph.graph import StateGraph, END + +from .graph import BaseAgent +from .state import AgentState + + +class ConversationAgent(BaseAgent): + """处理多轮对话的代理""" + + def __init__(self, model_section: Optional[str] = None): + super().__init__(model_section) + self.conversation_history: List[BaseMessage] = [] + + def _build_graph(self) -> StateGraph: + """构建对话专用图""" + workflow = StateGraph(AgentState) + + workflow.add_node("analyze_intent", self._analyze_intent) + workflow.add_node("normalize_input", self._normalize_input) + workflow.add_node("generate_response", self._generate_response) + workflow.add_node("update_context", self._update_context) + + workflow.add_edge("analyze_intent", "normalize_input") + workflow.add_edge("normalize_input", "generate_response") + workflow.add_edge("generate_response", "update_context") + workflow.add_edge("update_context", END) + + workflow.set_entry_point("analyze_intent") + + return workflow.compile() + + def _analyze_intent(self, state: AgentState) -> AgentState: + """分析用户意图与对话上下文""" + user_message = state.messages[-1] if state.messages else None + + if user_message and isinstance(user_message, HumanMessage): + content = user_message.content.lower() + + if any(word in content for word in ["hello", "hi", "hey", "greetings"]): + state.context["intent"] = "greeting" + elif any(word in content for word in ["help", "assist", "support"]): + state.context["intent"] = "help" + elif "?" in content: + state.context["intent"] = "question" + else: + state.context["intent"] = "general" + + state.current_step = "intent_analyzed" + return state + + def _generate_response(self, state: AgentState) -> AgentState: + """结合对话历史生成回复""" + all_messages = self.conversation_history + state.messages + + if all_messages: + response = self.model.invoke(all_messages) + state.messages.append(response) + + state.current_step = "response_generated" + return state + + def _update_context(self, state: AgentState) -> AgentState: + """更新对话上下文与历史""" + for message in state.messages: + if isinstance(message, (HumanMessage, AIMessage)): + self.conversation_history.append(message) + + if len(self.conversation_history) > 10: + self.conversation_history = self.conversation_history[-10:] + + state.current_step = "context_updated" + return state + + def run(self, user_input: str, **kwargs) -> Dict[str, Any]: + """运行对话并维护历史""" + initial_state = AgentState( + messages=[HumanMessage(content=user_input)], + context=kwargs + ) + + result = self.graph.invoke(initial_state) + + return { + "messages": result.get("messages", []), + "context": result.get("context", {}), + "conversation_history": self.conversation_history, + "final_step": result.get("current_step", "unknown") + } + + def stream_run(self, user_input: str): + """流式运行对话并维护历史""" + all_messages = self.conversation_history + [HumanMessage(content=user_input)] + full_text = "" + + for chunk in self.model.stream(all_messages): + if hasattr(chunk, "content") and chunk.content: + full_text += chunk.content + yield chunk.content + + self.conversation_history.append(HumanMessage(content=user_input)) + self.conversation_history.append(AIMessage(content=full_text)) + + if len(self.conversation_history) > 10: + self.conversation_history = self.conversation_history[-10:] diff --git a/agent/graph.py b/agent/graph.py new file mode 100644 index 0000000..b73d08a --- /dev/null +++ b/agent/graph.py @@ -0,0 +1,54 @@ +from typing import Any, Dict, Optional +from langchain_core.messages import HumanMessage +from langgraph.graph import StateGraph, END + +from services.llm_factory import create_chat_model +from .state import AgentState +from . import nodes + + +class BaseAgent: + """包含通用功能的基础代理类""" + + def __init__(self, model_section: Optional[str] = None): + self.model = create_chat_model(model_section) + self.graph = self._build_graph() + + def _build_graph(self) -> StateGraph: + """构建代理状态图""" + workflow = StateGraph(AgentState) + + workflow.add_node("process_input", nodes.process_input) + workflow.add_node("normalize_input", self._normalize_input) + workflow.add_node("generate_response", self._generate_response) + + workflow.add_edge("process_input", "normalize_input") + workflow.add_edge("normalize_input", "generate_response") + workflow.add_edge("generate_response", END) + + workflow.set_entry_point("process_input") + + return workflow.compile() + + def _generate_response(self, state: AgentState) -> AgentState: + """使用 LLM 生成回复""" + return nodes.generate_response(state, self.model) + + def _normalize_input(self, state: AgentState) -> AgentState: + """规范化用户输入""" + return nodes.normalize_input(state, self.model) + + def run(self, user_input: str, **kwargs) -> Dict[str, Any]: + """运行代理并处理用户输入""" + initial_state = AgentState( + messages=[HumanMessage(content=user_input)], + context=kwargs + ) + + result = self.graph.invoke(initial_state) + + return { + "messages": result.get("messages", []), + "context": result.get("context", {}), + "final_step": result.get("current_step", "unknown") + } diff --git a/agent/nodes.py b/agent/nodes.py new file mode 100644 index 0000000..136f89a --- /dev/null +++ b/agent/nodes.py @@ -0,0 +1,43 @@ +from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage +from .state import AgentState +from services.prompt_manager import PromptManager +from services.template_matcher import TemplateMatcher + + +def process_input(state: AgentState) -> AgentState: + """处理用户输入""" + state.current_step = "processed" + return state + + +def generate_response(state: AgentState, model) -> AgentState: + """使用 LLM 生成回复""" + if state.messages: + response = model.invoke(state.messages) + state.messages.append(response) + return state + + +def normalize_input(state: AgentState, model) -> AgentState: + """将用户输入规范化为标准英文语句""" + if not state.messages: + return state + + last_message = state.messages[-1] + if not isinstance(last_message, HumanMessage): + return state + + prompt_manager = PromptManager() + system_prompt = SystemMessage( + content=prompt_manager.get("system", "english_normalizer") + ) + + response = model.invoke([system_prompt, HumanMessage(content=last_message.content)]) + normalized = response.content if hasattr(response, "content") else str(response) + + state.context["original_input"] = last_message.content + state.context["normalized_input"] = normalized + + matcher = TemplateMatcher() + state.context["table_match"] = matcher.match(normalized) + return state diff --git a/agent/state.py b/agent/state.py new file mode 100644 index 0000000..8ca1fc4 --- /dev/null +++ b/agent/state.py @@ -0,0 +1,14 @@ +from typing import Any, Dict, List +from langchain_core.messages import BaseMessage + + +class AgentState: + """代理工作流的状态定义""" + messages: List[BaseMessage] + current_step: str + context: Dict[str, Any] + + def __init__(self, messages: List[BaseMessage] = None, current_step: str = "start", context: Dict[str, Any] = None): + self.messages = messages or [] + self.current_step = current_step + self.context = context or {} diff --git a/agent/tool.py b/agent/tool.py new file mode 100644 index 0000000..06ea8e8 --- /dev/null +++ b/agent/tool.py @@ -0,0 +1,91 @@ +from typing import Dict, Any, List, Optional +from langchain_core.messages import BaseMessage, HumanMessage +from langchain_core.tools import BaseTool +from langgraph.graph import StateGraph, END +from langgraph.prebuilt import ToolNode + +from .graph import BaseAgent +from .state import AgentState +from tools.calculator import CalculatorTool +from tools.web_search import WebSearchTool +from tools.rest_api_tool import RestApiTool +from tools.sr_api_tool import SrApiQueryTool + + +class ToolAgent(BaseAgent): + """可使用工具完成任务的代理""" + + def __init__(self, model_section: Optional[str] = None, tools: List[BaseTool] = None): + if tools is None: + tools = [CalculatorTool(), WebSearchTool(), RestApiTool(), SrApiQueryTool()] + + self.tools = tools + self.tool_node = ToolNode(tools) + super().__init__(model_section) + + def _build_graph(self) -> StateGraph: + """构建可使用工具的图""" + workflow = StateGraph(AgentState) + + workflow.add_node("normalize_input", self._normalize_input) + workflow.add_node("agent", self._agent_node) + workflow.add_node("tools", self.tool_node) + + workflow.add_edge("normalize_input", "agent") + workflow.add_edge("tools", "agent") + + workflow.add_conditional_edges( + "agent", + self._should_use_tools, + { + "tools": "tools", + "end": END, + } + ) + + workflow.set_entry_point("normalize_input") + + return workflow.compile() + + def _agent_node(self, state: AgentState) -> AgentState: + """决定是否调用工具的代理节点""" + model_with_tools = self.model.bind_tools(self.tools) + + if state.messages: + try: + response = model_with_tools.invoke(state.messages) + except Exception as e: + error_text = str(e) + if "tool choice" in error_text and "auto" in error_text: + fallback_model = self.model.bind_tools(self.tools, tool_choice="none") + response = fallback_model.invoke(state.messages) + else: + raise + state.messages.append(response) + + return state + + def _should_use_tools(self, state: AgentState) -> str: + """判断是否需要使用工具""" + last_message = state.messages[-1] + + if hasattr(last_message, 'tool_calls') and last_message.tool_calls: + return "tools" + + return "end" + + def run(self, user_input: str, **kwargs) -> Dict[str, Any]: + """运行工具型代理""" + initial_state = AgentState( + messages=[HumanMessage(content=user_input)], + context=kwargs + ) + + result = self.graph.invoke(initial_state) + + return { + "messages": result.get("messages", []), + "context": result.get("context", {}), + "tools_used": [tool.name for tool in self.tools], + "final_step": result.get("current_step", "unknown") + } diff --git a/agent/utils.py b/agent/utils.py new file mode 100644 index 0000000..2449dbe --- /dev/null +++ b/agent/utils.py @@ -0,0 +1 @@ +# Agent 内部工具函数（按需扩展） diff --git a/agents/base_agent.py b/agents/base_agent.py index 9c7fd23..c2acc4d 100644 --- a/agents/base_agent.py +++ b/agents/base_agent.py @@ -1,76 +1,6 @@ -from typing import Any, Dict, List, Optional -from langchain_core.messages import BaseMessage, HumanMessage -from langchain_openai import ChatOpenAI -from langgraph.graph import StateGraph, END -from config import Config +"""兼容导出：请优先使用 agent 包""" +from agent.graph import BaseAgent +from agent.state import AgentState -class AgentState: - """State definition for the agent workflow""" - messages: List[BaseMessage] - current_step: str - context: Dict[str, Any] - - def __init__(self, messages: List[BaseMessage] = None, current_step: str = "start", context: Dict[str, Any] = None): - self.messages = messages or [] - self.current_step = current_step - self.context = context or {} - - -class BaseAgent: - """Base agent class with common functionality""" - - def __init__(self, model_name: str = Config.DEFAULT_MODEL): - self.model = ChatOpenAI( - model=model_name, - api_key=Config.OPENAI_API_KEY, - temperature=0.1, - max_retries=Config.MAX_RETRIES, - timeout=Config.TIMEOUT - ) - self.graph = self._build_graph() - - def _build_graph(self) -> StateGraph: - """Build the state graph for the agent""" - workflow = StateGraph(AgentState) - - # Add nodes and edges - workflow.add_node("process_input", self._process_input) - workflow.add_node("generate_response", self._generate_response) - - # Define edges - workflow.add_edge("process_input", "generate_response") - workflow.add_edge("generate_response", END) - - # Set entry point - workflow.set_entry_point("process_input") - - return workflow.compile() - - def _process_input(self, state: AgentState) -> AgentState: - """Process user input""" - # This is a base implementation - subclasses should override - state.current_step = "processed" - return state - - def _generate_response(self, state: AgentState) -> AgentState: - """Generate response using the LLM""" - if state.messages: - response = self.model.invoke(state.messages) - state.messages.append(response) - return state - - def run(self, user_input: str, **kwargs) -> Dict[str, Any]: - """Run the agent with user input""" - initial_state = AgentState( - messages=[HumanMessage(content=user_input)], - context=kwargs - ) - - result = self.graph.invoke(initial_state) - - return { - "messages": result.messages, - "context": result.context, - "final_step": result.current_step - } \ No newline at end of file +__all__ = ["BaseAgent", "AgentState"] \ No newline at end of file diff --git a/agents/conversation_agent.py b/agents/conversation_agent.py index 36e0cce..6de1eff 100644 --- a/agents/conversation_agent.py +++ b/agents/conversation_agent.py @@ -1,44 +1,44 @@ -from typing import Dict, Any, List +from typing import Dict, Any, List, Optional from langchain_core.messages import BaseMessage, HumanMessage, AIMessage from langgraph.graph import StateGraph, END from .base_agent import BaseAgent, AgentState class ConversationAgent(BaseAgent): - """Agent for handling multi-turn conversations""" + """处理多轮对话的代理""" - def __init__(self, model_name: str = None): - super().__init__(model_name) + def __init__(self, model_section: Optional[str] = None): + super().__init__(model_section) self.conversation_history: List[BaseMessage] = [] def _build_graph(self) -> StateGraph: - """Build conversation-specific graph""" + """构建对话专用图""" workflow = StateGraph(AgentState) - # Add nodes + # 添加节点 workflow.add_node("analyze_intent", self._analyze_intent) workflow.add_node("generate_response", self._generate_response) workflow.add_node("update_context", self._update_context) - # Define edges + # 定义边 workflow.add_edge("analyze_intent", "generate_response") workflow.add_edge("generate_response", "update_context") workflow.add_edge("update_context", END) - # Set entry point + # 设置入口节点 workflow.set_entry_point("analyze_intent") return workflow.compile() def _analyze_intent(self, state: AgentState) -> AgentState: - """Analyze user intent and conversation context""" - # Simple intent analysis - can be enhanced with more sophisticated logic + """分析用户意图与对话上下文""" + # 简单意图分析，可用更复杂逻辑增强 user_message = state.messages[-1] if state.messages else None if user_message and isinstance(user_message, HumanMessage): content = user_message.content.lower() - # Basic intent detection + # 基础意图识别 if any(word in content for word in ["hello", "hi", "hey", "greetings"]): state.context["intent"] = "greeting" elif any(word in content for word in ["help", "assist", "support"]): @@ -46,49 +46,8 @@ class ConversationAgent(BaseAgent): elif "?" in content: state.context["intent"] = "question" else: - state.context["intent"] = "general" - - state.current_step = "intent_analyzed" - return state - - def _generate_response(self, state: AgentState) -> AgentState: - """Generate response considering conversation history""" - # Combine conversation history with current message - all_messages = self.conversation_history + state.messages - - if all_messages: - response = self.model.invoke(all_messages) - state.messages.append(response) - - state.current_step = "response_generated" - return state - - def _update_context(self, state: AgentState) -> AgentState: - """Update conversation context and history""" - # Add the conversation to history (excluding system messages) - for message in state.messages: - if isinstance(message, (HumanMessage, AIMessage)): - self.conversation_history.append(message) - - # Limit conversation history to avoid token limits - if len(self.conversation_history) > 10: - self.conversation_history = self.conversation_history[-10:] - - state.current_step = "context_updated" - return state - - def run(self, user_input: str, **kwargs) -> Dict[str, Any]: - """Run conversation with history management""" - initial_state = AgentState( - messages=[HumanMessage(content=user_input)], - context=kwargs - ) - - result = self.graph.invoke(initial_state) - - return { - "messages": result.messages, - "context": result.context, - "conversation_history": self.conversation_history, - "final_step": result.current_step - } \ No newline at end of file + """兼容导出：请优先使用 agent 包""" + + from agent.conversation import ConversationAgent + + __all__ = ["ConversationAgent"] \ No newline at end of file diff --git a/agents/tool_agent.py b/agents/tool_agent.py index 915ee51..db73474 100644 --- a/agents/tool_agent.py +++ b/agents/tool_agent.py @@ -1,86 +1,5 @@ -from typing import Dict, Any, List, Optional -from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage -from langchain_core.tools import BaseTool -from langgraph.graph import StateGraph, END -from langgraph.prebuilt import ToolNode -from .base_agent import BaseAgent, AgentState -from tools.calculator import CalculatorTool -from tools.web_search import WebSearchTool +"""兼容导出：请优先使用 agent 包""" +from agent.tool import ToolAgent -class ToolAgent(BaseAgent): - """Agent that can use tools to accomplish tasks""" - - def __init__(self, model_name: str = None, tools: List[BaseTool] = None): - # Initialize with default tools if none provided - if tools is None: - tools = [CalculatorTool(), WebSearchTool()] - - self.tools = tools - self.tool_node = ToolNode(tools) - super().__init__(model_name) - - def _build_graph(self) -> StateGraph: - """Build tool-using graph""" - workflow = StateGraph(AgentState) - - # Add nodes - workflow.add_node("agent", self._agent_node) - workflow.add_node("tools", self.tool_node) - - # Define edges - workflow.add_edge("tools", "agent") - - # Conditional routing - workflow.add_conditional_edges( - "agent", - self._should_use_tools, - { - "tools": "tools", - "end": END, - } - ) - - # Set entry point - workflow.set_entry_point("agent") - - return workflow.compile() - - def _agent_node(self, state: AgentState) -> AgentState: - """Agent node that decides whether to use tools""" - # Bind tools to the model - model_with_tools = self.model.bind_tools(self.tools) - - # Get the last message - if state.messages: - response = model_with_tools.invoke(state.messages) - state.messages.append(response) - - return state - - def _should_use_tools(self, state: AgentState) -> str: - """Determine if tools should be used""" - last_message = state.messages[-1] - - # If the last message has tool calls, route to tools - if hasattr(last_message, 'tool_calls') and last_message.tool_calls: - return "tools" - - # Otherwise, end the workflow - return "end" - - def run(self, user_input: str, **kwargs) -> Dict[str, Any]: - """Run the tool-using agent""" - initial_state = AgentState( - messages=[HumanMessage(content=user_input)], - context=kwargs - ) - - result = self.graph.invoke(initial_state) - - return { - "messages": result.messages, - "context": result.context, - "tools_used": [tool.name for tool in self.tools], - "final_step": result.current_step - } \ No newline at end of file +__all__ = ["ToolAgent"] \ No newline at end of file diff --git a/api/__init__.py b/api/__init__.py new file mode 100644 index 0000000..c17ce8b --- /dev/null +++ b/api/__init__.py @@ -0,0 +1 @@ +"""API 包""" diff --git a/api/dependencies.py b/api/dependencies.py new file mode 100644 index 0000000..c3c6f16 --- /dev/null +++ b/api/dependencies.py @@ -0,0 +1,17 @@ +from fastapi import Request + + +def get_workflow_manager(request: Request): + return request.app.state.workflow_manager + + +def get_nacos_manager(request: Request): + return request.app.state.nacos_manager + + +def get_service_config(request: Request): + return request.app.state.service_config + + +def get_tool_router(request: Request): + return request.app.state.tool_router diff --git a/api/endpoints.py b/api/endpoints.py new file mode 100644 index 0000000..b371ea3 --- /dev/null +++ b/api/endpoints.py @@ -0,0 +1,82 @@ +from fastapi import APIRouter, HTTPException, Depends +from fastapi.responses import StreamingResponse + +from schemas.agent_input import AgentInput +from schemas.agent_output import AgentOutput +from schemas.tool_input import ToolInput +from schemas.tool_output import ToolOutput +from workflows.workflow_manager import WorkflowType +from api.dependencies import get_workflow_manager, get_nacos_manager, get_service_config, get_tool_router + + +router = APIRouter() + + +def _resolve_workflow_type(value: str) -> WorkflowType: + try: + return WorkflowType(value) + except Exception as e: + raise ValueError(f"不支持的工作流类型: {value}") from e + + +@router.get("/health") +def health_check(service_config=Depends(get_service_config)): + return { + "status": "ok", + "service_name": service_config.service_name, + "model_section": service_config.metadata.get("model_section", "") + } + + +@router.get("/nacos/status") +def nacos_status(nacos_manager=Depends(get_nacos_manager)): + return nacos_manager.status() + + +@router.post("/api/workflows", response_model=AgentOutput) +def run_workflow(payload: AgentInput, workflow_manager=Depends(get_workflow_manager)): + try: + workflow_type = _resolve_workflow_type(payload.workflow_type) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + + result = workflow_manager.execute_workflow( + workflow_type=workflow_type, + user_input=payload.input, + session_id=payload.session_id, + ) + + return AgentOutput( + session_id=result["session_id"], + workflow_type=result["workflow_type"], + result=result["result"], + ) + + +@router.post("/api/workflows/stream") +def run_workflow_stream(payload: AgentInput, workflow_manager=Depends(get_workflow_manager)): + try: + workflow_type = _resolve_workflow_type(payload.workflow_type) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + + if workflow_type != WorkflowType.CONVERSATION: + raise HTTPException(status_code=400, detail="仅支持对话工作流的流式输出") + + agent = workflow_manager.get_workflow(workflow_type) + + def event_stream(): + try: + for token in agent.stream_run(payload.input): + yield f"data: {token}\n\n" + yield "event: end\ndata: [DONE]\n\n" + except Exception as e: + yield f"event: error\ndata: {str(e)}\n\n" + + return StreamingResponse(event_stream(), media_type="text/event-stream") + + +@router.post("/api/tools/execute", response_model=ToolOutput) +def run_tool(payload: ToolInput, tool_router=Depends(get_tool_router)): + result = tool_router.call(payload.tool_name, payload.payload) + return ToolOutput(**result) diff --git a/config.py b/config.py index 0ce7b47..da811b4 100644 --- a/config.py +++ b/config.py @@ -1,27 +1,5 @@ -import os -from dotenv import load_dotenv +"""兼容导出：请优先使用 config 包""" -# Load environment variables -load_dotenv() +from config.settings import Config -class Config: - """Application configuration""" - - # API Keys - OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") - - # Model configurations - DEFAULT_MODEL = "gpt-4o" - - # Application settings - MAX_RETRIES = 3 - TIMEOUT = 30 - - @classmethod - def validate_config(cls): - """Validate that required configuration is present""" - if not cls.OPENAI_API_KEY: - raise ValueError("OPENAI_API_KEY is required. Please set it in your .env file") - -# Validate configuration on import -Config.validate_config() \ No newline at end of file +__all__ = ["Config"] \ No newline at end of file diff --git a/config/__init__.py b/config/__init__.py new file mode 100644 index 0000000..6b76ab9 --- /dev/null +++ b/config/__init__.py @@ -0,0 +1,3 @@ +from .settings import Config + +__all__ = ["Config"] diff --git a/config/config.ini.example b/config/config.ini.example new file mode 100644 index 0000000..0e03063 --- /dev/null +++ b/config/config.ini.example @@ -0,0 +1,68 @@ +[General] +# 默认使用的模型配置区域 +DEFAULT_MODEL_SECTION = gpt-4o +# 最大重试次数 +MAX_RETRIES = 3 +# 请求超时时间 +TIMEOUT = 30 + +[gpt-4o] +# 模型名称 +MODEL_NAME = gpt-4o +# API 密钥 +OPENAI_API_KEY = your_openai_api_key_here + +[gpt-3.5-turbo] +# 模型名称 +MODEL_NAME = gpt-3.5-turbo +# API 密钥 +OPENAI_API_KEY = your_openai_api_key_here + +[app] +# 服务名称 +service_name = more-dots-api +# 监听地址 +host = 0.0.0.0 +# 监听端口 +port = 8000 +# 服务版本 +version = 1.0.0 +# 默认模型配置段（可覆盖 General.DEFAULT_MODEL_SECTION） +model_section = gpt-4o + +[ragflow] +url = http://10.122.176.97:21020 +dataset_ids = f5b8b854d63a11f083230242c0a8e006 +document_ids = 819cf100f52611f0a7fa0242c0a8e006 +api_key = ragflow-xxxxx +retrieval = /api/v1/retrieval +upload = /api/v1/documents +cache_ttl = 600 + +[redis] +enabled = false +url = redis://localhost:6379/0 +db = 0 + +[nacos] +# 是否启用 Nacos 注册 +enabled = false +# Nacos 服务器地址 +server = localhost:8848 +# 命名空间 +namespace = public +# 分组名称 +group_name = DEFAULT_GROUP +# 集群名称 +cluster_name = DEFAULT +# 账号（可选） +username = +# 密码（可选） +password = +# 心跳间隔（秒） +heartbeat_interval = 5 +# 权重 +weight = 1.0 +# 是否临时实例 +ephemeral = true + diff --git a/config/prompts.yaml b/config/prompts.yaml new file mode 100644 index 0000000..8185d6a --- /dev/null +++ b/config/prompts.yaml @@ -0,0 +1,20 @@ +system: + default: | + You are a helpful assistant. Follow the user's instructions carefully. + sql_generator: | + You are an assistant that converts user intent into SQL. + Ensure the SQL is correct, safe, and syntactically valid. + +user: + default: | + Please respond clearly and concisely. + english_normalizer: | + You are a translation and normalization assistant. + Convert the user's input to a clear, grammatically correct English sentence suitable for SQL intent. + If the input is already English, polish it. + Return only the final English sentence without extra explanations. + +business: + sql_generation: | + Based on the user's intent, generate a SQL query. + Use correct table/field names and return only SQL. diff --git a/config/ragflow_templates/README.md b/config/ragflow_templates/README.md new file mode 100644 index 0000000..cb6dd91 --- /dev/null +++ b/config/ragflow_templates/README.md @@ -0,0 +1,4 @@ +本目录存放用于 RAGFlow 匹配的模板配置（JSON）。 + +约定：每个 JSON 文件对应一个表名（key），包含模板列表（value）。 +示例文件：orders.json、customers.json。 diff --git a/config/ragflow_templates/example_table.json b/config/ragflow_templates/example_table.json new file mode 100644 index 0000000..b662350 --- /dev/null +++ b/config/ragflow_templates/example_table.json @@ -0,0 +1,8 @@ +{ + "table": "example_table", + "templates": [ + "example_table created in {date}", + "count of records in example_table", + "example_table where status = {status}" + ] +} diff --git a/config/settings.py b/config/settings.py new file mode 100644 index 0000000..92d4fa2 --- /dev/null +++ b/config/settings.py @@ -0,0 +1,76 @@ +import os +import configparser +from typing import Optional + + +class Config: + """从 config.ini 读取的应用配置""" + + _config = configparser.ConfigParser() + _root_dir = os.path.dirname(os.path.dirname(__file__)) + _config_path = os.path.join(_root_dir, 'config', 'config.ini') + + # 在类初始化时加载配置 + if not os.path.exists(_config_path): + raise FileNotFoundError( + f"Configuration file not found at: {_config_path}. " + "Please copy 'config/config.ini.example' to 'config/config.ini' and fill in your details." + ) + + try: + with open(_config_path, "r", encoding="utf-8") as f: + _config.read_file(f) + except UnicodeDecodeError: + with open(_config_path, "r", encoding="gbk") as f: + _config.read_file(f) + + # 通用设置 + DEFAULT_MODEL_SECTION: str = _config.get('General', 'DEFAULT_MODEL_SECTION', fallback='gpt-4o') + MAX_RETRIES: int = _config.getint('General', 'MAX_RETRIES', fallback=3) + TIMEOUT: int = _config.getint('General', 'TIMEOUT', fallback=30) + + @classmethod + def get_model_config(cls, section: Optional[str] = None) -> dict: + """ + 获取指定模型配置段。 + 若 section 为 None，则使用默认模型配置段。 + """ + if section is None: + section = cls.DEFAULT_MODEL_SECTION + + if not cls._config.has_section(section): + raise ValueError(f"Model section '{section}' not found in config.ini") + + config = dict(cls._config.items(section)) + + # 确保必需的键存在 + if 'model_name' not in config or 'openai_api_key' not in config: + raise ValueError(f"Model section '{section}' must contain 'model_name' and 'openai_api_key'") + + return { + "model": config['model_name'], + "api_key": config['openai_api_key'], + "base_url": config.get('url') or config.get('base_url') + } + + @classmethod + def get_section(cls, section: str) -> dict: + """获取指定配置段的键值对（键名会被转为小写）""" + if not cls._config.has_section(section): + return {} + return dict(cls._config.items(section)) + + @classmethod + def validate_config(cls): + """校验默认模型配置是否存在且有效""" + try: + default_config = cls.get_model_config() + if not default_config.get("api_key") or 'your_openai_api_key_here' in default_config.get("api_key"): + raise ValueError(f"API key for default model '{cls.DEFAULT_MODEL_SECTION}' is missing or a placeholder.") + except (ValueError, configparser.Error) as e: + raise ValueError(f"Configuration validation failed: {e}") + + +# 如有需要可在导入时做初始校验， +# 但已移到 main.py 以便更可控地执行。 +# 如需在导入时校验，可在此调用 Config.validate_config() diff --git a/config/sql_prompts/README.md b/config/sql_prompts/README.md new file mode 100644 index 0000000..a7da3b4 --- /dev/null +++ b/config/sql_prompts/README.md @@ -0,0 +1,4 @@ +本目录存放业务 SQL 提示词（JSON 格式）。 + +约定：每个 JSON 文件对应一个数据库表模型。 +示例文件：order.json、customer.json 等。 diff --git a/config/sql_prompts/apbo_eta_ful.json b/config/sql_prompts/apbo_eta_ful.json new file mode 100644 index 0000000..71fe25f --- /dev/null +++ b/config/sql_prompts/apbo_eta_ful.json @@ -0,0 +1,364 @@ +{ + "meta": { + "domain": "订单物料最新状态与历史变更查询", + "description": "此模型用于查询订单物料的最新状态和ETA相关信息，支持查询历史变更记录。该表包含最新状态数据（data_flag='Newest'）和历史变更数据（data_flag='Changelog'）的联合结果。所有日期字段在查询时自动转换为字符类型。通过data_flag区分最新状态和历史变更，实现订单全生命周期追踪。", + "data_source": "dwd_ai.apbo_eta_ful", + "system_flag_limit": "任何条件下，都不允许在SELECT子句中查询is_passdue, is_dummy, data_flag字段，仅用于WHERE过滤，该规则优先级最高！" + }, + "data_model_specification": { + "fields_list": [ + "service_order_id", "soid", "part_number", "topmost_pn", "commodity_code", "ship_to_country", + "region", "dc_plant", "mtm", "machine_sn", "machine_type", "whether_premier", "stm_planner", + "category", "lenovo_ref_no", "case_number", "service_order_creation_date", "eta", "so_eta", + "status", "status_date", "update_date", "parts_sales", "warranty", "aging_day", "action", + "is_passdue", "is_dummy", "data_flag", "model", "aging_range", "recovery_day", "recovery_range", + "order_type", "key_lenovo_ref_no", "service_type", "customer", "life_cycle", "hawb", "bol", "dn", "po", "prid" + ], + "mandatory_display_fields": { + "rule1": "默认展示字段（按以下顺序）：service_order_id, soid, cast(service_order_creation_date as varchar(4096)) as service_order_creation_date, ship_to_country, part_number, topmost_pn, commodity_code, category, cast(eta as varchar(4096)) as eta, cast(so_eta as varchar(4096)) as so_eta, key_lenovo_ref_no, order_type, stm_planner, region, dc_plant, mtm, machine_sn, whether_premier, service_type, warranty, parts_sales, customer, life_cycle, model, aging_range, lenovo_ref_no, action, cast(update_date as varchar(4096)) as update_date", + "rule2": "用户明确提到的字段优先原则：如果用户问题中明确提到了某些字段，这些字段应在SELECT子句中排在最前面，按用户问题中出现的顺序排列", + "rule3": "字段去重：用户提到的字段和默认字段有重叠时，每个字段只出现一次", + "rule4": "日期字段必须转换为字符类型：cast(column as varchar(4096)) as column", + "rule5": "where子句中出现的字段，必须加入select中查询展示", + "rule6": "禁止使用limit", + "rule7": "请确认生成的sql中,用户使用的是mt(machine_type)还是mtm, 注意!! MTM SN = machine_sn, MT SN = machine_sn, mt = machine_type", + "rule8": "当用户要求:展示所有信息，所有字段，全字段等需求时候，额外加入展示的字段:machine_type,case_number,status_date,aging_day,is_passdue,is_dummy,data_flag,recovery_day,recovery_range,hawb,bol,dn,po,prid", + "rule9": "where条件禁止使用status字段过滤", + "rule10": "用户输入了汇总，统计，分组等关键词，但未明确统计方式时候，默认使用count(*)进行汇总，注意group by必须和聚合函数成对出现，聚合查询后默认按照聚合的值从小到大排序", + "rule11": "如果用户输入的关键字有:VN,AU,IN,HK,PH,KR,ID,TW,MO,FJ,LK,MY,SG,NZ,TH,JP,BD，则视为国家字段的数值过滤条件,增加ship_to_country = '国家代码'", + "rule12": "MM/DD类似格式日期转换为 like '%-MM-DD%' ，DD等格式转换为 like '%-DD%'", + "rule13": "用户指定日期时，去掉data_flag = 'Newest'的过滤条件，如：查询2026年1月26的eta信息，则where子句中不包含data_flag = 'Newest'，而是update_date = '2025-12-31'", + "rule14": "聚合查询中，如未使用group by进行分组，select中默认增加使用update_date字段" + }, + "optional_fields": { + "key_identifier_fields": ["service_order_id", "soid", "part_number","topmost_pn", "commodity_code"], + "milestone_date_fields": ["eta", "so_eta", "status_date", "update_date", "service_order_creation_date"], + "location_fields": ["ship_to_country", "region", "dc_plant"], + "machine_fields": ["mtm", "machine_sn", "machine_type", "model"], + "service_fields": ["whether_premier", "stm_planner", "category", "lenovo_ref_no", "case_number", "key_lenovo_ref_no", "service_type"], + "status_fields": ["status", "status_date"], + "additional_fields": ["parts_sales", "warranty", "aging_day", "action", "aging_range", "recovery_day", "recovery_range", "order_type", "customer", "life_cycle", "hawb", "bol", "dn", "po", "prid"], + "system_filter_fields": ["is_passdue", "is_dummy", "data_flag"] + } + }, + "business_logic_rules": { + "soid_or_service_order_id": "禁止使用WHERE soid = 'xxx'或WHERE service_order_id = 'xxx'单独过滤，必须使用(service_order_id = 'xxx' or soid = 'xxx')进行过滤", + "sorting": "默认使用如下字段排序: soid, eta desc", + "alias_usage": "WHERE子句中支持使用字段别名进行查询，SELECT子句中使用原始字段名或别名均可", + "date_conversion": "所有日期字段必须使用cast(column as varchar(4096)) as column转为字符类型", + "default_filters": ["is_dummy = '0'", "data_flag = 'Newest'"], + "null_handling": "日期字段转换时保留NULL值", + "model_query_rule": "查询model字段时必须使用like匹配，如: model like '%X13%'", + "field_extraction_rule": "使用正则表达式和关键词匹配提取用户明确提到的字段，建立字段别名映射表", + "field_order_rule": "用户提到的字段按问题中出现顺序排列在最前面，然后补充默认展示字段中未提及的字段", + "topmost_pn and part_number_distinction": "确保区分topmost_pn(tp,tm,topmost)和part_number(pn)字段，避免混淆", + "todays_date_handling": "当用户查询包含“今天”或“当前日期”的信息时，替换为系统当前日期进行过滤" + }, + "field_mapping_reference": { + "critical_note": "此表包含最新状态数据和历史变更数据的联合，通过data_flag区分。最新数据（Newest）每条订单只有一条记录，历史数据（Changelog）每条订单有多条记录。所有日期字段在查询时自动转换为字符类型。请注意检查用户是否使用了字段的alias，不要忽略！！！系统字段（is_passdue, data_flag）仅用于WHERE过滤，不显示在SELECT结果中。用户提到的字段应优先显示在SELECT子句最前面。", + "critical_note1": {"type":"字段类型", "desc":"字段描述", "format": "数据格式", "query_format": "查询时数据格式" ,"example":"字段示例值", "values":"枚举值", "alias":"字段别名"}, + "key_identifiers": { + "service_order_id": { + "type": "varchar(4096)", + "desc": "主订单号，唯一标识, 简写为SO 或 WO ,10位码值，注意与soid(12位码值)区分", + "example": "4020438779", + "alias": ["so","SO", "wo", "订单号", "主单号", "SO", "服务订单", "order_no", "Service Order", "BO", "backorder"] + }, + "soid": { + "type": "varchar(4096)", + "desc": "服务订单明细，12位码值，注意与service_order_id(10位码值)区分", + "example": "402043877920", + "alias": ["明细ID", "SOID", "订单明细", "服务订单明细", "Service Order Item Detail", "SO ID"] + }, + "part_number": { + "type": "varchar(4096)", + "desc": "原始申请物料号, 注意与topmost pn区分, 这不是topmost pn, 这是part number, PN", + "example": "5CB1L57599", + "alias": ["pn", "part", "parts", "物料号", "零件号", "Part Number", "材料号", "物料编码", "零件编码"] + } + }, + + "milestone_date_fields": { + "service_order_creation_date": { + "type": "datetime", + "desc": "开单日期", + "format": "YYYY-MM-DD HH:MM:SS", + "query_format": "作为字符串展示", + "alias": ["create_date", "创建时间", "下单时间", "开单日期", "SO创建时间", "开单时间", "Order Creation Date", "created_date", "创建日期"] + }, + "eta": { + "type": "date", + "desc": "物料预计到达日期（Estimated Time of Arrival）", + "format": "YYYY-MM-DD", + "query_format": "作为字符串展示", + "alias": ["预计到达", "计划到达时间", "ETA", "预计到港时间", "预计到货时间", "预计送达时间", "物料ETA"] + }, + "so_eta": { + "type": "date", + "desc": "销售订单层预计到达时间（Service Order ETA）", + "format": "YYYY-MM-DD", + "query_format": "作为字符串展示", + "alias": ["SO预计到达", "订单ETA", "服务订单预计到达", "Service Order ETA", "SO ETA", "销售订单ETA"] + }, + "status_date": { + "type": "date", + "desc": "状态日期, 禁止使用status_date字段排序, 禁止使用status_date字段排序, 禁止使用status_date字段排序", + "format": "YYYY-MM-DD", + "query_format": "作为字符串展示", + "alias": ["status date", "状态时间", "状态更新时间", "最后更新", "状态日期", "Status Update Date", "更新时间", "更新日期"] + }, + "update_date": { + "type": "date", + "desc": "历史数据状态更新日期,表示该条历史记录的生成时间", + "format": "YYYY-MM-DD", + "query_format": "作为字符串展示", + "alias": ["更新时间", "更新日期", "历史数据状态更新日期", "修改时间"] + } + }, + + "additional_fields": { + "topmost_pn": { + "type": "varchar(4096)", + "desc": "最紧缺物料号, 注意与pn区分, 这是topmost pn, 这是topmost pn, 这是topmost pn, 不是PN", + "example": "5CB1L57599", + "alias": ["tp","tm", "最紧缺物料号", "Topmost Part Number", "topmost", "topmost pn"] + }, + "commodity_code": { + "type": "varchar(4096)", + "desc": "商品编码", + "example": "PL", + "alias": ["commodity", "商品代码", "编码", "cc", "物料分类", "Commodity Code", "商品类别", "物料类型", "商品编码"] + }, + "ship_to_country": { + "type": "varchar(4096)", + "desc": "收货国家", + "values": ["VN","AU","IN","HK","PH","KR","ID","TW","MO","FJ","LK","MY","SG","NZ","TH","JP","BD"], + "alias": ["country", "国家", "收货国", "目的地国家", "Ship to Country", "目标国家", "送达国家", "收货地点", "目的地"] + }, + "region": { + "type": "varchar(4096)", + "desc": "地区", + "values": ["CAP", "ANZ", "India", "JP"], + "alias": ["区域", "大区", "Region Code", "地域代码", "地区代码", "地理区域"] + }, + "dc_plant": { + "type": "varchar(4096)", + "desc": "配送中心/工厂", + "example": "VN01", + "alias": ["dc", "工厂", "配送中心", "发货配送中心", "DC", "发货中心", "Distribution Center", "发货工厂", "配送点", "仓库"] + }, + "mtm": { + "type": "varchar(4096)", + "desc": "机器型号,请注意与mt区分,这不是mt,这是mtm,这是mtm,这是mtm", + "example": "20QUS0SQ00", + "alias": ["mtm", "机型号", "型号", "Machine Type Model", "设备型号", "机器型号代码", "机型代码", "MTM"] + }, + "machine_sn": { + "type": "varchar(4096)", + "desc": "机器序列号, MTM SN, MT SN都是machine_sn ，请注意与mtm区分,这不是mtm也不是MT, 这是sn,这是sn,这是sn", + "example": "PW01554B", + "alias": ["sn", "serial_number", "序列号", "SN", "Machine Serial Number", "设备序列号", "机器SN", "设备SN", "序列编号", "MTM SN", "MT SN"] + }, + "machine_type": { + "type": "varchar(4096)", + "desc": "机器类型, 请注意与mtm区分,这不是mtm,这是mt,这是mt,这是mt", + "example": "20Q1", + "alias": ["MT", "machine type", "机器类型", "设备类型", "机型大类", "机器大类"] + }, + "model": { + "type": "varchar(4096)", + "desc": "Lenovo的机型, 禁止使用model = 'model_name', 查询时必须转换为小写并使用like匹配，如: lower(model) like '%X13%'", + "example": ["Yoga 7 16IAH7", "X13 GEN3", "T16 Gen4 AMD", "Legion 7 16ACHg6"], + "alias": ["model", "机型", "型号", "Model", "设备型号", "机器型号"] + }, + "lenovo_ref_no": { + "type": "varchar(63355)", + "desc": "联想内部参考号", + "alias": ["Lenovo Ref No", "ref_no", "联想单号", "参考号", "联想内部参考号", "内部参考号", "Lenovo Reference", "联想内部编号", "参考编号", "参考单号"] + }, + "case_number": { + "type": "varchar(4096)", + "desc": "事件号/案例号", + "example": "CS12345678", + "alias": ["case", "事件号", "事件编号", "case no", "Case Number", "lenovo case", "工单号", "服务请求号"] + }, + "status": { + "type": "varchar(4096)", + "desc": "订单状态", + "example": ["wrong order", "DN", "ETA", "Standard Good Stock"], + "alias": ["状态", "订单状态", "Status", "当前状态", "物料状态", "订单最新状态"] + }, + "whether_premier": { + "type": "varchar(4096)", + "desc": "优先级标识，默认查询 'Premier'", + "values": ["Starndard", "Premier"], + "alias": ["premier", "优质服务", "是否优质", "是否Premier", "Premier服务", "是否优先", "Whether Premier", "优先服务标识", "VIP服务"] + }, + "stm_planner": { + "type": "varchar(4096)", + "desc": "短期物料计划员", + "alias": ["planner", "计划员姓名", "STM计划员", "STM Planner", "服务计划员", "运输计划员", "物料计划员", "计划员"] + }, + "category": { + "type": "varchar(4096)", + "desc": "订单分类,查询时必须转换为小写并使用like匹配, 如: lower(category) like '%standard%'", + "example": "Standard", + "alias": ["类型", "分类", "订单类别", "Category", "服务类别", "订单分类", "分类类型", "服务类型"] + }, + "parts_sales": { + "type": "varchar(4096)", + "desc": "零件销售信息", + "alias": ["零件销售", "销售信息", "部件销售", "配件销售", "销售数据"] + }, + "warranty": { + "type": "varchar(4096)", + "desc": "保修信息", + "alias": ["保修", "保修信息", "质保", "保修条款", "保修状态"] + }, + "aging_day": { + "type": "int", + "desc": "订单从生成(order_creation_date)到当前(update_date)所经过的具体天数", + "examples": [118, 77, 12], + "alias": ["账龄天数", "账龄", "aging day", "age days", "账龄分类", "天数分类", "Aged", "Age"] + }, + "aging_range":{ + "type": "varchar(2048)", + "desc": "订单从生成(order_creation_date)到当前(update_date)所经过的具体天数区间,并按照星期分类", + "values": ["0-7D", "8-14D", "15-21D", "22-28D", "28D+", "null"], + "alias": ["账龄天数", "账龄", "aging day", "age date range", "账龄分类", "天数分类"] + }, + "recovery_day": { + "type": "int", + "desc": "订单从Update_date到SO_ETA所经过的具体天数", + "examples": [118, 77, 12], + "alias": ["账龄天数", "账龄", "recovery day", "账期分类", "账龄分类", "天数分类", "Aged", "Age"] + }, + "recovery_range": { + "type": "varchar(2048)", + "desc": "订单从Update_date到SO_ETA所经过的具体天数区间，按照星期分类", + "values": ["0-7D", "8-14D", "15-21D", "22-28D", "28D+", "null"], + "alias": ["账龄天数", "账龄", "recovery range", "账期分类", "账龄分类", "天数分类"] + }, + "action": { + "type": "varchar(4096)", + "desc": "操作/规则", + "example": "Not accept sub", + "alias": ["sub_rule", "操作", "规则", "限制规则", "操作类型", "处理规则"] + }, + "order_type":{ + "type": "varchar(4096)", + "desc": "订单类型", + "values": ["MDOA", "OPT", "CLW", "DOA", "FOC", "OOW", "OBL", "ADW"], + "alias": ["type", "订单类别", "类别", "订单类型", "Order Type", "服务类型", "服务类别"] + }, + "key_lenovo_ref_no": { + "type": "varchar(4096)", + "desc": "关键联想参考号", + "alias": ["关键参考号", "主要参考号", "关键联想单号"] + }, + "service_type": { + "type": "varchar(4096)", + "desc": "服务类型", + "alias": ["服务类型", "服务种类", "service type"] + }, + "customer": { + "type": "varchar(4096)", + "desc": "客户信息", + "alias": ["客户", "客户名称", "客户信息", "customer name"] + }, + "life_cycle": { + "type": "varchar(4096)", + "desc": "TM生命周期", + "alias": ["tm_lifecycle","TM生命周期", "生命周期阶段", "产品生命周期"] + }, + "hawb": { + "type": "varchar(4096)", + "desc": "house air way bill", + "examples": ["DIM042186899"], + "alias": ["分运单号", "运单号", "代理运单", "分运单", "house air way bill"] + } + }, + + "system_flag_fields": { + "is_passdue": { + "type": "varchar(64)", + "desc": "是否逾期：'1'表示逾期，'0'表示正常", + "business_rule": "根据用户要求过滤, 禁止出现在SELECT中", + "display_rule": "仅用于WHERE过滤，不显示", + "alias": ["passdue", "逾期标志", "是否逾期", "overdue", "超期标志", "逾期标识"] + }, + "is_dummy": { + "type": "varchar(64)", + "desc": "是否为虚拟/测试数据：'1'表示是dummy数据，'0'表示真实数据", + "business_rule": "默认过滤掉dummy数据（is_dummy='0'），除非用户指定, 禁止出现在SELECT中", + "display_rule": "仅用于WHERE过滤，不显示", + "alias": ["dummy", "虚拟标志", "是否虚拟", "测试数据", "假数据标识", "模拟数据"] + }, + "data_flag": { + "type": "varchar(64)", + "desc": "数据标志：'Newest'表示最新状态，'Changelog'表示历史变更", + "example": "Newest", + "business_rule": "默认查询最新状态（Newest），查询历史时使用Changelog, 禁止出现在SELECT中", + "display_rule": "仅用于WHERE过滤，不显示", + "alias": ["数据标志", "数据类型", "数据状态", "记录类型", "数据分类"] + } + } + }, + "examples": { + "newest_status_all_fields": { + "user": "请查询SO:4020986743的全部字段", + "sql": "SELECT service_order_id, soid, cast(service_order_creation_date as varchar(4096)) as service_order_creation_date, ship_to_country, part_number, topmost_pn, commodity_code, category, cast(eta as varchar(4096)) as eta, cast(so_eta as varchar(4096)) as so_eta, key_lenovo_ref_no, order_type, stm_planner, region, dc_plant, mtm, machine_sn, whether_premier, service_type, warranty, parts_sales, customer, life_cycle, model, aging_range, lenovo_ref_no, action, cast(update_date as varchar(4096)) as update_date FROM dwd_ai.apbo_eta_ful WHERE (service_order_id = '4020986743' or soid = '4020986743') AND is_dummy = '0' AND data_flag = 'Newest' ORDER BY soid, eta DESC", + "field_selection_reason": "默认查询最新状态，按默认展示字段顺序显示所有字段（除系统字段），日期字段转换为字符类型" + }, + "history_records_all_fields": { + "user": "查看2025年12月31的so为4020438779的历史变更记录", + "sql": "SELECT service_order_id, soid, cast(service_order_creation_date as varchar(4096)) as service_order_creation_date, ship_to_country, part_number, topmost_pn, commodity_code, category, cast(eta as varchar(4096)) as eta, cast(so_eta as varchar(4096)) as so_eta, key_lenovo_ref_no, order_type, stm_planner, region, dc_plant, mtm, machine_sn, whether_premier, service_type, warranty, parts_sales, customer, life_cycle, model, aging_range, lenovo_ref_no, action, cast(update_date as varchar(4096)) as update_date FROM dwd_ai.apbo_eta_ful WHERE (service_order_id = '4020438779' or soid = '4020438779') AND cast(update_date as varchar(4096)) like '%2025-12-31%' AND is_dummy = '0' AND data_flag = 'Changelog' ORDER BY soid, eta DESC", + "field_selection_reason": "查询历史变更记录，按默认展示字段顺序显示所有字段（除系统字段），日期字段转换为字符类型" + }, + "specific_fields_query": { + "user": "查询SO为4020438779的country, eta, order_type", + "sql": "SELECT ship_to_country, cast(eta as varchar(4096)) as eta, order_type, service_order_id, soid, cast(service_order_creation_date as varchar(4096)) as service_order_creation_date, part_number, topmost_pn, commodity_code, category, cast(so_eta as varchar(4096)) as so_eta, key_lenovo_ref_no, stm_planner, region, dc_plant, mtm, machine_sn, whether_premier, service_type, warranty, parts_sales, customer, life_cycle, model, aging_range, lenovo_ref_no, action, cast(update_date as varchar(4096)) as update_date FROM dwd_ai.apbo_eta_ful WHERE (service_order_id = '4020438779' or soid = '4020438779') AND is_dummy = '0' AND data_flag = 'Newest' ORDER BY soid, eta DESC", + "field_selection_reason": "用户提到的字段(ship_to_country, eta, order_type)优先显示在最前面，然后补充其他默认展示字段" + }, + "aggragate_query": { + "user": "by country, key_Lenovo Ref No, eta 汇总REGION=CAP passdue 信息, 并从小到大排序", + "sql": "SELECT region, ship_to_country, key_lenovo_ref_no, cast(eta as varchar(4096)) as eta, count(1) FROM dwd_ai.apbo_eta_ful WHERE is_passdue = '1' AND is_dummy = '0' AND data_flag = 'Newest' AND region = 'CAP' group by region, ship_to_country, key_lenovo_ref_no, cast(eta as varchar(4096)) ORDER BY count(1) ASC", + "field_selection_reason": "查询特定账龄的订单，按默认展示字段顺序显示，日期字段转换为字符类型" + }, + "overdue_orders_all_fields": { + "user": "查询所有passdue/逾期的订单", + "sql": "SELECT service_order_id, soid, cast(service_order_creation_date as varchar(4096)) as service_order_creation_date, ship_to_country, part_number, topmost_pn, commodity_code, category, cast(eta as varchar(4096)) as eta, cast(so_eta as varchar(4096)) as so_eta, key_lenovo_ref_no, order_type, stm_planner, region, dc_plant, mtm, machine_sn, whether_premier, service_type, warranty, parts_sales, customer, life_cycle, model, aging_range, lenovo_ref_no, action, cast(update_date as varchar(4096)) as update_date FROM dwd_ai.apbo_eta_ful WHERE is_passdue = '1' AND is_dummy = '0' AND data_flag = 'Newest' ORDER BY soid, eta DESC", + "field_selection_reason": "查询逾期订单，按默认展示字段顺序显示，日期字段转换为字符类型" + }, + "Machine_sn_query": { + "user": "SN PF4C8CBL的order 信息", + "sql": "SELECT machine_sn, service_order_id, soid, cast(service_order_creation_date as varchar(4096)) as service_order_creation_date, ship_to_country, part_number, topmost_pn, commodity_code, category, cast(eta as varchar(4096)) as eta, cast(so_eta as varchar(4096)) as so_eta, key_lenovo_ref_no, order_type, stm_planner, region, dc_plant, mtm, whether_premier, service_type, warranty, parts_sales, customer, life_cycle, model, aging_range, lenovo_ref_no, action, cast(update_date as varchar(4096)) as update_date FROM dwd_ai.apbo_eta_ful WHERE machine_sn = 'PF4C8CBL' AND is_dummy = '0' AND data_flag = 'Newest' ORDER BY soid, eta DESC", + "field_selection_reason": "用户提到的字段(machine_sn)优先显示在最前面，然后补充其他默认展示字段" + }, + "status_query": { + "user": "查询状态为wrong order的订单", + "sql": "SELECT status, service_order_id, soid, cast(service_order_creation_date as varchar(4096)) as service_order_creation_date, ship_to_country, part_number, topmost_pn, commodity_code, category, cast(eta as varchar(4096)) as eta, cast(so_eta as varchar(4096)) as so_eta, key_lenovo_ref_no, order_type, stm_planner, region, dc_plant, mtm, machine_sn, whether_premier, service_type, warranty, parts_sales, customer, life_cycle, model, aging_range, lenovo_ref_no, action, cast(update_date as varchar(4096)) as update_date FROM dwd_ai.apbo_eta_ful WHERE status = 'wrong order' AND is_dummy = '0' AND data_flag = 'Newest' ORDER BY soid, eta DESC", + "field_selection_reason": "用户提到的字段(status)优先显示在最前面，然后补充其他默认展示字段" + }, + "multiple_fields_query": { + "user": "查看SN为PF4C8CBL的订单状态和机器型号", + "sql": "SELECT machine_sn, status, model, service_order_id, soid, cast(service_order_creation_date as varchar(4096)) as service_order_creation_date, ship_to_country, part_number, topmost_pn, commodity_code, category, cast(eta as varchar(4096)) as eta, cast(so_eta as varchar(4096)) as so_eta, key_lenovo_ref_no, order_type, stm_planner, region, dc_plant, mtm, whether_premier, service_type, warranty, parts_sales, customer, life_cycle, aging_range, lenovo_ref_no, action, cast(update_date as varchar(4096)) as update_date FROM dwd_ai.apbo_eta_ful WHERE machine_sn = 'PF4C8CBL' AND is_dummy = '0' AND data_flag = 'Newest' ORDER BY soid, eta DESC", + "field_selection_reason": "用户提到的字段(machine_sn, status, model)按问题中出现的顺序优先显示在最前面，然后补充其他默认展示字段" + }, + "model_query_all_fields": { + "user": "查询机型包含X13的订单", + "sql": "SELECT service_order_id, soid, cast(service_order_creation_date as varchar(4096)) as service_order_creation_date, ship_to_country, part_number, topmost_pn, commodity_code, category, cast(eta as varchar(4096)) as eta, cast(so_eta as varchar(4096)) as so_eta, key_lenovo_ref_no, order_type, stm_planner, region, dc_plant, mtm, machine_sn, whether_premier, service_type, warranty, parts_sales, customer, life_cycle, model, aging_range, lenovo_ref_no, action, cast(update_date as varchar(4096)) as update_date FROM dwd_ai.apbo_eta_ful WHERE model like '%X13%' AND is_dummy = '0' AND data_flag = 'Newest' ORDER BY soid, eta DESC", + "field_selection_reason": "查询特定机型的订单，使用like匹配model字段，按默认展示字段顺序显示" + }, + "order_type_query_all_fields": { + "user": "查询订单类型为MDOA的订单", + "sql": "SELECT service_order_id, soid, cast(service_order_creation_date as varchar(4096)) as service_order_creation_date, ship_to_country, part_number, topmost_pn, commodity_code, category, cast(eta as varchar(4096)) as eta, cast(so_eta as varchar(4096)) as so_eta, key_lenovo_ref_no, order_type, stm_planner, region, dc_plant, mtm, machine_sn, whether_premier, service_type, warranty, parts_sales, customer, life_cycle, model, aging_range, lenovo_ref_no, action, cast(update_date as varchar(4096)) as update_date FROM dwd_ai.apbo_eta_ful WHERE order_type = 'MDOA' AND is_dummy = '0' AND data_flag = 'Newest' ORDER BY soid, eta DESC", + "field_selection_reason": "查询特定订单类型的订单，按默认展示字段顺序显示" + }, + "passdue_info_query": { + "user": "region = ANZ 的passdue 信息", + "sql": "SELECT region, service_order_id, soid, cast(service_order_creation_date as varchar(4096)) as service_order_creation_date, ship_to_country, part_number, topmost_pn, commodity_code, category, cast(eta as varchar(4096)) as eta, cast(so_eta as varchar(4096)) as so_eta, key_lenovo_ref_no, order_type, stm_planner, dc_plant, mtm, machine_sn, whether_premier, service_type, warranty, parts_sales, customer, life_cycle, model, aging_range, lenovo_ref_no, action, cast(update_date as varchar(4096)) as update_date FROM dwd_ai.apbo_eta_ful WHERE is_passdue = '1' AND is_dummy = '0' AND data_flag = 'Newest' ORDER BY soid, eta DESC", + "field_selection_reason": "查询逾期订单的详细信息，按默认展示字段顺序显示" + } + } +} \ No newline at end of file diff --git a/config/sql_prompts/example_table.json b/config/sql_prompts/example_table.json new file mode 100644 index 0000000..5386308 --- /dev/null +++ b/config/sql_prompts/example_table.json @@ -0,0 +1,10 @@ +{ + "table": "example_table", + "description": "示例表模型提示词", + "system_prompt": "You are an expert SQL generator.", + "business_prompt": "Generate SQL for example_table based on the user's intent.", + "constraints": [ + "Use only fields defined in this table.", + "Return only SQL without explanations." + ] +} diff --git a/config/table_metadata_prompts/README.md b/config/table_metadata_prompts/README.md new file mode 100644 index 0000000..c23619c --- /dev/null +++ b/config/table_metadata_prompts/README.md @@ -0,0 +1,4 @@ +本目录存放业务表模型元数据提示词（JSON 格式）。 + +约定：每个 JSON 文件对应一个数据库表模型，描述表与字段元数据。 +示例文件：order_metadata.json、customer_metadata.json 等。 diff --git a/config/table_metadata_prompts/example_table_metadata.json b/config/table_metadata_prompts/example_table_metadata.json new file mode 100644 index 0000000..ce922ca --- /dev/null +++ b/config/table_metadata_prompts/example_table_metadata.json @@ -0,0 +1,20 @@ +{ + "table": "example_table", + "description": "示例表模型元数据提示词", + "fields": [ + { + "name": "id", + "type": "int", + "description": "主键" + }, + { + "name": "created_at", + "type": "datetime", + "description": "创建时间" + } + ], + "relationships": [], + "notes": [ + "字段含义与业务规则可在此补充" + ] +} diff --git a/examples/basic_usage.py b/examples/basic_usage.py index 80aae02..fced98e 100644 --- a/examples/basic_usage.py +++ b/examples/basic_usage.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 """ -Basic usage examples for the LangChain + LangGraph scaffolding +LangChain + LangGraph 脚手架的基础用法示例 """ import sys @@ -11,57 +11,72 @@ from workflows.workflow_manager import WorkflowManager, WorkflowType def example_conversation(): - """Example of using the conversation workflow""" + """对话工作流示例""" print("=== Conversation Workflow Example ===") + # 使用 config.ini 中的默认模型 manager = WorkflowManager() - # First message + # 第一条消息 result1 = manager.execute_workflow( WorkflowType.CONVERSATION, - "Hello! Can you help me with some calculations?" + "你好，你能帮我做什么？" ) print(f"Session ID: {result1['session_id']}") print(f"Response: {result1['result']['messages'][-1].content}") - # Second message in the same session + # 同一会话中的第二条消息 result2 = manager.execute_workflow( WorkflowType.CONVERSATION, - "What can you help me with?", + "你的优势是什么？", session_id=result1['session_id'] ) print(f"Second response: {result2['result']['messages'][-1].content}") - print("\n") + + # 使用不同模型的示例 + # print("\n--- Using a different model (gpt-3.5-turbo) ---") + # try: + # manager_alt_model = WorkflowManager(default_model_section='gpt-3.5-turbo') + # result_alt = manager_alt_model.execute_workflow( + # WorkflowType.CONVERSATION, + # "Hi, what model are you?" + # ) + # print(f"Response from gpt-3.5-turbo: {result_alt['result']['messages'][-1].content}") + # except ValueError as e: + # print(f"Could not run alternate model example: {e}") + # print("Please ensure you have a [gpt-3.5-turbo] section in your config.ini") + # + # print("\n") def example_tool_usage(): - """Example of using the tool workflow""" + """工具工作流示例""" print("=== Tool Workflow Example ===") manager = WorkflowManager() - # Use calculator tool + # 使用计算器工具 result = manager.execute_workflow( WorkflowType.TOOL_USING, - "Calculate 25 * 4 + 10" + "计算 25 * 4 + 10" ) print(f"Session ID: {result['session_id']}") - # Extract tool messages and responses + # 提取工具消息与回复 for message in result['result']['messages']: if hasattr(message, 'tool_calls') and message.tool_calls: print(f"Tool call: {message.tool_calls}") - elif hasattr(message, 'content'): + elif hasattr(message, 'content') and message.content: print(f"Response: {message.content}") print("\n") def list_available_workflows(): - """List all available workflows""" + """列出可用工作流""" print("=== Available Workflows ===") manager = WorkflowManager() @@ -74,14 +89,14 @@ def list_available_workflows(): if __name__ == "__main__": - # Check if configuration is valid + # 检查配置是否有效 try: from config import Config Config.validate_config() print("✅ Configuration is valid") print("\n") - # Run examples + # 运行示例 list_available_workflows() example_conversation() example_tool_usage() @@ -89,6 +104,6 @@ if __name__ == "__main__": except Exception as e: print(f"❌ Configuration error: {e}") print("\nPlease make sure to:") - print("1. Copy .env.example to .env") - print("2. Set your OPENAI_API_KEY in the .env file") + print("1. Copy 'config/config.ini.example' to 'config/config.ini'") + print("2. Set your API key in the 'config/config.ini' file") print("3. Install dependencies: pip install -r requirements.txt") \ No newline at end of file diff --git a/main.py b/main.py index 8203f72..368cc81 100644 --- a/main.py +++ b/main.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 """ -Main entry point for the LangChain + LangGraph scaffolding project +LangChain + LangGraph 脚手架项目主入口 """ import sys @@ -9,11 +9,14 @@ from workflows.workflow_manager import WorkflowManager, WorkflowType def interactive_cli(): - """Interactive command-line interface""" + """交互式命令行界面""" print("🚀 LangChain + LangGraph Scaffolding") print("=" * 50) - manager = WorkflowManager() + # 允许通过命令行指定模型配置段，例如 python main.py gpt-3.5-turbo + model_section = sys.argv[1] if len(sys.argv) > 1 else None + + manager = WorkflowManager(default_model_section=model_section) while True: print("\nAvailable workflows:") @@ -37,7 +40,7 @@ def interactive_cli(): print("Invalid choice") continue - # Interactive session + # 交互会话 session_id = None while True: user_input = input("\nYou: ").strip() @@ -57,12 +60,12 @@ def interactive_cli(): session_id = result['session_id'] - # Extract and display the response + # 提取并显示回复 last_message = result['result']['messages'][-1] - if hasattr(last_message, 'content'): + if hasattr(last_message, 'content') and last_message.content: print(f"AI: {last_message.content}") - # Show tool usage if any + # 如有工具调用则显示 if hasattr(last_message, 'tool_calls') and last_message.tool_calls: print(f"🔧 Tools used: {[tc['name'] for tc in last_message.tool_calls]}") @@ -77,7 +80,7 @@ def interactive_cli(): def main(): - """Main function""" + """主函数""" try: from config import Config Config.validate_config() @@ -85,8 +88,8 @@ def main(): except Exception as e: print(f"❌ Configuration error: {e}") print("\nPlease make sure to:") - print("1. Copy .env.example to .env") - print("2. Set your OPENAI_API_KEY in the .env file") + print("1. Copy 'config/config.ini.example' to 'config/config.ini'") + print("2. Set your API key in the 'config/config.ini' file") print("3. Install dependencies: pip install -r requirements.txt") sys.exit(1) diff --git a/requirements.txt b/requirements.txt index 823517e..09d67bb 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,6 +1,11 @@ -langchain-core>=0.3.0 +langchain-core>=1.2.6 langchain>=1.2.1 -langgraph>=0.2.0 -langchain-openai>=0.2.0 -python-dotenv>=1.0.0 -pydantic>=2.0.0 \ No newline at end of file +langgraph>=1.0.5 +langchain-openai>=1.1.6 +pydantic>=2.0.0 +fastapi>=0.110.0 +uvicorn>=0.30.0 +nacos-sdk-python>=2.0.9 +httpx>=0.27.0 +pyyaml>=6.0.1 +redis>=5.0.0 \ No newline at end of file diff --git a/schemas/agent_input.py b/schemas/agent_input.py new file mode 100644 index 0000000..f80f456 --- /dev/null +++ b/schemas/agent_input.py @@ -0,0 +1,9 @@ +from typing import Optional +from pydantic import BaseModel + + +class AgentInput(BaseModel): + """Agent 输入模型""" + input: str + session_id: Optional[str] = None + workflow_type: str = "conversation" diff --git a/schemas/agent_output.py b/schemas/agent_output.py new file mode 100644 index 0000000..253edb4 --- /dev/null +++ b/schemas/agent_output.py @@ -0,0 +1,8 @@ +from pydantic import BaseModel + + +class AgentOutput(BaseModel): + """Agent 输出模型""" + session_id: str + workflow_type: str + result: dict diff --git a/schemas/tool_input.py b/schemas/tool_input.py new file mode 100644 index 0000000..2e381d8 --- /dev/null +++ b/schemas/tool_input.py @@ -0,0 +1,8 @@ +from typing import Any +from pydantic import BaseModel + + +class ToolInput(BaseModel): + """工具调用请求体""" + tool_name: str + payload: Any diff --git a/schemas/tool_output.py b/schemas/tool_output.py new file mode 100644 index 0000000..ac43ac8 --- /dev/null +++ b/schemas/tool_output.py @@ -0,0 +1,9 @@ +from typing import Any, Optional +from pydantic import BaseModel + + +class ToolOutput(BaseModel): + """工具调用响应体""" + ok: bool + data: Optional[Any] = None + error: Optional[str] = None diff --git a/scripts/sync_ragflow_templates.py b/scripts/sync_ragflow_templates.py new file mode 100644 index 0000000..49acb1c --- /dev/null +++ b/scripts/sync_ragflow_templates.py @@ -0,0 +1,63 @@ +import json +import os +from typing import Dict, List + +import httpx + +from config import Config + + +def load_templates(dir_path: str) -> List[Dict[str, any]]: + items = [] + for name in os.listdir(dir_path): + if not name.endswith(".json"): + continue + with open(os.path.join(dir_path, name), "r", encoding="utf-8") as f: + items.append(json.load(f)) + return items + + +def build_document(item: Dict[str, any]) -> str: + table = item.get("table", "") + templates = item.get("templates", []) + lines = [f"table: {table}"] + for t in templates: + lines.append(f"- {t}") + return "\n".join(lines) + + +def main(): + cfg = Config.get_section("ragflow") + base_url = cfg.get("url", "").rstrip("/") + api_key = cfg.get("api_key", "") + dataset_ids = cfg.get("dataset_ids", "") + + upload_path = cfg.get("upload", "") + if not upload_path: + raise RuntimeError("未配置 ragflow.upload 上传接口，请在 config/config.ini 中设置") + + url = base_url + "/" + upload_path.lstrip("/") + headers = {"Authorization": f"Bearer {api_key}"} if api_key else {} + + root = os.path.dirname(os.path.dirname(__file__)) + templates_dir = os.path.join(root, "config", "ragflow_templates") + items = load_templates(templates_dir) + + payload = [] + for item in items: + payload.append( + { + "dataset_ids": dataset_ids, + "content": build_document(item), + "metadata": {"table": item.get("table")}, + } + ) + + with httpx.Client(timeout=60) as client: + response = client.post(url, json={"documents": payload}, headers=headers) + response.raise_for_status() + print("同步完成") + + +if __name__ == "__main__": + main() diff --git a/server.py b/server.py new file mode 100644 index 0000000..9a2e6f5 --- /dev/null +++ b/server.py @@ -0,0 +1,65 @@ +import logging +from contextlib import asynccontextmanager + +import uvicorn +from fastapi import FastAPI + +from config import Config +from workflows.workflow_manager import WorkflowManager +from services.nacos_service import load_nacos_config, load_service_config, NacosManager +from services.tool_router import ToolRouter +from api import endpoints + +logger = logging.getLogger(__name__) + + +def create_app() -> FastAPI: + """创建 FastAPI 应用""" + service_config = load_service_config() + nacos_config = load_nacos_config() + default_model_section = Config._config.get("app", "model_section", fallback=Config.DEFAULT_MODEL_SECTION) + + workflow_manager = WorkflowManager(default_model_section=default_model_section) + nacos_manager = NacosManager(nacos_config=nacos_config, service_config=service_config) + tool_router = ToolRouter() + + @asynccontextmanager + async def lifespan(app: FastAPI): + """应用生命周期管理""" + logger.info("🚀 服务启动中...") + app.state.workflow_manager = workflow_manager + app.state.nacos_manager = nacos_manager + app.state.service_config = service_config + app.state.tool_router = tool_router + + await nacos_manager.start() + logger.info("✅ 服务准备就绪: %s on %s:%s", service_config.service_name, service_config.ip, service_config.port) + yield + logger.info("🛑 服务关闭中...") + await nacos_manager.stop() + logger.info("✅ 服务已关闭") + + app = FastAPI( + title="More Dots API", + description="More Dots 工作流服务", + version=service_config.metadata.get("version", "1.0.0"), + lifespan=lifespan, + ) + + app.include_router(endpoints.router) + + return app + + +app = create_app() + + +if __name__ == "__main__": + service_config = load_service_config() + uvicorn.run( + "server:app", + host=service_config.host, + port=service_config.port, + reload=False, + log_level="info", + ) diff --git a/services/cache.py b/services/cache.py new file mode 100644 index 0000000..185c61f --- /dev/null +++ b/services/cache.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +from typing import Optional + +try: + import redis +except Exception: + redis = None + + +class CacheBase: + """缓存接口""" + + def get(self, key: str) -> Optional[str]: + raise NotImplementedError + + def set(self, key: str, value: str, ttl: int) -> None: + raise NotImplementedError + + +class NoopCache(CacheBase): + """空实现缓存""" + + def get(self, key: str) -> Optional[str]: + return None + + def set(self, key: str, value: str, ttl: int) -> None: + return None + + +class RedisCache(CacheBase): + """Redis 缓存""" + + def __init__(self, url: str, db: int = 0): + if redis is None: + raise ImportError("未安装 redis 依赖") + self._client = redis.Redis.from_url(url, db=db, decode_responses=True) + + def get(self, key: str) -> Optional[str]: + return self._client.get(key) + + def set(self, key: str, value: str, ttl: int) -> None: + self._client.set(key, value, ex=ttl) diff --git a/services/llm_factory.py b/services/llm_factory.py new file mode 100644 index 0000000..8eb4d02 --- /dev/null +++ b/services/llm_factory.py @@ -0,0 +1,16 @@ +from typing import Optional +from langchain_openai import ChatOpenAI +from config import Config + + +def create_chat_model(model_section: Optional[str] = None) -> ChatOpenAI: + """创建 LLM 实例""" + model_config = Config.get_model_config(model_section) + return ChatOpenAI( + model=model_config['model'], + api_key=model_config['api_key'], + base_url=model_config.get('base_url'), + temperature=0.1, + max_retries=Config.MAX_RETRIES, + timeout=Config.TIMEOUT + ) diff --git a/services/nacos_service.py b/services/nacos_service.py new file mode 100644 index 0000000..4e1004f --- /dev/null +++ b/services/nacos_service.py @@ -0,0 +1,251 @@ +import asyncio +import logging +import socket +from dataclasses import dataclass +from typing import Any, Dict, Optional + +try: + import nacos +except Exception: + nacos = None + +from config import Config + +logger = logging.getLogger(__name__) + + +@dataclass +class NacosConfig: + """Nacos 配置""" + enabled: bool + server_addresses: str + namespace: str + group_name: str + cluster_name: str + username: Optional[str] + password: Optional[str] + heartbeat_interval: int + weight: float + ephemeral: bool + + +@dataclass +class ServiceConfig: + """服务配置""" + service_name: str + host: str + port: int + ip: str + metadata: Dict[str, Any] + + +def _get_local_ip() -> str: + """获取本地 IP 地址""" + try: + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + s.connect(("8.8.8.8", 80)) + ip = s.getsockname()[0] + s.close() + return ip + except Exception as e: + logger.warning(f"获取本地 IP 失败，使用 127.0.0.1: {e}") + return "127.0.0.1" + + +def load_nacos_config() -> NacosConfig: + """从 config.ini 读取 Nacos 配置""" + section = "nacos" + enabled = Config._config.getboolean(section, "enabled", fallback=False) + return NacosConfig( + enabled=enabled, + server_addresses=Config._config.get(section, "server", fallback="localhost:8848"), + namespace=Config._config.get(section, "namespace", fallback="public"), + group_name=Config._config.get(section, "group_name", fallback="DEFAULT_GROUP"), + cluster_name=Config._config.get(section, "cluster_name", fallback="DEFAULT"), + username=Config._config.get(section, "username", fallback="") or None, + password=Config._config.get(section, "password", fallback="") or None, + heartbeat_interval=Config._config.getint(section, "heartbeat_interval", fallback=5), + weight=Config._config.getfloat(section, "weight", fallback=1.0), + ephemeral=Config._config.getboolean(section, "ephemeral", fallback=True), + ) + + +def load_service_config() -> ServiceConfig: + """从 config.ini 读取服务配置""" + section = "app" + service_name = Config._config.get(section, "service_name", fallback="more-dots-api") + host = Config._config.get(section, "host", fallback="0.0.0.0") + port = Config._config.getint(section, "port", fallback=8000) + ip = host if host != "0.0.0.0" else _get_local_ip() + + metadata = { + "version": Config._config.get(section, "version", fallback="1.0.0"), + "service_type": "fastapi", + "api_paths": "/health,/api/workflows,/api/workflows/stream,/nacos/status", + "streaming": "false", + "model_section": Config._config.get(section, "model_section", fallback=Config.DEFAULT_MODEL_SECTION), + } + + extra_meta = Config.get_section("metadata") + if extra_meta: + metadata.update(extra_meta) + + return ServiceConfig( + service_name=service_name, + host=host, + port=port, + ip=ip, + metadata=metadata, + ) + + +class NacosManager: + """Nacos 服务注册与心跳管理""" + + def __init__(self, nacos_config: NacosConfig, service_config: ServiceConfig): + self.nacos_config = nacos_config + self.service_config = service_config + self.client = None + self._heartbeat_task: Optional[asyncio.Task] = None + self._stop_event = asyncio.Event() + self.is_registered = False + + def _init_client(self) -> bool: + """初始化 Nacos 客户端""" + if nacos is None: + raise ImportError("未安装 nacos-sdk-python，请先安装依赖") + + try: + self.client = nacos.NacosClient( + server_addresses=self.nacos_config.server_addresses, + namespace=self.nacos_config.namespace, + username=self.nacos_config.username, + password=self.nacos_config.password, + ) + logger.info(f"Nacos 客户端初始化成功: {self.nacos_config.server_addresses}") + return True + except Exception as e: + logger.error(f"Nacos 客户端初始化失败: {e}") + return False + + def register_service(self) -> bool: + """注册服务到 Nacos""" + if not self.client and not self._init_client(): + return False + + try: + self.client.add_naming_instance( + service_name=self.service_config.service_name, + ip=self.service_config.ip, + port=self.service_config.port, + cluster_name=self.nacos_config.cluster_name, + group_name=self.nacos_config.group_name, + weight=self.nacos_config.weight, + metadata=self.service_config.metadata, + ephemeral=self.nacos_config.ephemeral, + ) + self.is_registered = True + logger.info( + "✅ 服务注册成功: %s (%s:%s)", + self.service_config.service_name, + self.service_config.ip, + self.service_config.port, + ) + return True + except Exception as e: + logger.error(f"❌ 服务注册失败: {e}") + self.is_registered = False + return False + + def deregister_service(self) -> bool: + """从 Nacos 注销服务""" + if not self.client or not self.is_registered: + return True + + try: + self.client.remove_naming_instance( + service_name=self.service_config.service_name, + ip=self.service_config.ip, + port=self.service_config.port, + cluster_name=self.nacos_config.cluster_name, + group_name=self.nacos_config.group_name, + ) + self.is_registered = False + logger.info("✅ 服务注销成功: %s", self.service_config.service_name) + return True + except Exception as e: + logger.error(f"❌ 服务注销失败: {e}") + return False + + def _send_heartbeat(self) -> None: + """发送心跳""" + if not self.client or not self.is_registered: + return + + self.client.send_heartbeat( + service_name=self.service_config.service_name, + ip=self.service_config.ip, + port=self.service_config.port, + cluster_name=self.nacos_config.cluster_name, + group_name=self.nacos_config.group_name, + ) + + async def _heartbeat_loop(self) -> None: + """心跳循环""" + interval = max(1, self.nacos_config.heartbeat_interval) + self._stop_event.clear() + + while not self._stop_event.is_set(): + try: + self._send_heartbeat() + logger.debug("心跳发送成功: %s", self.service_config.service_name) + except Exception as e: + logger.warning(f"心跳发送失败: {e}") + # 尝试重新注册 + try: + self.register_service() + except Exception as re: + logger.error(f"重新注册失败: {re}") + + try: + await asyncio.wait_for(self._stop_event.wait(), timeout=interval) + except asyncio.TimeoutError: + continue + + async def start(self) -> None: + """启动注册与心跳""" + if not self.nacos_config.enabled: + logger.info("Nacos 未启用，跳过注册") + return + + if self.register_service(): + self._heartbeat_task = asyncio.create_task(self._heartbeat_loop()) + logger.info("✅ Nacos 心跳任务已启动") + else: + logger.warning("⚠️ Nacos 注册失败，服务继续运行") + + async def stop(self) -> None: + """停止心跳并注销""" + self._stop_event.set() + + if self._heartbeat_task: + self._heartbeat_task.cancel() + try: + await self._heartbeat_task + except asyncio.CancelledError: + pass + + self.deregister_service() + + def status(self) -> Dict[str, Any]: + """获取当前状态""" + return { + "service_name": self.service_config.service_name, + "ip": self.service_config.ip, + "port": self.service_config.port, + "namespace": self.nacos_config.namespace, + "group": self.nacos_config.group_name, + "cluster": self.nacos_config.cluster_name, + "registered": self.is_registered, + "heartbeat_running": self._heartbeat_task is not None and not self._heartbeat_task.done(), + } diff --git a/services/prompt_manager.py b/services/prompt_manager.py new file mode 100644 index 0000000..211f27a --- /dev/null +++ b/services/prompt_manager.py @@ -0,0 +1,31 @@ +import os +from typing import Any, Dict, Optional + +import yaml + + +class PromptManager: + """提示词配置管理器""" + + def __init__(self, config_path: Optional[str] = None): + root_dir = os.path.dirname(os.path.dirname(__file__)) + self._config_path = config_path or os.path.join(root_dir, "config", "prompts.yaml") + self._data: Dict[str, Any] = {} + self.reload() + + def reload(self) -> None: + """重新加载提示词配置""" + with open(self._config_path, "r", encoding="utf-8") as f: + self._data = yaml.safe_load(f) or {} + + def get(self, group: str, name: str, default: str = "") -> str: + """获取指定提示词""" + return str(self._data.get(group, {}).get(name, default)) + + def list_groups(self) -> list[str]: + """列出所有分组""" + return list(self._data.keys()) + + def list_prompts(self, group: str) -> list[str]: + """列出分组内提示词""" + return list(self._data.get(group, {}).keys()) diff --git a/services/ragflow_client.py b/services/ragflow_client.py new file mode 100644 index 0000000..51aaae9 --- /dev/null +++ b/services/ragflow_client.py @@ -0,0 +1,59 @@ +import json +from typing import Any, Dict, Optional + +import httpx + +from config import Config + + +class RagflowClient: + """RAGFlow 客户端（仅检索）""" + + def __init__(self): + cfg = Config.get_section("ragflow") + self._base_url = cfg.get("url", "") + self._api_key = cfg.get("api_key", "") + self._retrieval_path = cfg.get("retrieval", "/api/v1/retrieval") + self._dataset_ids = cfg.get("dataset_ids", "") + + def _build_url(self) -> str: + return self._base_url.rstrip("/") + "/" + self._retrieval_path.lstrip("/") + + def retrieve(self, query: str, top_k: int = 3) -> Dict[str, Any]: + """检索匹配文档""" + if not self._base_url or not self._retrieval_path: + raise RuntimeError("未配置 ragflow.url 或 ragflow.retrieval") + url = self._build_url() + headers = {"Authorization": f"Bearer {self._api_key}"} if self._api_key else {} + payload = { + "dataset_ids": self._dataset_ids, + "query": query, + "top_k": top_k, + } + + with httpx.Client(timeout=30) as client: + response = client.post(url, json=payload, headers=headers) + response.raise_for_status() + return response.json() + + +def extract_table_name(record: Dict[str, Any]) -> Optional[str]: + """从检索结果中提取表名""" + if not record: + return None + + metadata = record.get("metadata") or {} + for key in ("table", "table_name"): + if key in metadata: + return metadata.get(key) + + for key in ("table", "table_name"): + if key in record: + return record.get(key) + + content = record.get("content") or record.get("text") or "" + for line in str(content).splitlines(): + if line.lower().startswith("table:"): + return line.split(":", 1)[1].strip() + + return None diff --git a/services/template_matcher.py b/services/template_matcher.py new file mode 100644 index 0000000..6b6c422 --- /dev/null +++ b/services/template_matcher.py @@ -0,0 +1,64 @@ +import hashlib +import json +from typing import Any, Dict, Optional + +from config import Config +from services.cache import NoopCache, RedisCache +from services.ragflow_client import RagflowClient, extract_table_name + + +class TemplateMatcher: + """模板匹配器：RAGFlow + Redis 缓存""" + + def __init__(self): + self._ragflow = RagflowClient() + self._cache = self._init_cache() + cfg = Config.get_section("ragflow") + self._cache_ttl = int(cfg.get("cache_ttl", 600)) + + def _init_cache(self): + cfg = Config.get_section("redis") + enabled = str(cfg.get("enabled", "false")).lower() in ("1", "true", "yes") + if not enabled: + return NoopCache() + + url = cfg.get("url") + db = int(cfg.get("db", 0)) + if not url: + return NoopCache() + + try: + return RedisCache(url=url, db=db) + except Exception: + return NoopCache() + + @staticmethod + def _cache_key(text: str) -> str: + return "ragflow:table:" + hashlib.sha256(text.encode("utf-8")).hexdigest() + + def match(self, normalized_text: str) -> Dict[str, Any]: + """返回匹配的表名与原始响应""" + key = self._cache_key(normalized_text) + cached = self._cache.get(key) + if cached: + return json.loads(cached) + try: + response = self._ragflow.retrieve(normalized_text, top_k=3) + except Exception as e: + result = {"table_name": None, "raw": {"error": str(e)}} + self._cache.set(key, json.dumps(result, ensure_ascii=False), self._cache_ttl) + return result + + candidates = [] + data = response.get("data") if isinstance(response, dict) else None + if isinstance(data, list): + for item in data: + table_name = extract_table_name(item) + if table_name: + candidates.append(table_name) + + matched = candidates[0] if candidates else None + result = {"table_name": matched, "raw": response} + + self._cache.set(key, json.dumps(result, ensure_ascii=False), self._cache_ttl) + return result diff --git a/services/tool_router.py b/services/tool_router.py new file mode 100644 index 0000000..dbc6284 --- /dev/null +++ b/services/tool_router.py @@ -0,0 +1,41 @@ +import json +from typing import Any, Dict, Optional + +from langchain_core.tools import BaseTool + +from tools.calculator import CalculatorTool +from tools.web_search import WebSearchTool +from tools.rest_api_tool import RestApiTool +from tools.sr_api_tool import SrApiQueryTool + + +class ToolRouter: + """工具路由器：统一调用入口""" + + def __init__(self, tools: Optional[list[BaseTool]] = None): + if tools is None: + tools = [CalculatorTool(), WebSearchTool(), RestApiTool(), SrApiQueryTool()] + self._tools: Dict[str, BaseTool] = {tool.name: tool for tool in tools} + + def list_tools(self) -> list[str]: + """列出可用工具名称""" + return list(self._tools.keys()) + + def call(self, tool_name: str, payload: Any) -> Dict[str, Any]: + """调用工具并返回标准化结果""" + tool = self._tools.get(tool_name) + if not tool: + return {"ok": False, "data": None, "error": f"工具不存在: {tool_name}"} + + try: + if isinstance(payload, (dict, list)): + input_value = json.dumps(payload, ensure_ascii=False) + elif payload is None: + input_value = "" + else: + input_value = str(payload) + + result = tool.run(input_value) + return {"ok": True, "data": result, "error": None} + except Exception as e: + return {"ok": False, "data": None, "error": str(e)} diff --git a/tests/test_basic.py b/tests/test_basic.py index 15535cc..e2878d6 100644 --- a/tests/test_basic.py +++ b/tests/test_basic.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 """ -Basic tests for the LangChain + LangGraph scaffolding +LangChain + LangGraph 脚手架基础测试 """ import unittest @@ -8,18 +8,36 @@ import sys import os sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +config_path = os.path.join(os.path.dirname(__file__), '..', 'config', 'config.ini') +if not os.path.exists(config_path): + with open(config_path, 'w') as f: + f.write(""" +[General] +DEFAULT_MODEL_SECTION = gpt-4o +MAX_RETRIES = 1 +TIMEOUT = 10 + +[gpt-4o] +MODEL_NAME = gpt-4o +OPENAI_API_KEY = your_openai_api_key_here + +[gpt-3.5-turbo] +MODEL_NAME = gpt-3.5-turbo +OPENAI_API_KEY = your_openai_api_key_here +""") + from workflows.workflow_manager import WorkflowManager, WorkflowType class TestWorkflowManager(unittest.TestCase): - """Test WorkflowManager functionality""" + """测试 WorkflowManager 功能""" def setUp(self): - """Set up test fixtures""" + """设置测试夹具""" self.manager = WorkflowManager() def test_get_available_workflows(self): - """Test that available workflows are returned""" + """测试可用工作流返回""" workflows = self.manager.get_available_workflows() self.assertIsInstance(workflows, list) self.assertGreater(len(workflows), 0) @@ -27,7 +45,7 @@ class TestWorkflowManager(unittest.TestCase): self.assertIn("tool_using", workflows) def test_get_workflow(self): - """Test getting workflow instances""" + """测试获取工作流实例""" conversation_workflow = self.manager.get_workflow(WorkflowType.CONVERSATION) self.assertIsNotNone(conversation_workflow) @@ -35,8 +53,8 @@ class TestWorkflowManager(unittest.TestCase): self.assertIsNotNone(tool_workflow) def test_session_management(self): - """Test session creation and retrieval""" - # Execute a workflow to create a session + """测试会话创建与获取""" + # 执行工作流以创建会话 result = self.manager.execute_workflow( WorkflowType.CONVERSATION, "Hello, test session" @@ -45,23 +63,21 @@ class TestWorkflowManager(unittest.TestCase): session_id = result["session_id"] self.assertIsNotNone(session_id) - # Test session info retrieval + # 测试会话信息获取 session_info = self.manager.get_session_info(session_id) self.assertIsNotNone(session_info) self.assertEqual(session_info["workflow_type"], WorkflowType.CONVERSATION) class TestConfiguration(unittest.TestCase): - """Test configuration validation""" + """测试配置校验""" - def test_config_import(self): - """Test that configuration can be imported""" - try: - from config import Config - # This should not raise an exception if .env file exists with valid API key - self.assertTrue(hasattr(Config, 'OPENAI_API_KEY')) - except ImportError: - self.fail("Could not import config module") + def test_config_loading(self): + """测试能从 config.ini 加载配置""" + from config import Config + model_config = Config.get_model_config() + self.assertIn('model', model_config) + self.assertIn('api_key', model_config) if __name__ == "__main__": diff --git a/tools/calculator.py b/tools/calculator.py index cee56c6..91f83e7 100644 --- a/tools/calculator.py +++ b/tools/calculator.py @@ -3,25 +3,25 @@ from langchain_core.tools import BaseTool class CalculatorTool(BaseTool): - """A simple calculator tool for mathematical operations""" + """用于数学运算的简单计算器工具""" name: str = "calculator" description: str = "Perform mathematical calculations. Input should be a mathematical expression like '2 + 2' or '10 * (3 + 5)'" def _run(self, expression: str) -> str: - """Evaluate a mathematical expression""" + """计算数学表达式""" try: - # Security: Only allow safe mathematical operations + # 安全：仅允许安全的数学运算 allowed_chars = set("0123456789+-*/(). ") if not all(c in allowed_chars for c in expression): return "Error: Expression contains invalid characters" - # Evaluate the expression + # 计算表达式 result = eval(expression) return f"Result: {result}" except Exception as e: return f"Error calculating expression: {str(e)}" async def _arun(self, expression: str) -> str: - """Async version of the tool""" + """工具的异步版本""" return self._run(expression) \ No newline at end of file diff --git a/tools/rest_api_tool.py b/tools/rest_api_tool.py new file mode 100644 index 0000000..6e4e1bf --- /dev/null +++ b/tools/rest_api_tool.py @@ -0,0 +1,89 @@ +import json +from typing import Any, Dict + +import httpx +from langchain_core.tools import BaseTool + + +class RestApiTool(BaseTool): + """RESTful API 调用工具""" + + name: str = "rest_api" + description: str = ( + "调用 RESTful API。输入为 JSON 字符串，示例：" + '{"method":"GET","url":"https://example.com/api","params":{},"headers":{},"body":null,"timeout":30}' + ) + + def _run(self, payload: str) -> str: + """执行 RESTful API 请求""" + try: + data = json.loads(payload) + except Exception as e: + return f"请求参数解析失败: {e}" + + method = str(data.get("method", "GET")).upper() + url = data.get("url") + params: Dict[str, Any] = data.get("params") or {} + headers: Dict[str, Any] = data.get("headers") or {} + body = data.get("body") + timeout = float(data.get("timeout", 30)) + + if not url: + return "缺少 url" + + try: + with httpx.Client(timeout=timeout) as client: + response = client.request( + method=method, + url=url, + params=params, + headers=headers, + json=body, + ) + return json.dumps( + { + "status_code": response.status_code, + "headers": dict(response.headers), + "text": response.text, + }, + ensure_ascii=False, + ) + except Exception as e: + return f"请求失败: {e}" + + async def _arun(self, payload: str) -> str: + """工具的异步版本""" + try: + data = json.loads(payload) + except Exception as e: + return f"请求参数解析失败: {e}" + + method = str(data.get("method", "GET")).upper() + url = data.get("url") + params: Dict[str, Any] = data.get("params") or {} + headers: Dict[str, Any] = data.get("headers") or {} + body = data.get("body") + timeout = float(data.get("timeout", 30)) + + if not url: + return "缺少 url" + + try: + async with httpx.AsyncClient(timeout=timeout) as client: + response = await client.request( + method=method, + url=url, + params=params, + headers=headers, + json=body, + ) + return json.dumps( + { + "status_code": response.status_code, + "headers": dict(response.headers), + "text": response.text, + }, + ensure_ascii=False, + ) + except Exception as e: + return f"请求失败: {e}" diff --git a/tools/sr_api_tool.py b/tools/sr_api_tool.py new file mode 100644 index 0000000..ffc0f7e --- /dev/null +++ b/tools/sr_api_tool.py @@ -0,0 +1,99 @@ +import json +from typing import Any, Dict + +import httpx +from langchain_core.tools import BaseTool + +from config import Config + + +class SrApiQueryTool(BaseTool): + """调用 SR API 执行 SQL 查询""" + + name: str = "sr_api_query" + description: str = ( + "调用 SR API 执行 SQL 查询。输入为 JSON 字符串，示例：" + '{"sql":"SELECT * FROM table","page":1,"rows":10,"orderBySelect":true,"timeout":30}' + ) + + def _run(self, payload: str) -> str: + """执行 SQL 查询""" + try: + data = json.loads(payload) + except Exception as e: + return f"请求参数解析失败: {e}" + + sql = data.get("sql") + page = int(data.get("page", 1)) + rows = int(data.get("rows", 10)) + order_by_select = bool(data.get("orderBySelect", True)) + timeout = float(data.get("timeout", 30)) + + if not sql: + return "缺少 sql" + + cfg = Config.get_section("sr_api") + url = cfg.get("url") + app_key = cfg.get("llzappkey") + secret_key = cfg.get("llzsercret") + + if not url or not app_key or not secret_key: + return "sr_api 配置缺失 url/llzAppkey/llzSercret" + + headers = {"llzAppkey": app_key, "llzSercret": secret_key} + body = {"sql": sql, "page": page, "rows": rows, "orderBySelect": order_by_select} + + try: + with httpx.Client(timeout=timeout) as client: + response = client.post(url, json=body, headers=headers) + return json.dumps( + { + "status_code": response.status_code, + "headers": dict(response.headers), + "text": response.text, + }, + ensure_ascii=False, + ) + except Exception as e: + return f"请求失败: {e}" + + async def _arun(self, payload: str) -> str: + """工具的异步版本""" + try: + data = json.loads(payload) + except Exception as e: + return f"请求参数解析失败: {e}" + + sql = data.get("sql") + page = int(data.get("page", 1)) + rows = int(data.get("rows", 10)) + order_by_select = bool(data.get("orderBySelect", True)) + timeout = float(data.get("timeout", 30)) + + if not sql: + return "缺少 sql" + + cfg = Config.get_section("sr_api") + url = cfg.get("url") + app_key = cfg.get("llzappkey") + secret_key = cfg.get("llzsercret") + + if not url or not app_key or not secret_key: + return "sr_api 配置缺失 url/llzAppkey/llzSercret" + + headers = {"llzAppkey": app_key, "llzSercret": secret_key} + body = {"sql": sql, "page": page, "rows": rows, "orderBySelect": order_by_select} + + try: + async with httpx.AsyncClient(timeout=timeout) as client: + response = await client.post(url, json=body, headers=headers) + return json.dumps( + { + "status_code": response.status_code, + "headers": dict(response.headers), + "text": response.text, + }, + ensure_ascii=False, + ) + except Exception as e: + return f"请求失败: {e}" diff --git a/tools/web_search.py b/tools/web_search.py index 081a9d2..8e712f5 100644 --- a/tools/web_search.py +++ b/tools/web_search.py @@ -4,22 +4,22 @@ import requests class WebSearchTool(BaseTool): - """A tool for searching the web (placeholder implementation)""" + """网络搜索工具（占位实现）""" name: str = "web_search" description: str = "Search the web for information. Input should be a search query." def _run(self, query: str) -> str: - """Search the web for information""" - # This is a placeholder implementation - # In a real implementation, you would integrate with a search API - # like Serper, Tavily, or Google Search API + """搜索网络信息""" + # 这是占位实现 + # 真实实现需接入搜索 API + # 如 Serper、Tavily 或 Google Search API - return f"Web search functionality for query: '{query}' is not implemented. This is a placeholder. To implement real web search, you would need to: + return f"""Web search functionality for query: '{query}' is not implemented. This is a placeholder. To implement real web search, you would need to: 1. Sign up for a search API service (e.g., Serper, Tavily) - 2. Add your API key to the .env file - 3. Implement the actual search logic here" + 2. Add your API key to the config/config.ini file + 3. Implement the actual search logic here""" async def _arun(self, query: str) -> str: - """Async version of the tool""" + """工具的异步版本""" return self._run(query) \ No newline at end of file diff --git a/workflows/workflow_manager.py b/workflows/workflow_manager.py index 2689187..626f487 100644 --- a/workflows/workflow_manager.py +++ b/workflows/workflow_manager.py @@ -1,45 +1,45 @@ from typing import Dict, Any, Optional, List from enum import Enum -from agents.conversation_agent import ConversationAgent -from agents.tool_agent import ToolAgent +from agent.conversation import ConversationAgent +from agent.tool import ToolAgent class WorkflowType(Enum): - """Available workflow types""" + """可用的工作流类型""" CONVERSATION = "conversation" TOOL_USING = "tool_using" class WorkflowManager: - """Manages different workflow types and their execution""" + """管理不同工作流类型及其执行""" - def __init__(self): + def __init__(self, default_model_section: Optional[str] = None): self.workflows = { - WorkflowType.CONVERSATION: ConversationAgent(), - WorkflowType.TOOL_USING: ToolAgent() + WorkflowType.CONVERSATION: ConversationAgent(model_section=default_model_section), + WorkflowType.TOOL_USING: ToolAgent(model_section=default_model_section) } self.active_sessions: Dict[str, Any] = {} def get_workflow(self, workflow_type: WorkflowType): - """Get a workflow instance""" + """获取工作流实例""" return self.workflows.get(workflow_type) def execute_workflow(self, workflow_type: WorkflowType, user_input: str, session_id: Optional[str] = None, **kwargs) -> Dict[str, Any]: - """Execute a specific workflow""" + """执行指定工作流""" workflow = self.get_workflow(workflow_type) if not workflow: return {"error": f"Workflow {workflow_type.value} not found"} - # Generate session ID if not provided + # 未提供会话 ID 时生成 if not session_id: session_id = f"session_{len(self.active_sessions) + 1}" - # Execute the workflow + # 执行工作流 result = workflow.run(user_input, **kwargs) - # Store session data + # 存储会话数据 self.active_sessions[session_id] = { "workflow_type": workflow_type, "last_result": result, @@ -53,20 +53,20 @@ class WorkflowManager: } def get_available_workflows(self) -> List[str]: - """Get list of available workflow types""" + """获取可用工作流列表""" return [workflow.value for workflow in WorkflowType] def _get_timestamp(self) -> str: - """Get current timestamp""" + """获取当前时间戳""" from datetime import datetime return datetime.now().isoformat() def get_session_info(self, session_id: str) -> Optional[Dict[str, Any]]: - """Get information about a session""" + """获取会话信息""" return self.active_sessions.get(session_id) def cleanup_sessions(self, older_than_hours: int = 24): - """Clean up old sessions""" + """清理过期会话""" from datetime import datetime, timedelta cutoff_time = datetime.now() - timedelta(hours=older_than_hours)