LangChain智能体开发入门：空气质量查询实战

1. 项目概述：LangChain智能体开发入门指南

作为一名长期从事AI应用开发的工程师，我经常遇到新手开发者对大模型应用开发既向往又畏惧的情况。今天，我将通过一个空气质量查询智能体的完整开发案例，带大家从零开始掌握LangChain智能体的核心开发技能。这个项目特别适合具备基础Python能力的开发者，通过3-4小时的学习就能搭建出你的第一个可交互AI应用。

LangChain智能体的本质是"大模型+工具+记忆"的三位一体架构。就像组装一台电脑，我们需要选择合适的大脑（大模型）、配备必要的外设（工具），还要确保它有良好的记忆管理能力。与传统编程不同，这种架构让我们能够快速构建出理解自然语言、调用外部API、保持对话上下文的智能应用。

2. 环境准备与安全配置

2.1 开发环境搭建

在开始编码前，我们需要配置正确的开发环境。许多新手容易犯依赖冲突的错误，特别是当同时安装多个AI相关库时。以下是经过验证的依赖组合：

bash复制# 核心依赖
pip install langchain-core==0.1.0 langchain-openai==0.1.0 langchain-community==0.0.1

# 辅助工具
pip install python-dotenv==1.0.0 requests==2.31.0

# 开发工具（可选但推荐）
pip install ipython==8.12.0 black==23.3.0

这里特别说明几个关键选择：

• langchain-core包含了最基础的链式调用和智能体逻辑
• langchain-openai提供了与OpenAI兼容API的对接能力
• langchain-community包含各种第三方工具集成

重要提示：避免直接安装langchain元包，它可能带来不必要的依赖冲突。按需安装细分模块是更专业的做法。

2.2 安全配置最佳实践

API密钥管理是AI开发中最容易被忽视的安全环节。我见过太多开发者将密钥硬编码在代码中然后上传到GitHub，导致严重的经济损失。以下是企业级的安全实践：

1. 创建.env文件：

ini复制# .env.example (记得复制为.env并填写真实值)
OPENAI_API_KEY=your_api_key_here
AIR_QUALITY_API_KEY=your_aqi_key_here
MODEL_NAME=gpt-3.5-turbo

2. 配置.gitignore：

gitignore复制# 确保不会提交敏感文件
.env
*.env.local

3. 使用安全加载方式：

python复制from dotenv import load_dotenv
import os

load_dotenv(override=True)  # 显式禁止覆盖现有环境变量

api_key = os.environ["OPENAI_API_KEY"]  # 使用enviroment访问更安全

这种配置方式有三大优势：

• 密钥永远不会进入版本控制系统
• 不同环境可以轻松切换配置
• 符合12-factor应用原则

3. 核心模块开发

3.1 大模型初始化与配置

初始化大模型是构建智能体的第一步，这相当于为我们的AI赋予"大脑"。以下是经过生产环境验证的初始化代码：

python复制from langchain_openai import ChatOpenAI

def create_llm_model():
    return ChatOpenAI(
        model=os.getenv("MODEL_NAME", "gpt-3.5-turbo"),
        temperature=0.3,  # 平衡创造性和准确性
        max_tokens=1024,
        frequency_penalty=0.5,  # 降低重复内容
        presence_penalty=0.5,  # 鼓励多样性
        timeout=30,  # 避免长时间挂起
    )

关键参数解析：

• temperature：0.3是一个平衡值，适合查询类应用。创作类可提高到0.7
• max_tokens：限制响应长度，防止意外消耗
• timeout：必须设置，避免网络问题导致线程阻塞

3.2 工具开发实战

工具是智能体与外界交互的"手脚"。我们以空气质量查询为例，展示如何开发健壮的工具：

python复制from typing import Optional
from langchain_core.tools import tool
import requests
from pydantic import BaseModel, Field

class AirQualityInput(BaseModel):
    location: str = Field(description="城市名称，如'北京'")
    days: Optional[int] = Field(1, description="查询天数，默认为1")

@tool(args_schema=AirQualityInput)
def get_air_quality(location: str, days: int = 1) -> str:
    """查询指定城市未来N天的空气质量预测"""
    try:
        # 这里使用模拟API，实际应替换为真实服务
        base_url = "https://api.airvisual.com/v2/city"
        params = {
            "city": location,
            "days": days,
            "key": os.getenv("AIR_QUALITY_API_KEY")
        }
        
        response = requests.get(base_url, params=params, timeout=10)
        response.raise_for_status()
        
        data = response.json()
        return format_air_quality_data(data)
    except requests.exceptions.RequestException as e:
        return f"查询失败：{str(e)}"
    except Exception as e:
        return "处理空气质量数据时发生意外错误"

def format_air_quality_data(data: dict) -> str:
    """格式化API响应为易读文本"""
    result = []
    for day in data.get("data", []):
        day_info = (
            f"日期：{day['date']}\n"
            f"空气质量指数(AQI)：{day['aqi']}\n"
            f"主要污染物：{day['main_pollutant']}\n"
            f"建议：{day['health_recommendations']}"
        )
        result.append(day_info)
    return "\n\n".join(result) if result else "未获取到有效数据"

