使用 Python 和 LangChain 构建智能 AI 助手：从入门到实战

16 4 月, 2026 7点热度 0人点赞 0条评论

引言

随着大语言模型的快速发展，构建智能 AI 助手已成为开发者的热门需求。本文将手把手教你使用 Python 和 LangChain 框架，从零开始打造一个实用的 AI 助手。无论你是编程新手还是经验丰富的开发者，都能从中获得启发喵～

什么是 LangChain？

LangChain 是一个用于开发语言模型应用的开源框架，它提供了模块化抽象，让开发者能够轻松组合各种组件，构建复杂的 AI 应用。核心优势包括：

组件化设计：链、代理、工具等模块可自由组合
多模型支持：兼容 OpenAI、Anthropic、本地模型等
记忆管理：支持对话历史持久化
工具集成：轻松连接外部 API 和数据源

环境准备

首先安装必要的依赖包：

# 创建虚拟环境
python -m venv ai-assistant
cd ai-assistant
source bin/activate  # Windows 使用：Scripts\activate

# 安装核心依赖
pip install langchain langchain-openai python-dotenv
pip install langchain-community faiss-cpu

创建.env文件存储 API 密钥：

OPENAI_API_KEY=your-api-key-here
MODEL_NAME=gpt-3.5-turbo

构建基础 AI 助手

让我们从最简单的对话助手开始：

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from dotenv import load_dotenv
import os

# 加载环境变量
load_dotenv()

# 初始化模型
llm = ChatOpenAI(
    model=os.getenv("MODEL_NAME", "gpt-3.5-turbo"),
    temperature=0.7
)

# 定义系统提示
system_prompt = """你是一个专业的技术助手，擅长解答编程问题。
请用简洁清晰的语言回答，必要时提供代码示例。"""

# 创建对话链
def chat_with_ai(user_input: str) -> str:
    messages = [
        SystemMessage(content=system_prompt),
        HumanMessage(content=user_input)
    ]
    response = llm.invoke(messages)
    return response.content

# 测试
if __name__ == "__main__":
    print("AI 助手已就绪！输入\"退出\"结束对话")
    while True:
        user_input = input("\n你：")
        if user_input == "退出":
            break
        response = chat_with_ai(user_input)
        print(f"AI: {response}")

添加记忆功能

让助手记住对话历史，实现更自然的连续对话：

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain

# 创建记忆组件
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

# 创建对话链
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

# 带记忆的对话
def chat_with_memory(user_input: str) -> str:
    response = conversation.invoke({"input": user_input})
    return response["response"]

集成自定义工具

让 AI 助手能够执行实际任务，比如查询天气、搜索信息等：

from langchain.tools import tool
from langchain.agents import initialize_agent, AgentType
import requests

@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气信息"""
    api_key = "your-weather-api-key"
    url = f"http://api.openweathermap.org/data/2.5/weather?q={city}&appid={api_key}&units=metric&lang=zh_cn"
    try:
        response = requests.get(url)
        data = response.json()
        weather = data["weather"][0]["description"]
        temp = data["main"]["temp"]
        return f"{city}当前天气：{weather}，温度：{temp}°C"
    except:
        return "暂时无法获取天气信息"

@tool
def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        result = eval(expression)
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算错误：{str(e)}"

# 初始化工具代理
tools = [get_weather, calculate]
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# 使用代理
def run_agent_task(task: str) -> str:
    response = agent.run(task)
    return response

构建知识库问答系统

让 AI 助手基于你的文档回答问题：

from langchain.text_splitter import CharacterTextSplitter
from langchain_community.embeddings import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain.chains import RetrievalQA

# 加载文档
def load_documents(file_path: str) -> list:
    with open(file_path, "r", encoding="utf-8") as f:
        text = f.read()
    return [text]

# 创建向量存储
def create_vector_store(documents: list):
    # 分割文本
    text_splitter = CharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=200
    )
    chunks = text_splitter.split_documents(documents)
    
    # 创建嵌入
    embeddings = OpenAIEmbeddings()
    
    # 构建向量索引
    vectorstore = FAISS.from_documents(chunks, embeddings)
    return vectorstore

# 创建问答链
def create_qa_chain(vectorstore):
    qa_chain = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",
        retriever=vectorstore.as_retriever()
    )
    return qa_chain

# 使用示例
docs = load_documents("your-document.txt")
vectorstore = create_vector_store(docs)
qa_chain = create_qa_chain(vectorstore)

# 提问
question = "文档中关于 API 认证的部分是怎么说的？"
answer = qa_chain.run(question)
print(answer)

最佳实践与优化建议

1. 成本控制

使用max_tokens限制输出长度
对长文本进行摘要后再处理
缓存常见问题的回答

2. 性能优化

使用异步调用提高并发能力
批量处理多个请求
选择合适的模型（快速响应 vs 高质量）

3. 安全考虑

不要将 API 密钥硬编码在代码中
对用户输入进行验证和过滤
设置请求频率限制

完整项目示例

将以上组件整合为一个完整的 AI 助手应用：

# main.py - 完整的 AI 助手应用
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferMemory
from langchain.agents import initialize_agent, AgentType, Tool
from dotenv import load_dotenv
import os

load_dotenv()

class AIAssistant:
    def __init__(self):
        self.llm = ChatOpenAI(temperature=0.7)
        self.memory = ConversationBufferMemory(
            memory_key="chat_history",
            return_messages=True
        )
        self.tools = self._setup_tools()
        self.agent = self._create_agent()
    
    def _setup_tools(self):
        # 定义工具列表
        return []
    
    def _create_agent(self):
        return initialize_agent(
            self.tools,
            self.llm,
            agent=AgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION,
            memory=self.memory,
            verbose=True
        )
    
    def chat(self, user_input: str) -> str:
        response = self.agent.run({"input": user_input})
        return response
    
    def run(self):
        print("🤖 AI 助手已启动！输入\"退出\"结束")
        while True:
            user_input = input("\n你：")
            if user_input in ["退出", "quit", "exit"]:
                print("再见！")
                break
            response = self.chat(user_input)
            print(f"AI: {response}")

if __name__ == "__main__":
    assistant = AIAssistant()
    assistant.run()