From 27a1b8a7e94d93daadda8430c40401f7eadf87e0 Mon Sep 17 00:00:00 2001 From: cjw <792430652@qq.com> Date: Tue, 17 Feb 2026 02:31:39 +0800 Subject: [PATCH] init --- .env.example | 6 ++ .gitignore | 17 ++++ README.md | 174 +++++++++++++++++++++++++++++++++- agents/base_agent.py | 76 +++++++++++++++ agents/conversation_agent.py | 94 ++++++++++++++++++ agents/tool_agent.py | 86 +++++++++++++++++ config.py | 27 ++++++ examples/basic_usage.py | 94 ++++++++++++++++++ main.py | 95 +++++++++++++++++++ requirements.txt | 6 ++ tests/test_basic.py | 68 +++++++++++++ tools/calculator.py | 27 ++++++ tools/web_search.py | 25 +++++ workflows/workflow_manager.py | 83 ++++++++++++++++ 14 files changed, 877 insertions(+), 1 deletion(-) create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 agents/base_agent.py create mode 100644 agents/conversation_agent.py create mode 100644 agents/tool_agent.py create mode 100644 config.py create mode 100644 examples/basic_usage.py create mode 100644 main.py create mode 100644 requirements.txt create mode 100644 tests/test_basic.py create mode 100644 tools/calculator.py create mode 100644 tools/web_search.py create mode 100644 workflows/workflow_manager.py diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..f6f27ba --- /dev/null +++ b/.env.example @@ -0,0 +1,6 @@ +# 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 new file mode 100644 index 0000000..d278616 --- /dev/null +++ b/.gitignore @@ -0,0 +1,17 @@ +# Python 虚拟环境 +venv/ +.venv/ + +# Python 缓存文件 +__pycache__/ +*.pyc +*.pyo +*.pyd + +# IDE 配置（可选） +.idea/ +.vscode/ + +# 项目临时文件 +.DS_Store +*.log \ No newline at end of file diff --git a/README.md b/README.md index a5338ea..1aff3e0 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,174 @@ -# more_dots +# LangChain + LangGraph Scaffolding + +一个使用 LangChain 和 LangGraph 构建的 AI 应用脚手架项目，提供模块化的代理和工作流管理。 + +## 特性 + +- 🚀 **模块化架构**: 基于代理和工作流的模块化设计 +- 🔧 **工具集成**: 支持自定义工具和函数调用 +- 💬 **多轮对话**: 内置对话状态管理和上下文维护 +- 📊 **工作流管理**: 多种工作流类型，支持会话和工具使用 +- ⚙️ **配置管理**: 统一的环境变量和配置管理 +- 🧪 **测试支持**: 包含基础测试和示例代码 + +## 项目结构 + +``` +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 # 配置文件 +├── requirements.txt # 依赖包列表 +├── .env.example # 环境变量示例 +├── main.py # 主程序入口 +└── README.md # 项目说明 +``` + +## 快速开始 + +### 1. 安装依赖 + +```bash +pip install -r requirements.txt +``` + +### 2. 配置环境变量 + +```bash +# 复制环境变量文件 +cp .env.example .env + +# 编辑 .env 文件，设置你的 OpenAI API 密钥 +OPENAI_API_KEY=your_openai_api_key_here +``` + +### 3. 运行示例 + +```bash +# 运行基础示例 +python examples/basic_usage.py + +# 运行交互式 CLI +python main.py +``` + +## 使用指南 + +### 基础用法 + +```python +from workflows.workflow_manager import WorkflowManager, WorkflowType + +# 创建工作流管理器 +manager = WorkflowManager() + +# 使用对话工作流 +result = manager.execute_workflow( + WorkflowType.CONVERSATION, + "Hello! How can you help me?" +) + +# 使用工具工作流 +result = manager.execute_workflow( + WorkflowType.TOOL_USING, + "Calculate 15 * 3 + 7" +) +``` + +### 自定义工具 + +创建新的工具类： + +```python +from langchain_core.tools import BaseTool + +class CustomTool(BaseTool): + name = "custom_tool" + description = "A custom tool for specific tasks" + + def _run(self, input: str) -> str: + # 实现工具逻辑 + return f"Processed: {input}" +``` + +### 扩展代理 + +创建新的代理类型： + +```python +from agents.base_agent import BaseAgent + +class CustomAgent(BaseAgent): + def _build_graph(self): + # 实现自定义图结构 + pass + + def _custom_node(self, state): + # 自定义节点逻辑 + return state +``` + +## 工作流类型 + +| 工作流类型 | 描述 | 适用场景 | +|-----------|------|----------| +| `conversation` | 多轮对话代理 | 聊天机器人、客服系统 | +| `tool_using` | 工具使用代理 | 任务执行、数据分析 | + +## 开发指南 + +### 添加新功能 + +1. **新工具**: 在 `tools/` 目录下创建新的工具类 +2. **新代理**: 在 `agents/` 目录下继承 `BaseAgent` 类 +3. **新工作流**: 在 `workflows/` 目录下扩展工作流管理器 + +### 测试 + +```bash +# 运行所有测试 +python -m pytest tests/ + +# 运行特定测试 +python -m pytest tests/test_basic.py +``` + +### 调试 + +项目使用标准的 Python 日志系统，可以通过设置环境变量启用调试模式： + +```python +import logging +logging.basicConfig(level=logging.DEBUG) +``` + +## 依赖项 + +主要依赖包： + +- `langchain-core`: LangChain 核心功能 +- `langchain`: LangChain 主包 +- `langgraph`: LangGraph 图工作流 +- `langchain-openai`: OpenAI 集成 +- `python-dotenv`: 环境变量管理 +- `pydantic`: 数据验证 + +## 许可证 + +MIT License + +## 贡献 + +欢迎提交 Issue 和 Pull Request 来改进这个项目！ diff --git a/agents/base_agent.py b/agents/base_agent.py new file mode 100644 index 0000000..9c7fd23 --- /dev/null +++ b/agents/base_agent.py @@ -0,0 +1,76 @@ +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 + + +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 diff --git a/agents/conversation_agent.py b/agents/conversation_agent.py new file mode 100644 index 0000000..36e0cce --- /dev/null +++ b/agents/conversation_agent.py @@ -0,0 +1,94 @@ +from typing import Dict, Any, List +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) + 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"]): + 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: + """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 diff --git a/agents/tool_agent.py b/agents/tool_agent.py new file mode 100644 index 0000000..915ee51 --- /dev/null +++ b/agents/tool_agent.py @@ -0,0 +1,86 @@ +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 + + +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 diff --git a/config.py b/config.py new file mode 100644 index 0000000..0ce7b47 --- /dev/null +++ b/config.py @@ -0,0 +1,27 @@ +import os +from dotenv import load_dotenv + +# Load environment variables +load_dotenv() + +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 diff --git a/examples/basic_usage.py b/examples/basic_usage.py new file mode 100644 index 0000000..80aae02 --- /dev/null +++ b/examples/basic_usage.py @@ -0,0 +1,94 @@ +#!/usr/bin/env python3 +""" +Basic usage examples for the LangChain + LangGraph scaffolding +""" + +import sys +import os +sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from workflows.workflow_manager import WorkflowManager, WorkflowType + + +def example_conversation(): + """Example of using the conversation workflow""" + print("=== Conversation Workflow Example ===") + + 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") + + +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" + ) + + 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'): + print(f"Response: {message.content}") + + print("\n") + + +def list_available_workflows(): + """List all available workflows""" + print("=== Available Workflows ===") + + manager = WorkflowManager() + workflows = manager.get_available_workflows() + + for workflow in workflows: + print(f"- {workflow}") + + print("\n") + + +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() + + 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("3. Install dependencies: pip install -r requirements.txt") \ No newline at end of file diff --git a/main.py b/main.py new file mode 100644 index 0000000..8203f72 --- /dev/null +++ b/main.py @@ -0,0 +1,95 @@ +#!/usr/bin/env python3 +""" +Main entry point for the LangChain + LangGraph scaffolding project +""" + +import sys +import os +from workflows.workflow_manager import WorkflowManager, WorkflowType + + +def interactive_cli(): + """Interactive command-line interface""" + print("🚀 LangChain + LangGraph Scaffolding") + print("=" * 50) + + manager = WorkflowManager() + + while True: + print("\nAvailable workflows:") + for i, workflow in enumerate(manager.get_available_workflows(), 1): + print(f"{i}. {workflow}") + print("0. Exit") + + try: + choice = input("\nSelect workflow (0-2): ").strip() + + if choice == "0": + print("Goodbye!") + break + elif choice == "1": + workflow_type = WorkflowType.CONVERSATION + print("\n💬 Conversation Mode - Type 'quit' to return to menu") + elif choice == "2": + workflow_type = WorkflowType.TOOL_USING + print("\n🔧 Tool Mode - Type 'quit' to return to menu") + else: + print("Invalid choice") + continue + + # Interactive session + session_id = None + while True: + user_input = input("\nYou: ").strip() + + if user_input.lower() in ['quit', 'exit', 'q']: + break + + if not user_input: + continue + + try: + result = manager.execute_workflow( + workflow_type, + user_input, + session_id=session_id + ) + + session_id = result['session_id'] + + # Extract and display the response + last_message = result['result']['messages'][-1] + if hasattr(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]}") + + except Exception as e: + print(f"❌ Error: {e}") + + except KeyboardInterrupt: + print("\n\nGoodbye!") + break + except Exception as e: + print(f"Error: {e}") + + +def main(): + """Main function""" + try: + from config import Config + Config.validate_config() + interactive_cli() + 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("3. Install dependencies: pip install -r requirements.txt") + sys.exit(1) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..cbcd862 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,6 @@ +langchain-core>=0.3.0 +langchain>=0.3.0 +langgraph>=0.2.0 +langchain-openai>=0.2.0 +python-dotenv>=1.0.0 +pydantic>=2.0.0 \ No newline at end of file diff --git a/tests/test_basic.py b/tests/test_basic.py new file mode 100644 index 0000000..15535cc --- /dev/null +++ b/tests/test_basic.py @@ -0,0 +1,68 @@ +#!/usr/bin/env python3 +""" +Basic tests for the LangChain + LangGraph scaffolding +""" + +import unittest +import sys +import os +sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from workflows.workflow_manager import WorkflowManager, WorkflowType + + +class TestWorkflowManager(unittest.TestCase): + """Test WorkflowManager functionality""" + + 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) + self.assertIn("conversation", workflows) + 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) + + tool_workflow = self.manager.get_workflow(WorkflowType.TOOL_USING) + 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" + ) + + 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") + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tools/calculator.py b/tools/calculator.py new file mode 100644 index 0000000..cee56c6 --- /dev/null +++ b/tools/calculator.py @@ -0,0 +1,27 @@ +from typing import Dict, Any +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/web_search.py b/tools/web_search.py new file mode 100644 index 0000000..081a9d2 --- /dev/null +++ b/tools/web_search.py @@ -0,0 +1,25 @@ +from typing import Dict, Any, List +from langchain_core.tools import BaseTool +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 + + 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" + + 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 new file mode 100644 index 0000000..2689187 --- /dev/null +++ b/workflows/workflow_manager.py @@ -0,0 +1,83 @@ +from typing import Dict, Any, Optional, List +from enum import Enum +from agents.conversation_agent import ConversationAgent +from agents.tool_agent 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): + self.workflows = { + WorkflowType.CONVERSATION: ConversationAgent(), + WorkflowType.TOOL_USING: ToolAgent() + } + 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 + 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, + "timestamp": self._get_timestamp() + } + + return { + "session_id": session_id, + "workflow_type": workflow_type.value, + "result": result + } + + 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) + + sessions_to_remove = [] + for session_id, session_data in self.active_sessions.items(): + session_time = datetime.fromisoformat(session_data["timestamp"]) + if session_time < cutoff_time: + sessions_to_remove.append(session_id) + + for session_id in sessions_to_remove: + del self.active_sessions[session_id] + + return len(sessions_to_remove) \ No newline at end of file