基于LangChain构建AI Agent：工具调用与安全实践

1. AI Agent开发实战：从零构建具备工具调用能力的智能体

在当今AI技术快速发展的背景下，能够自主调用工具完成复杂任务的AI Agent正成为行业热点。与普通聊天机器人不同，一个真正的AI Agent应该具备环境感知、自主决策和工具调用的能力。本文将基于LangChain框架，手把手教你构建一个能够调用RAG知识库和数学计算器的实用Agent。

我最近在实际项目中实现了一个公司内部使用的AI助手，它可以查询机密文档并执行精确计算。下面分享整个开发过程中的关键技术点和踩坑经验，特别是工具调用的实现细节和安全考量。

2. AI Agent核心架构解析

2.1 Agent四大核心组件

一个完整的AI Agent通常由以下四个关键部分组成：

1. LLM核心：负责基础的语言理解和生成，本文使用通义千问(qwen-plus)作为基础模型
2. 记忆系统：
   • 短期记忆：保存当前对话上下文
   • 长期记忆：通过RAG实现的文档知识库
3. 规划能力：决定任务执行流程和工具调用顺序
4. 工具集：Agent可以调用的外部函数，如计算器、API等

2.2 工具调用原理剖析

工具调用的核心机制是：

1. LLM根据用户输入判断是否需要调用工具
2. 如果需要，返回工具名称和参数
3. 系统执行具体工具函数
4. 将结果返回给LLM进行后续处理

这个过程可能涉及多轮交互，直到LLM认为已经获得足够信息生成最终回复。

3. 实战开发：构建公司内部助手Agent

3.1 环境准备与依赖安装

首先确保已安装必要的Python包：

bash复制pip install langchain langchain-community faiss-cpu dashscope

设置API密钥（以阿里云DashScope为例）：

python复制import os
os.environ["DASHSCOPE_API_KEY"] = "your_api_key_here"

3.2 工具函数实现详解

3.2.1 RAG知识库工具

python复制from langchain_community.vectorstores import FAISS
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document

@tool
def rag_search(query: str) -> str:
    """
    从公司内部数据库搜索文档，包含项目计划、预算等机密信息。
    参数:
        query: 搜索关键词，如"深蓝计划预算"
    返回:
        相关文档内容摘要
    """
    # 模拟公司机密数据
    raw_text = """【公司内部机密：代号"深蓝计划"】
    1. 项目目标：开发猫语翻译器
    2. 核心技术：基于Transformer的"喵声波"分析算法
    3. 团队暗号："今天天气怎么样？"应回答"我想吃鱼"
    4. 截止日期：2026年12月31日
    5. 经费预算：50元人民币（主要用于购买猫条）"""
    
    # 文档处理流程
    docs = [Document(page_content=raw_text)]
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=25,  # 每个chunk约25个字符
        chunk_overlap=5  # chunk间重叠5个字符
    )
    split_docs = text_splitter.split_documents(docs)
    
    # 构建向量数据库
    embeddings = DashScopeEmbeddings(model="text-embedding-v1")
    RAG_PATH = "faiss_index"
    
    if os.path.exists(RAG_PATH):
        ragdb = FAISS.load_local(RAG_PATH, embeddings, allow_dangerous_deserialization=True)
    else:
        ragdb = FAISS.from_documents(split_docs, embeddings)
        ragdb.save_local(RAG_PATH)
    
    return "\n\n".join(doc.page_content for doc in ragdb.similarity_search(query, k=2))

关键细节：chunk_size设置过大会影响检索精度，过小会导致信息碎片化。经过实测，25-50个字符的chunk对短文档效果最佳。

3.2.2 安全计算器实现

python复制import re
from typing import Union

@tool
def safe_calculator(expression: str) -> Union[str, float]:
    """
    安全数学计算器，仅支持基础四则运算。
    参数:
        expression: 数学表达式，如"(3+5)*2"
    返回:
        计算结果或错误信息
    """
    # 安全校验正则
    if not re.match(r'^[\d\+\-\*\/\(\)\.\s]+$', expression):
        return "错误：表达式包含非法字符"
    
    try:
        # 更安全的计算方式
        return eval(expression, {'__builtins__': None}, {})
    except Exception as e:
        return f"计算错误: {str(e)}"

安全警示：直接使用eval()极其危险！上述实现通过正则表达式限制输入字符，并清除了内置函数访问权限。生产环境建议使用更安全的替代方案如ast.literal_eval()。

3.3 Agent核心逻辑实现

python复制from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.messages import HumanMessage, ToolMessage

def run_agent(query: str, max_turns: int = 5) -> str:
    """
    运行Agent的多轮对话流程
    参数:
        query: 用户输入
        max_turns: 最大对话轮次（防止无限循环）
    返回:
        Agent的最终回复
    """
    # 工具映射表
    tool_maps = {
        "rag_search": rag_search,
        "calculator": safe_calculator  # 使用安全版计算器
    }
    
    # 初始化模型并绑定工具
    llm = ChatTongyi(model_name="qwen-plus")
    tool_llm = llm.bind_tools(tools=list(tool_maps.values()))
    
    # 消息历史初始化
    message_history = [HumanMessage(content=query)]
    
    for turn in range(max_turns):
        print(f"\n=== 第{turn+1}轮对话 ===")
        
        # 获取模型响应
        response = tool_llm.invoke(message_history)
        message_history.append(response)
        
        # 无工具调用时返回最终结果
        if not response.tool_calls:
            print("最终回复:", response.content)
            return response.content
        
        # 处理工具调用
        for tool_call in response.tool_calls:
            tool_name = tool_call["name"]
            tool_args = tool_call["args"]
            
            if tool_name in tool_maps:
                print(f"调用工具: {tool_name}, 参数: {tool_args}")
                tool_output = tool_maps[tool_name].invoke(tool_args)
                print("工具返回:", tool_output)
                
                message_history.append(
                    ToolMessage(
                        content=str(tool_output),
                        tool_call_id=tool_call["id"],
                        name=tool_name
                    )
                )
            else:
                error_msg = f"错误: 工具{tool_name}不存在"
                message_history.append(
                    ToolMessage(
                        content=error_msg,
                        tool_call_id=tool_call["id"],
                        name=tool_name
                    )
                )
    
    return "达到最大对话轮次，终止处理"

