langchain-agent/
├── langchain_agent/              # 核心代码包
│   ├── __init__.py              # 包初始化和公共接口
│   ├── agents/                  # Agent实现
│   │   ├── __init__.py
│   │   └── sql_agent.py         # EnhancedSQLAgent核心实现
│   ├── tools/                   # 工具系统
│   │   ├── sql_toolkit.py       # 增强SQL工具包
│   │   └── database_tools.py    # 6个专业数据库工具
│   ├── config/                  # 配置管理
│   │   ├── settings.py          # 统一设置管理
│   │   ├── llm_config.py        # LLM配置和创建
│   │   └── database_config.py   # 数据库连接配置
│   └── utils/                   # 工具函数
│       ├── __init__.py
│       ├── embeddings.py        # 嵌入模型管理
│       └── retrieval.py         # 向量检索系统
├── examples/                    # 使用示例
│   ├── basic_usage.py           # 基础交互式示例
│   └── advanced_usage.py        # 高级功能演示
├── tests/                       # 测试套件
│   ├── test_agents/             # Agent功能测试
│   ├── test_tools/              # 工具功能测试
│   └── test_config/             # 配置系统测试
├── docs/                        # 项目文档
├── pyproject.toml              # 项目配置
├── requirements.txt            # 生产依赖
├── requirements-dev.txt        # 开发依赖
└── .env                        # 环境配置

"""
LLM配置模块
提供统一的LLM配置和创建接口，支持多种LLM提供商。
"""
from typing import Optional, Dict, Any
from dataclasses import dataclass
from langchain_core.language_models import BaseLanguageModel
from langchain_deepseek import ChatDeepSeek
from langchain_agent.config.settings import get_settings

def create_deepseek_llm(config: Optional[LLMConfig] = None) -> ChatDeepSeek:
    """
    创建DeepSeek语言模型实例
    Args:
        config: LLM配置，如果为None则使用默认设置
    Returns:
        ChatDeepSeek: DeepSeek聊天模型实例
    Raises:
        ValueError: 当API密钥无效时
        Exception: 当创建模型失败时
    """
    if config is None:
        settings = get_settings()
        config = LLMConfig(
            api_key=settings.deepseek_api_key,
            model=settings.llm_model,
            temperature=settings.llm_temperature,
            max_tokens=settings.llm_max_tokens,
            timeout=settings.llm_timeout
        )
    if not config.api_key:
        raise ValueError(
            "DeepSeek API key is required. Please set DEEPSEEK_API_KEY in .env file.\n"
            "Get your API key from: https://platform.deepseek.com/api_keys"
        )
    try:
        llm = ChatDeepSeek(
            api_key=config.api_key,
            model=config.model,
            temperature=config.temperature,
            max_tokens=config.max_tokens,
            timeout=config.timeout,
            **config.extra_params
        )
        return llm

"""
数据库配置模块
提供数据库连接和配置管理。
"""
from typing import Optional, Dict, Any
from dataclasses import dataclass
from pathlib import Path
from langchain_community.utilities.sql_database import SQLDatabase

def create_database_connection(config: Optional[DatabaseConfig] = None) -> SQLDatabase:
    """
    创建数据库连接
    Args:
        config: 数据库配置，如果为None则使用默认设置
    Returns:
        SQLDatabase: LangChain数据库连接实例
    Raises:
        ValueError: 当配置无效时
        Exception: 当连接失败时
    """
    if config is None:
        config = DatabaseConfig(
            database_path=settings.database_path,
            database_type=settings.database_type
        )
    config.validate()
    try:
        db = SQLDatabase.from_uri(
            config.database_uri,
            **config.connection_params
        )
        return db
    except Exception as e:
        raise Exception(f"Failed to create database connection: {str(e)}")

"""
数据库工具
提供各种数据库操作工具，遵循LangChain工具设计模式。
"""
from typing import Optional, Dict, Any
from langchain_core.tools import BaseTool
from pydantic import Field

class TableSchemaTool(BaseTool):
    """表结构查询工具"""
    name: str = "table_schema"
    description: str = (
        "获取指定表的结构信息，包括列名、数据类型等，输入表名"
    )
    db: SQLDatabase = Field(exclude=True)
    def _run(self, table_name: str) -> str:
        """查询表结构"""
        try:
            # 验证表是否存在
            available_tables = self.db.get_usable_table_names()
            if table_name not in available_tables:
                return f"表 '{table_name}' 不存在。可用的表: {', '.join(available_tables)}"
            # 获取表结构
            table_info = self.db.get_table_info_no_throw([table_name])
            return f"表 '{table_name}' 的结构信息:\n{table_info}"
        except Exception as e:
            return f"查询表结构时出错: {str(e)}"

