用 LangChain 打造你的第一个 AI Agent：从零到一的实践指南（含关键代码）

## 项目架构### 目录结构```langchain-agent/├── langchain_agent/              # 核心代码包│   ├── __init__.py              # 包初始化和公共接口│   ├── agents/                  # Agent实现│   │   ├── __init__.py│   │   └── sql_agent.py         # EnhancedSQLAgent核心实现│   ├── tools/                   # 工具系统│   │   ├── __init__.py│   │   ├── sql_toolkit.py       # 增强SQL工具包│   │   └── database_tools.py    # 6个专业数据库工具│   ├── config/                  # 配置管理│   │   ├── __init__.py│   │   ├── 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                        # 环境配置```### 核心组件#### EnhancedSQLAgent- **ReAct架构**: 实现思考-行动-观察循环- **安全检查**: 多层防护机制- **工具检索**: 智能选择相关工具- **错误处理**: 完善的异常处理#### 专业工具集1. **database_info** - 数据库概览信息2. **table_list** - 表名列表查询3. **table_schema** - 表结构详细信息4. **sample_data** - 示例数据展示5. **record_count** - 记录数量统计6. **sql_query** - SQL查询执行#### 智能检索系统- **向量检索**: 基于FAISS的语义检索- **嵌入模型**: 支持本地和云端模型- **缓存优化**: 智能模型缓存管理

"""LLM配置模块提供统一的LLM配置和创建接口，支持多种LLM提供商。"""from typing import Optional, Dict, Anyfrom dataclasses import dataclassfrom langchain_core.language_models import BaseLanguageModelfrom langchain_deepseek import ChatDeepSeekfrom 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, Anyfrom dataclasses import dataclassfrom pathlib import Pathfrom langchain_community.utilities.sql_database import SQLDatabasefrom langchain_agent.config.settings import get_settings

def create_database_connection(config: Optional[DatabaseConfig] = None) -> SQLDatabase:    """    创建数据库连接
    Args:        config: 数据库配置，如果为None则使用默认设置
    Returns:        SQLDatabase: LangChain数据库连接实例
    Raises:        ValueError: 当配置无效时        Exception: 当连接失败时    """    if config is None:        settings = get_settings()        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, Anyfrom langchain_core.tools import BaseToolfrom langchain_community.utilities.sql_database import SQLDatabasefrom 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, Optionalfrom langchain_core.tools import BaseToolfrom langchain_core.language_models import BaseLanguageModelfrom langchain_community.utilities.sql_database import SQLDatabasefrom 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, Callablefrom langchain_core.tools import BaseToolfrom langchain_core.documents import Documentfrom langchain_community.vectorstores import FAISSfrom 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 if retrieved_tools else tools            except Exception as e:                print(f"Warning: Tool retrieval failed: {e}, returning all tools")                return tools
        return retrieve_tools

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

        # 检查是否包含最终答案        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)

graph TD    A[query方法调用] --> B[AgentExecutor.run启动]    B --> C[初始化intermediate_steps = []]    C --> D[开始推理循环]    D --> E[工具检索阶段]    E --> F[向量检索相关工具]    F --> G[构建提示模板]    G --> H[LLM推理阶段]    H --> I[发送提示给DeepSeek]    I --> J[解析LLM输出]    J --> K{输出类型判断}    K -->|AgentFinish| L[返回最终答案]    K -->|AgentAction| M[执行工具]    M --> N[调用SQL工具]    N --> O[更新intermediate_steps]    O --> D    L --> P[返回给用户]

# 在CustomPromptTemplate.format()中调用tools = self.tools_getter("显示orders表中最贵的订单记录")# 输出: 使用向量检索选择工具: ['table_schema', 'sql_query', 'sample_data', 'database_info', 'record_count', 'table_list']

prompt = """你是一个专业的SQL数据库查询助手。你可以使用以下工具来帮助用户查询数据库:
table_schema: 获取指定表的结构信息，包括列名、数据类型等，输入表名sql_query: 执行SQL查询获取具体数据记录，如查找最贵订单、价格最高商品、特定条件的记录等。用于复杂查询：最大值、最小值、排序、筛选、聚合统计、条件查询。输入完整的SQL SELECT语句。sample_data: 获取指定表的前5条示例数据，用于了解数据内容，输入表名database_info: 获取数据库的基本信息，包括表列表等record_count: 统计指定表中的记录总数，输入表名table_list: 列出数据库中的所有表名，用于了解数据库结构
⚠️ 严格安全规则 - 必须遵守:1. 绝对禁止执行或提供任何修改数据的SQL语句2. 禁止的操作包括: DELETE, UPDATE, INSERT, DROP, ALTER, CREATE, TRUNCATE3. 如果用户要求这些操作，必须立即拒绝，不提供任何相关SQL语句4. 只能帮助用户查询数据(SELECT)，不能修改数据5. 当遇到危险请求时，回答格式: "抱歉，我不能执行[操作类型]操作，因为这会修改数据库。我只能帮助您查询数据。"
使用以下格式回答:
问题: 用户的输入问题思考: 你应该思考要做什么动作: 要采取的动作，应该是 [table_schema, sql_query, sample_data, database_info, record_count, table_list] 中的一个动作输入: 动作的输入观察结果: 动作的结果... (这个思考/动作/动作输入/观察结果可以重复N次)思考: 我现在知道最终答案了最终答案: 对原始问题的最终答案
开始!
问题: 显示orders表中最贵的订单记录思考: """