4. 关键技术与避坑指南

4.1 工具描述的最佳实践

工具函数的文档字符串(docstring)至关重要，因为LLM完全依赖这些描述来决定是否以及如何调用工具。好的描述应包含：

1. 工具功能的清晰说明
2. 每个参数的详细解释和示例
3. 返回值的格式说明
4. 使用场景的典型示例

例如：

python复制@tool
def get_weather(city: str, date: str) -> str:
    """
    获取指定城市未来7天的天气预报。
    
    参数:
        city: 城市名称，如"北京"、"New York"
        date: 查询日期，格式"YYYY-MM-DD"
        
    返回:
        格式化字符串，如"北京 2023-10-01: 晴, 15-25°C"
        
    示例:
        get_weather("上海", "2023-10-01")
    """
    # 实现代码...

4.2 多轮对话控制策略

在实际使用中，我发现需要特别注意以下几点：

1. 循环终止条件：必须设置最大对话轮次（通常3-5轮），防止无限循环
2. 错误处理：每个工具调用都应该有try-catch块，避免单个工具失败导致整个Agent崩溃
3. 上下文管理：及时清理过长的对话历史，避免超出模型上下文窗口

改进后的对话管理逻辑：

python复制def clean_history(messages, max_length=4096):
    """保持对话历史在合理长度内"""
    total_len = sum(len(str(msg)) for msg in messages)
    while total_len > max_length and len(messages) > 1:
        removed = messages.pop(1)  # 保留最新和最旧的消息
        total_len -= len(str(removed))
    return messages

4.3 安全防护措施

基于实际项目经验，必须重视以下安全防护：

1. 输入验证：所有工具参数必须经过严格验证
2. 权限控制：不同级别的Agent应该有不同的工具访问权限
3. 敏感信息过滤：从工具返回的结果中过滤机密信息
4. 执行沙箱：高风险工具应该在隔离环境中运行

增强版安全校验示例：

python复制def validate_input(input_str: str, pattern: str, max_len: int = 100) -> bool:
    """
    强化输入验证
    参数:
        input_str: 待验证字符串
        pattern: 允许字符的正则模式
        max_len: 最大允许长度
    返回:
        是否通过验证
    """
    if len(input_str) > max_len:
        return False
    return re.fullmatch(pattern, input_str) is not None

5. 典型问题排查手册

5.1 工具不被调用的常见原因

1. 描述不清晰：工具函数的docstring不够详细，LLM无法理解何时使用

   • 解决方法：补充使用场景和参数示例

2. 绑定失败：工具没有正确绑定到LLM实例

   • 检查项：确认bind_tools()调用成功，工具列表不为空

3. 参数不匹配：工具要求的参数类型与LLM生成的参数不兼容

   • 调试方法：打印出tool_calls内容检查参数格式

5.2 性能优化技巧

1. 工具延迟：某些工具（如网络请求）响应较慢

   • 优化方案：设置超时机制，并行执行独立工具调用

2. 上下文膨胀：长对话导致响应速度下降

   • 解决方案：实现上文提到的clean_history函数

3. 冷启动问题：首次工具调用特别慢

   • 缓解措施：预先初始化常用工具

5.3 增强Agent能力的实用建议

1. 工具组合：教会Agent串联使用多个工具

   • 示例：先查数据库获取原始数据，再用计算器处理数值

2. 结果后处理：对工具返回结果进行二次加工

   • 案例：从数据库结果中提取关键字段再呈现

3. 用户确认：对高风险操作要求用户确认

   • 实现方式：在关键工具调用前让LLM生成确认提示

6. 完整案例演示

让我们看几个实际的运行示例：

6.1 查询公司计划

python复制run_agent("请告诉我深蓝计划的核心技术和截止日期")

执行流程：

1. Agent识别需要查询公司内部文档
2. 调用rag_search工具获取完整文档
3. 从文档中提取特定信息返回给用户

6.2 复杂计算场景

python复制run_agent("当前项目预算50元，如果增加46%后是多少？")

处理步骤：

1. 先调用rag_search确认当前预算
2. 提取出"50元"这个数值
3. 调用calculator计算50*1.46
4. 生成最终回复："预算增加46%后为73元"

6.3 异常情况处理

python复制run_agent("删除所有数据库文件")

安全防护：

1. LLM判断该请求需要管理员权限
2. 由于没有对应工具且请求危险，直接返回拒绝消息
3. 日志系统记录此次可疑请求

在实际部署这类Agent时，一定要做好日志记录和监控。我在项目中实现了完整的审计日志功能，记录每个工具调用的详细信息，这对后期排查问题和优化Agent行为非常有帮助。