"""
增强的SQL工具包
基于LangChain的SQLDatabaseToolkit，提供增强的SQL查询功能。
"""
from typing import List, Optional
from langchain_core.tools import BaseTool
from langchain_community.agent_toolkits.sql.toolkit import SQLDatabaseToolkit

    def _create_enhanced_tools(self) -> List[BaseTool]:
        """创建增强的工具集合"""
        tools = [
            DatabaseInfoTool(db=self.db),
            TableListTool(db=self.db),
            TableSchemaTool(db=self.db),
            SampleDataTool(db=self.db),
            RecordCountTool(db=self.db),
            QueryExecutorTool(db=self.db),
        ]

"""
检索工具
提供工具检索功能，用于根据查询内容选择最相关的工具。
"""
from typing import List, Callable
from langchain_core.tools import BaseTool
from langchain_core.documents import Document
from langchain_community.vectorstores import FAISS
from langchain_agent.utils.embeddings import create_embeddings_with_fallback, check_model_cache

        # 为每个工具创建文档
        docs = [
            Document(
                page_content=tool.description,
                metadata={"index": i, "tool_name": tool.name}
            )
            for i, tool in enumerate(tools)
        ]
        # 创建FAISS向量存储
        vectorstore = FAISS.from_documents(docs, embeddings)
        def retrieve_tools(query: str) -> List[BaseTool]:
            """根据查询使用向量检索选择相关工具"""
            try:
                # 使用向量检索
                relevant_docs = vectorstore.similarity_search(query, k=top_k)
                # 根据检索结果返回工具
                retrieved_tools = []
                for doc in relevant_docs:
                    tool_index = doc.metadata["index"]
                    if 0 <= tool_index < len(tools):
                        retrieved_tools.append(tools[tool_index])
                if settings.verbose:
                    print(f"🔍 使用向量检索选择工具: {[t.name for t in retrieved_tools]}")
                return retrieved_tools
            except Exception as e:
                print(f"Warning: Tool retrieval failed: {e}, returning all tools")
                return tools

embeddings = HuggingFaceEmbeddings(
    model_name=model_name,
    model_kwargs=model_kwargs,
    encode_kwargs={
        'normalize_embeddings': True,
        'batch_size': 32,  # 批处理大小
    },
    cache_folder=cache_dir,  # 指定缓存目录
)

# 定义提示模板
template = """你是一个专业的SQL数据库查询助手。你可以使用以下工具来帮助用户查询数据库:
{tools}
⚠️ 严格安全规则 - 必须遵守:
1. 绝对禁止执行或提供任何修改数据的SQL语句
2. 禁止的操作包括: DELETE, UPDATE, INSERT, DROP, ALTER, CREATE, TRUNCATE
3. 如果用户要求这些操作，必须立即拒绝，不提供任何相关SQL语句
4. 只能帮助用户查询数据(SELECT)，不能修改数据
5. 当遇到危险请求时，回答格式: "抱歉，我不能执行[操作类型]操作，因为这会修改数据库。我只能帮助您查询数据。"

使用以下格式回答:
问题: 用户的输入问题
思考: 你应该思考要做什么
动作: 要采取的动作，应该是 [{tool_names}] 中的一个
动作输入: 动作的输入
观察结果: 动作的结果
... (这个思考/动作/动作输入/观察结果可以重复N次)
思考: 我现在知道最终答案了
最终答案: 对原始问题的最终答案

开始!
问题: {input}
思考: {agent_scratchpad}"""

# 创建提示模板（使用官方推荐的tools_getter模式）
prompt = CustomPromptTemplate(
    template=template,
    tools_getter=tools_getter,  # 动态工具检索
    input_variables=["input", "intermediate_steps"]
)

# 检查是否包含最终答案
if "最终答案:" in llm_output:
    return AgentFinish(
        return_values={"output": llm_output.split("最终答案:")[-1].strip()},
        log=llm_output,
    )

# 解析动作
regex = r"动作:\s*(.*?)\n动作输入:\s*(.*)"
match = re.search(regex, llm_output, re.DOTALL)

# 创建Agent执行器
return AgentExecutor.from_agent_and_tools(
    agent=agent,
    tools=all_tools,  # 提供所有工具，但通过tools_getter动态选择
    verbose=verbose,
    max_iterations=self.settings.max_iterations
)

# 执行查询（工具选择在提示模板中动态进行）
result = self.agent_executor.run(question)