这个工具实现有几个值得注意的亮点：

1. 使用Pydantic模型进行输入验证
2. 完整的错误处理机制
3. 响应数据格式化处理
4. 明确的类型注解和文档字符串

4. 智能体组装与记忆管理

4.1 智能体组装

有了大脑和工具，现在我们需要将它们组装成完整的智能体：

python复制from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate

def create_agent(llm, tools):
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是专业的空气质量助手，用中文回答。"),
        ("human", "{input}"),
        ("placeholder", "{agent_scratchpad}"),
    ])
    
    agent = create_openai_tools_agent(llm, tools, prompt)
    return AgentExecutor(agent=agent, tools=tools, verbose=True)

这里的关键点：

• 使用专门的create_openai_tools_agent而非通用agent创建方法
• 明确的系统提示指导AI行为
• verbose=True便于调试

4.2 记忆管理进阶实现

对话记忆是智能体最难处理的部分之一。以下是经过优化的记忆管理方案：

python复制from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_core.runnables import RunnableLambda

class ConversationManager:
    def __init__(self, max_turns=5, max_tokens=2000):
        self.max_turns = max_turns
        self.max_tokens = max_tokens
        self.token_counter = llm.get_num_tokens  # 假设llm已初始化

    def trim_messages(self, messages: list[BaseMessage]) -> list[BaseMessage]:
        """智能修剪对话历史"""
        if not messages:
            return messages

        # 保留系统消息
        trimmed = [msg for msg in messages if msg.type == "system"]
        other_msgs = [msg for msg in messages if msg.type != "system"]
        
        # 按时间倒序，保留最新的
        other_msgs = other_msgs[-self.max_turns * 2 :]  # 每次对话通常有2条消息
        
        # 计算token数
        total_tokens = sum(self.token_counter(msg.content) for msg in other_msgs)
        
        # 如果还是太长，按token数修剪
        if total_tokens > self.max_tokens:
            temp_msgs = []
            current_tokens = 0
            for msg in reversed(other_msgs):
                msg_tokens = self.token_counter(msg.content)
                if current_tokens + msg_tokens <= self.max_tokens:
                    temp_msgs.insert(0, msg)
                    current_tokens += msg_tokens
                else:
                    break
            other_msgs = temp_msgs
        
        return trimmed + other_msgs

# 使用示例
conv_manager = ConversationManager()
agent_executor = create_agent(llm, tools).with_config(
    runnable_config={"configurable": {"session_id": "123"}}
)
agent_with_memory = RunnableLambda(conv_manager.trim_messages) | agent_executor

这个记忆管理系统实现了：

1. 基于对话轮数和token数的双重限制
2. 智能保留系统消息
3. 动态调整保留内容
4. 与LangChain执行器的无缝集成

5. 交互实现与测试

5.1 流式交互实现

为了让用户体验更好，我们实现流式响应：

python复制from langchain_core.runnables import RunnableConfig

def chat_loop(agent):
    history = []
    print("空气质量助手已启动！输入'退出'结束对话")
    
    while True:
        try:
            user_input = input("你：")
            if user_input.lower() in ["退出", "exit", "quit"]:
                break
                
            history.append(HumanMessage(content=user_input))
            
            print("助手：", end="", flush=True)
            full_response = ""
            
            for chunk in agent.stream(
                {"input": user_input, "chat_history": history},
                config=RunnableConfig(configurable={"session_id": "123"})
            ):
                if "output" in chunk:
                    print(chunk["output"][len(full_response):], end="", flush=True)
                    full_response = chunk["output"]
            
            history.append(AIMessage(content=full_response))
            print("\n")
            
        except KeyboardInterrupt:
            print("\n对话结束")
            break
        except Exception as e:
            print(f"\n发生错误：{str(e)}")
            continue

5.2 完整测试流程

让我们测试完整的智能体：

python复制# 初始化所有组件
llm = create_llm_model()
tools = [get_air_quality]
agent = create_agent(llm, tools)

# 启动对话
chat_loop(agent)

测试案例：

code复制你：北京的空气质量如何？
助手：正在查询北京空气质量...
当前AQI指数为78，空气质量良好。主要污染物为PM2.5。建议正常户外活动。

你：那上海呢？
助手：正在查询上海空气质量...
当前AQI指数为65，空气质量优。没有主要污染物。非常适合户外活动。

你：显示历史对话
助手：已显示最近2轮对话：
1. 你：北京的空气质量如何？
   助手：AQI 78，良好
2. 你：那上海呢？
   助手：AQI 65，优

你：退出

6. 生产环境优化建议

6.1 性能优化技巧

1. 工具并行化：

python复制from langchain.agents import ToolExecutor

tool_executor = ToolExecutor(tools)
# 在agent配置中设置parallel=True

2. 响应缓存：

python复制from langchain.cache import SQLiteCache
import langchain

langchain.llm_cache = SQLiteCache(database_path=".langchain.db")

3. 超时控制：

python复制agent = create_agent(llm, tools).with_config(
    {"run_name": "AirQualityAgent", "max_execution_time": 30}
)