# LLM分析提示并生成回应llm_output = """用户想要找到orders表中最贵的订单记录。我需要先了解orders表的结构，特别是金额相关的字段，然后执行查询找出最贵的记录。
动作: table_schema动作输入: orders"""

# CustomOutputParser.parse()解析LLM输出action = AgentAction(    tool="table_schema",    tool_input="orders",    log="用户想要找到orders表中最贵的订单记录...")

# 查找table_schema工具并执行tool = self.get_tool("table_schema")observation = tool.run("orders")
# 工具返回结果observation = """表 'orders' 的结构信息:
CREATE TABLE orders (        id INTEGER,         coupon_code VARCHAR(255),         create_time TIMESTAMP NOT NULL,         logistics_company VARCHAR(255),         note VARCHAR(255),         pay_time TIMESTAMP,         payment_method VARCHAR(255),         payment_transaction_id VARCHAR(255),         product_id BIGINT NOT NULL,         quantity INTEGER NOT NULL,         receiver_address VARCHAR(255) NOT NULL,         receiver_name VARCHAR(255) NOT NULL,         receiver_phone VARCHAR(255) NOT NULL,         specification VARCHAR(255),         status VARCHAR(255) NOT NULL,         total_amount NUMERIC(38, 2) NOT NULL,         tracking_number VARCHAR(255),         update_time TIMESTAMP NOT NULL,         user_id BIGINT NOT NULL,         PRIMARY KEY (id),         CHECK (status in ('PENDING_PAYMENT','PENDING_SHIPMENT','PENDING_RECEIPT','COMPLETED','CANCELLED')))
/*3 rows from orders table:id      coupon_code     create_time     logistics_company       note    pay_time        payment_method       payment_transaction_id  product_id      quantity        receiver_address        receiver_name receiver_phone  specification   status  total_amount    tracking_number update_time     user_id2       None    2024-11-26 07:57:55     None    None    None    None    None    2       4       上海市浦东新区张江高科技园区5号     张先生6 13812348152     None    CANCELLED       716.00  None    2024-04-06 15:37:55  73       None    2024-06-03 22:06:55     None    None    None    None    None    3       1       上海市浦东新区张江高科技园区48号    张先生15        13812340741     None    CANCELLED       413.00  None2024-04-08 12:41:55      894       None    2024-10-13 20:49:55     None    None    None    None    None    4       2       上海市浦东新区张江高科技园区37号    张先生4 13812342197     None    PENDING_RECEIPT 384.00  None    2024-04-18 16:51:55  33*/"""

intermediate_steps = [(action, observation)]

# 再次调用工具检索器（基于相同的查询）tools = self.tools_getter("显示orders表中最贵的订单记录")# 输出: 使用向量检索选择工具: ['table_schema', 'sql_query', 'sample_data', 'database_info', 'record_count', 'table_list']

prompt = """你是一个专业的SQL数据库查询助手。你可以使用以下工具来帮助用户查询数据库:
[工具列表相同...]
问题: 显示orders表中最贵的订单记录思考: 用户想要找到orders表中最贵的订单记录。我需要先了解orders表的结构，特别是金额相关的字段，然后执行查询找出最贵的记录。
动作: table_schema动作输入: orders
观察结果: 表 'orders' 的结构信息:[完整的表结构信息...]
思考: """

llm_output = """现在我已经了解了orders表的结构，看到有total_amount字段表示订单总金额。要查询最贵的订单记录，我需要按total_amount降序排序并取第一条记录。
动作: sql_query动作输入: SELECT * FROM orders ORDER BY total_amount DESC LIMIT 1"""

action = AgentAction(    tool="sql_query",    tool_input="SELECT * FROM orders ORDER BY total_amount DESC LIMIT 1",    log="现在我已经了解了orders表的结构...")

tool = self.get_tool("sql_query")observation = tool.run("SELECT * FROM orders ORDER BY total_amount DESC LIMIT 1")
# SQL查询结果observation = """[(744, None, datetime.datetime(2024, 4, 15, 9, 32, 55), None, None, None, None, None, 744, 1, '上海市浦东新区张江高科技园区744号', '张先生744', '13812340744', None, 'PENDING_RECEIPT', Decimal('997.00'), None, datetime.datetime(2024, 4, 15, 9, 32, 55), 744)]"""

intermediate_steps = [    (action1, observation1),  # table_schema的结果    (action2, observation2)   # sql_query的结果]

tools = self.tools_getter("显示orders表中最贵的订单记录")# 输出: 使用向量检索选择工具: [相同的工具列表]

prompt = """[工具列表...]
问题: 显示orders表中最贵的订单记录思考: [第一次的思考和行动...]观察结果: [表结构信息...]思考: [第二次的思考和行动...]观察结果: [SQL查询结果...]思考: """

llm_output = """我现在已经获得了orders表中最贵的订单记录。查询结果显示ID为744的订单，总金额为997元，是最贵的订单。我现在知道最终答案了。
最终答案: orders表中最贵的订单记录是ID为744的订单，总金额为997元。详细信息如下：- 订单ID: 744- 总金额: 997.00元- 创建时间: 2024-04-15 09:32:55- 收货人: 张先生744- 收货地址: 上海市浦东新区张江高科技园区744号- 收货电话: 13812340744- 订单状态: PENDING_RECEIPT (待收货)- 商品ID: 744- 数量: 1"""

final_result = AgentFinish(    return_values={"output": "orders表中最贵的订单记录是ID为744的订单..."},    log="我现在已经获得了orders表中最贵的订单记录...")

# AgentExecutor检测到AgentFinish，结束循环return final_result.return_values["output"]