feat(agent): reorganize imports and improve code structure

- Standardize import organization with clear sections (standard library, third-party, local)
- Move sys.path modification to top of imports for better clarity
- Fix circular import issues by reordering imports
- Remove duplicate imports and ensure proper import grouping
- Maintain all existing functionality while improving code readability
This commit is contained in:
YueGuobin 2026-03-05 18:06:36 +08:00
parent 08b5b44a50
commit 16559d2ce2
7 changed files with 27 additions and 31 deletions

View File

@ -40,30 +40,33 @@ The agent provides:
"""
# Standard library imports
import logging
import operator
import sys
from pathlib import Path
from typing import Annotated, Literal
# Third-party imports
from langchain.messages import AnyMessage, SystemMessage, ToolMessage
from langchain_core.runnables import RunnableConfig
from langgraph.graph import END, START, StateGraph
from langgraph.managed.is_last_step import RemainingSteps
from typing_extensions import TypedDict
import logging
# Add backend to path for prompt_manager
sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / "backend"))
# Local imports
from gns3server.agent.gns3_copilot.agent.context_manager import (
create_pre_model_hook,
)
from gns3server.agent.gns3_copilot.agent.model_factory import (
create_base_model_with_tools,
create_title_model,
)
from gns3server.agent.gns3_copilot.agent.context_manager import (
create_pre_model_hook,
)
from gns3server.agent.gns3_copilot.gns3_client import GNS3TopologyTool
from gns3server.agent.gns3_copilot.prompts import TITLE_PROMPT, load_system_prompt
import sys
from pathlib import Path
# Add backend to path for prompt_manager
sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / "backend"))
from gns3server.agent.gns3_copilot.tools_v2 import (
ExecuteMultipleDeviceConfigCommands,
ExecuteMultipleDeviceCommands,

View File

@ -37,9 +37,17 @@ Features:
"""
# Standard library imports
import asyncio
import concurrent.futures
import logging
from typing import Optional
import threading
from contextvars import ContextVar
from typing import Optional
from uuid import UUID
# Local imports
from gns3server.agent.gns3_copilot.gns3_client.custom_gns3fy import Gns3Connector
logger = logging.getLogger(__name__)
@ -77,9 +85,6 @@ def get_current_llm_config() -> Optional[dict]:
else:
logger.warning("LLM config not found in context")
return config
from uuid import UUID
from gns3server.agent.gns3_copilot.gns3_client.custom_gns3fy import Gns3Connector
# Fallback default URL
DEFAULT_GNS3_URL = "http://127.0.0.1:3080"
@ -412,9 +417,6 @@ def get_llm_config(user_id, jwt_token: str, app=None) -> Optional[dict]:
model = config['model']
api_key = config['api_key']
"""
import asyncio
import threading
try:
# Convert user_id to UUID if it's a string
if isinstance(user_id, str):
@ -429,7 +431,6 @@ def get_llm_config(user_id, jwt_token: str, app=None) -> Optional[dict]:
loop = asyncio.get_running_loop()
# We're in an async context with a running loop
# This shouldn't happen since this is a sync function
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor() as executor:
future = executor.submit(
asyncio.run,
@ -445,7 +446,6 @@ def get_llm_config(user_id, jwt_token: str, app=None) -> Optional[dict]:
loop = asyncio.get_event_loop()
if loop.is_running():
# Loop is running but not the running loop (edge case)
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor() as executor:
future = executor.submit(
asyncio.run,

View File

@ -33,6 +33,7 @@ and link count.
"""
import logging
from pprint import pprint
from typing import Any
from langchain.tools import BaseTool
@ -134,8 +135,6 @@ class GNS3ProjectInfoTool(BaseTool):
return {"error": f"Failed to retrieve project info: {str(e)}"}
if __name__ == "__main__":
from pprint import pprint
# Test the tool
tool = GNS3ProjectInfoTool()

View File

@ -33,6 +33,7 @@ specific GNS3 project by project ID. Returns nodes, links, and project metadata.
import copy
import logging
from pprint import pprint
from typing import Any
from langchain.tools import BaseTool
@ -144,8 +145,6 @@ class GNS3TopologyTool(BaseTool):
return data
if __name__ == "__main__":
from pprint import pprint
# Test the tool
tool = GNS3TopologyTool()

View File

@ -31,10 +31,9 @@ This module provides a tool to execute configuration commands on multiple device
import json
import logging
import os
import re
from typing import Any
from gns3server.agent.gns3_copilot.gns3_client import get_gns3_server_host
from langchain.tools import BaseTool
from langchain_core.callbacks import CallbackManagerForToolRun
from netmiko.exceptions import ReadTimeout
@ -43,6 +42,7 @@ from nornir.core import Nornir
from nornir.core.task import AggregatedResult, Result, Task
from nornir_netmiko.tasks import netmiko_send_config
from gns3server.agent.gns3_copilot.gns3_client import get_gns3_server_host
from gns3server.agent.gns3_copilot.utils import get_device_ports_from_topology
# config log
@ -366,8 +366,6 @@ class ExecuteMultipleDeviceConfigCommands(BaseTool):
Returns:
True if valid UUID format, False otherwise
"""
import re
uuid_pattern = r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$"
return bool(re.match(uuid_pattern, project_id, re.IGNORECASE))

View File

@ -33,6 +33,8 @@ import uuid
from datetime import datetime
from typing import Dict, Any
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage
def _ensure_string(content: Any) -> str:
"""Ensure content is a string, converting dicts/lists to JSON if needed."""
if isinstance(content, str):
@ -52,8 +54,6 @@ def convert_langchain_to_openai(lc_message) -> Dict[str, Any]:
Returns:
Dictionary in OpenAI-compatible format
"""
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, ToolMessage
# Generate message ID
msg_id = getattr(lc_message, 'id', None)
if msg_id is None:
@ -140,8 +140,6 @@ def convert_openai_to_langchain(msg: Dict[str, Any]):
Returns:
LangChain message
"""
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage, SystemMessage
role = msg.get("role", "user")
content = msg.get("content", "")

View File

@ -66,6 +66,7 @@ Author: Guobin Yue
import ast
import json
import logging
from datetime import datetime
from typing import Any
logger = logging.getLogger(__name__)
@ -309,8 +310,6 @@ def normalize_tool_response(
>>> normalize_tool_response([{"device_name": "R1", "status": "success"}])
{'success': True, 'total': 1, 'successful': 1, 'failed': 0, 'data': [...], 'metadata': {}}
"""
from datetime import datetime
metadata = {
"tool_name": tool_name,
"normalized_at": datetime.utcnow().isoformat()