6.2 监控与日志

1. 集成LangSmith：

python复制import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "AirQualityBot"

2. 自定义日志：

python复制import logging
from langchain.callbacks import FileCallbackHandler

log_handler = FileCallbackHandler("agent.log")
logging.basicConfig(level=logging.INFO)

6.3 安全增强

1. 输入净化：

python复制from langchain_core.utils import sanitize_input

def safe_input(prompt: str) -> str:
    user_input = input(prompt)
    return sanitize_input(user_input)

2. 输出过滤：

python复制from langchain.output_parsers import CommaSeparatedListOutputParser

output_parser = CommaSeparatedListOutputParser()
agent = agent | output_parser

7. 常见问题解决方案

7.1 工具调用失败

症状：智能体无法正确调用工具或返回错误结果

排查步骤：

1. 单独测试工具函数是否正常工作
2. 检查工具的描述文档是否清晰
3. 验证输入参数是否符合预期
4. 查看LangSmith跟踪了解决策过程

解决方案：

python复制@tool
def get_air_quality(city: str) -> str:
    """查询城市空气质量。输入必须是明确的城市名称。"""
    try:
        # 添加输入验证
        if not isinstance(city, str) or len(city) < 2:
            return "请输入有效的城市名称"
        
        # 其余逻辑不变
    except Exception as e:
        return f"工具执行出错：{str(e)}"

7.2 记忆丢失问题

症状：智能体忘记之前的对话内容

可能原因：

1. 记忆修剪过于激进
2. 会话ID未正确传递
3. 消息类型处理不当

修复方案：

python复制# 增强的记忆管理器
class EnhancedMemoryManager:
    def __init__(self):
        self.essential_messages = set()  # 存储关键消息ID
    
    def mark_essential(self, message: BaseMessage):
        self.essential_messages.add(id(message))
    
    def trim_messages(self, messages: list[BaseMessage]) -> list[BaseMessage]:
        return [
            msg for msg in messages 
            if id(msg) in self.essential_messages or msg.type == "human"
        ][-10:]  # 保留最近10条必要消息

7.3 响应速度慢

优化策略：

1. 启用流式响应
2. 实现工具预加载
3. 使用更轻量级的模型

配置示例：

python复制# 快速响应配置
fast_llm = ChatOpenAI(
    model="gpt-3.5-turbo",
    temperature=0,
    streaming=True,
    max_tokens=500
)

# 预加载工具
preloaded_tools = ToolExecutor(tools).preload()

8. 项目扩展方向

8.1 多工具集成

扩展工具集可以让智能体更强大：

python复制@tool
def get_weather_forecast(city: str, days: int = 3) -> str:
    """获取城市天气预报"""
    # 实现类似空气质量查询的逻辑

@tool
def get_pollution_explanation(pollutant: str) -> str:
    """解释污染物对人体健康的影响"""
    knowledge = {
        "PM2.5": "可深入肺部，引发呼吸系统疾病",
        "O3": "刺激呼吸道，加重哮喘"
    }
    return knowledge.get(pollutant, "暂无此污染物信息")

tools = [get_air_quality, get_weather_forecast, get_pollution_explanation]

8.2 多模态扩展

集成图像识别能力：

python复制from langchain_community.tools import ImageCaptionTool

image_tool = ImageCaptionTool()
tools.append(image_tool)

# 现在智能体可以处理图片输入了

8.3 领域知识增强

添加专业领域知识库：

python复制from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

# 假设我们有环境健康PDF文档
vectorstore = FAISS.from_texts(
    ["PM2.5长期暴露增加肺癌风险", "臭氧污染对儿童影响更大"],
    OpenAIEmbeddings()
)

retriever = vectorstore.as_retriever()
tools.append(retriever)

9. 项目部署方案

9.1 本地API服务

使用FastAPI构建本地服务：

python复制from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.post("/chat")
async def chat_endpoint(query: str):
    response = agent.invoke({"input": query})
    return {"response": response["output"]}

9.2 云部署方案

推荐部署平台：

1. Vercel：适合小型应用，简单易用
2. AWS Lambda：按需付费，自动扩展
3. Google Cloud Run：容器化部署，管理方便

Dockerfile示例：

dockerfile复制FROM python:3.9-slim

WORKDIR /app
COPY . .

RUN pip install --no-cache-dir -r requirements.txt

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

10. 持续学习建议

要成为LangChain开发专家，我建议：

1. 官方文档精读：LangChain文档更新频繁，每月至少浏览一次变更日志
2. 社区参与：加入LangChain Discord，参与问题讨论
3. 项目实践：每周尝试一个小型创新项目
4. 技术分享：写博客记录学习心得，教学相长

推荐学习路径：

1. 掌握基础链式调用
2. 精通工具开发和记忆管理
3. 学习高级编排（LangGraph）
4. 深入性能优化和监控

记住，AI开发是快速演进的领域，保持持续学习的态度比掌握任何特定技术都重要。我在开发这个空气质量智能体的过程中，就经历了三次大的架构调整，每次重构都带来了显著的性能提升和更简洁的代码结构。