Qwen3-4B-Thinking开源模型实战：使用LangChain构建带记忆的编程对话Agent

Asama浅间

1031人浏览 · 2026-03-21 02:28:57

Asama浅间 · 2026-03-21 02:28:57 发布

Qwen3-4B-Thinking开源模型实战：使用LangChain构建带记忆的编程对话Agent

1. 引言：当AI学会“思考”，编程助手能有多智能？

想象一下，你正在开发一个复杂的Python项目，遇到了一个棘手的问题。你向一个AI助手提问，它不仅能给出代码片段，还能记住你之前问过的所有问题，理解你项目的上下文，甚至能根据你之前的反馈调整回答策略。这听起来像是科幻电影里的场景，但现在，借助Qwen3-4B-Thinking模型和LangChain框架，我们可以轻松实现这样的智能编程对话助手。

今天要介绍的主角是Qwen3-4B-Thinking-2507-GPT-5-Codex-Distill-GGUF模型。这个模型在OpenAI的GPT-5-Codex的1000个高质量示例上进行了专门微调，特别擅长代码生成和编程问题解答。更重要的是，它具备“思考”能力——能够进行多步推理，处理复杂的编程任务。

但光有强大的模型还不够，我们需要一个框架来管理对话的上下文，让AI记住我们的对话历史，这就是LangChain的用武之地。LangChain是一个专门为构建基于大语言模型的应用而设计的框架，它提供了记忆管理、工具调用、链式处理等强大功能。

本文将带你从零开始，使用vllm部署Qwen3-4B-Thinking模型，然后通过LangChain构建一个带记忆功能的编程对话Agent，最后用chainlit搭建一个美观的Web界面。无论你是AI开发者还是普通程序员，都能跟着教程一步步实现自己的智能编程助手。

2. 环境准备与模型部署

2.1 理解我们的技术栈

在开始动手之前，先简单了解一下我们要用到的几个关键技术：

Qwen3-4B-Thinking模型：这是通义千问团队开发的4B参数模型，经过专门微调后具备优秀的代码生成能力。GGUF格式意味着它已经量化过，可以在消费级GPU上高效运行。
vllm：一个高性能的推理引擎，专门为大语言模型优化，能够提供极快的推理速度。
LangChain：大语言模型应用开发框架，核心功能包括对话记忆管理、工具调用、工作流编排等。
chainlit：一个专门为AI应用设计的Web界面框架，可以快速构建聊天界面。

2.2 检查模型部署状态

如果你使用的是预置的镜像环境，模型可能已经部署好了。我们可以通过以下命令检查部署状态：

# 查看模型服务日志
cat /root/workspace/llm.log

如果看到类似下面的输出，说明模型已经成功加载：

INFO 07-28 10:30:15 llm_engine.py:73] Initializing an LLM engine with config: model='Qwen3-4B-Thinking-2507-GPT-5-Codex-Distill-GGUF', tokenizer='Qwen3-4B-Thinking-2507-GPT-5-Codex-Distill-GGUF', ...
INFO 07-28 10:30:20 model_runner.py:58] Loading weights from /root/workspace/Qwen3-4B-Thinking-2507-GPT-5-Codex-Distill-GGUF
INFO 07-28 10:30:25 llm_engine.py:150] LLM engine is ready.

如果模型还在加载中，可能需要等待几分钟。vllm在首次加载模型时会进行一些优化操作，后续的推理速度会非常快。

2.3 快速测试模型

在开始构建复杂应用之前，我们先简单测试一下模型的基本功能。打开chainlit前端界面，你可以看到一个简洁的聊天窗口。

尝试问一些简单的编程问题，比如：

用Python写一个快速排序算法

或者更复杂一些的：

我有一个Pandas DataFrame，包含'name'、'age'、'salary'三列，如何按年龄分组并计算每组的平均工资？

观察模型的回答质量。Qwen3-4B-Thinking模型经过专门微调，应该能够给出准确、可运行的代码片段。

3. LangChain基础：理解记忆机制

3.1 为什么需要对话记忆？

传统的聊天机器人有一个明显的局限：它们通常只处理当前的问题，不记得之前的对话。这在编程场景中尤其不方便，因为编程问题往往是连续的、有上下文的。

比如，你可能会先问：“如何用Python读取CSV文件？”然后接着问：“那如果文件很大，怎么分块读取？”如果没有记忆功能，第二个问题就失去了上下文，AI无法理解“那”指的是什么。

LangChain提供了多种记忆机制来解决这个问题：

ConversationBufferMemory：最简单的记忆，保存所有对话历史
ConversationBufferWindowMemory：只保存最近N轮对话
ConversationSummaryMemory：对历史对话进行总结，节省token
ConversationKGMemory：基于知识图谱的记忆，能理解实体关系

3.2 安装必要的Python包

在开始编码之前，确保安装了必要的Python包：

pip install langchain langchain-community chainlit

如果你需要连接vllm服务，还需要安装：

pip install openai  # 用于连接vllm的OpenAI兼容接口

vllm提供了与OpenAI API兼容的接口，这意味着我们可以用标准的OpenAI客户端来调用它，这大大简化了集成工作。

4. 构建带记忆的编程对话Agent

4.1 连接vllm服务

首先，我们需要创建一个连接vllm服务的LangChain LLM对象：

from langchain_openai import ChatOpenAI
import os

# 设置vllm服务的地址
os.environ["OPENAI_API_BASE"] = "http://localhost:8000/v1"
os.environ["OPENAI_API_KEY"] = "token-abc123"  # vllm不需要真正的key，但需要设置一个值

# 创建LLM对象
llm = ChatOpenAI(
    model="Qwen3-4B-Thinking-2507-GPT-5-Codex-Distill-GGUF",
    temperature=0.3,  # 温度参数，控制创造性，编程场景建议较低值
    max_tokens=2048,   # 最大生成token数
)

这里有几个关键点需要注意：

temperature=0.3：对于编程任务，我们通常希望输出稳定、准确，所以温度设得较低
max_tokens=2048：根据模型的能力和我们的需求设置，对于代码生成来说通常足够
vllm默认在8000端口提供服务，但具体端口可能需要根据你的部署配置调整

4.2 创建对话记忆

接下来，我们创建一个带记忆的对话链：

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

# 创建记忆对象
memory = ConversationBufferMemory(
    memory_key="history",  # 记忆在prompt中的key
    return_messages=True   # 返回消息对象而不是字符串
)

# 创建对话链
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True  # 调试模式，可以看到详细的prompt构造过程
)

ConversationBufferMemory是最简单的记忆类型，它会保存所有的对话历史。对于编程对话场景，这通常是个不错的选择，因为我们需要完整的上下文来理解复杂的编程问题。

4.3 测试记忆功能

让我们测试一下记忆是否正常工作：

# 第一轮对话
response1 = conversation.predict(input="用Python写一个函数，计算斐波那契数列的第n项")
print("AI回复1:", response1)

# 第二轮对话 - 这里会用到记忆
response2 = conversation.predict(input="能不能优化一下，用记忆化递归？")
print("AI回复2:", response2)

# 查看记忆内容
print("\n当前记忆内容:")
print(memory.load_memory_variables({}))

运行这段代码，你会看到：

第一轮AI给出了基本的斐波那契函数
第二轮AI知道我们在讨论斐波那契数列，并给出了优化版本
记忆对象中保存了两轮完整的对话

这就是记忆机制的魅力——AI能够理解对话的连续性。

4.4 构建更智能的编程Agent

基本的对话链已经不错，但对于编程任务，我们可以做得更好。让我们创建一个专门的编程助手：

from langchain.agents import initialize_agent, AgentType
from langchain.tools import Tool
from langchain.memory import ConversationBufferWindowMemory

# 创建专门用于编程的记忆
programming_memory = ConversationBufferWindowMemory(
    memory_key="chat_history",
    k=10,  # 只保留最近10轮对话
    return_messages=True
)

# 定义一些工具（这里先定义空工具，后续可以扩展）
def python_code_executor(code: str) -> str:
    """执行Python代码并返回结果"""
    try:
        # 在实际应用中，这里应该安全地执行代码
        # 出于安全考虑，这里只返回提示
        return "代码执行功能需要在安全环境中实现"
    except Exception as e:
        return f"执行出错: {str(e)}"

# 创建工具列表
tools = [
    Tool(
        name="PythonCodeExecutor",
        func=python_code_executor,
        description="执行Python代码并返回结果。输入应该是有效的Python代码字符串。"
    ),
]

# 创建编程Agent
programming_agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION,
    memory=programming_memory,
    verbose=True,
    max_iterations=3,  # 最大迭代次数，防止无限循环
    handle_parsing_errors=True  # 处理解析错误
)

这个编程助手有几个特点：

使用ConversationBufferWindowMemory，只保留最近10轮对话，避免token过多
预留了代码执行工具接口（实际使用时需要安全考虑）
使用CONVERSATIONAL_REACT_DESCRIPTION类型的Agent，适合多轮对话场景

5. 使用chainlit构建Web界面

5.1 创建chainlit应用

chainlit让创建AI聊天界面变得非常简单。创建一个app.py文件：

import chainlit as cl
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
from langchain_openai import ChatOpenAI
import os

# 设置vllm连接
os.environ["OPENAI_API_BASE"] = "http://localhost:8000/v1"
os.environ["OPENAI_API_KEY"] = "token-abc123"

@cl.on_chat_start
async def on_chat_start():
    # 初始化LLM
    llm = ChatOpenAI(
        model="Qwen3-4B-Thinking-2507-GPT-5-Codex-Distill-GGUF",
        temperature=0.3,
        max_tokens=2048,
        streaming=True,  # 启用流式输出
    )
    
    # 创建记忆
    memory = ConversationBufferMemory(return_messages=True)
    
    # 创建对话链
    chain = ConversationChain(llm=llm, memory=memory, verbose=True)
    
    # 保存到用户会话
    cl.user_session.set("chain", chain)
    
    # 发送欢迎消息
    await cl.Message(
        content="你好！我是基于Qwen3-4B-Thinking的编程助手，我可以帮你解决编程问题，还能记住我们的对话历史。试试问我编程问题吧！"
    ).send()

@cl.on_message
async def on_message(message: cl.Message):
    # 获取对话链
    chain = cl.user_session.get("chain")
    
    # 创建回调处理器，用于流式输出
    cb = cl.AsyncLangchainCallbackHandler()
    
    # 调用模型
    response = await chain.apredict(
        input=message.content,
        callbacks=[cb]
    )
    
    # 发送回复
    await cl.Message(content=response).send()

这个简单的chainlit应用已经具备了基本功能：

聊天开始时初始化AI助手
处理用户消息并生成回复
支持流式输出（打字机效果）

5.2 增强界面功能

让我们添加一些实用功能，让界面更友好：

import chainlit as cl
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
from langchain_openai import ChatOpenAI
import os

@cl.on_chat_start
async def on_chat_start():
    # 设置应用配置
    settings = await cl.ChatSettings(
        [
            cl.input_widget.Slider(
                id="temperature",
                label="创造性",
                initial=0.3,
                min=0,
                max=1,
                step=0.1,
            ),
            cl.input_widget.Slider(
                id="max_tokens",
                label="最大生成长度",
                initial=2048,
                min=512,
                max=4096,
                step=256,
            ),
        ]
    ).send()
    
    # 保存设置
    cl.user_session.set("settings", settings)

@cl.set_starters
async def set_starters():
    # 设置快捷提问
    starters = [
        cl.Starter(
            label="写一个Python快速排序",
            message="用Python写一个快速排序算法，包含详细注释",
            icon="/public/python.svg"
        ),
        cl.Starter(
            label="Pandas数据清洗",
            message="我有一个包含缺失值的DataFrame，如何用Pandas进行数据清洗？",
            icon="/public/pandas.svg"
        ),
        cl.Starter(
            label="Flask Web应用",
            message="如何用Flask创建一个简单的REST API？",
            icon="/public/flask.svg"
        ),
        cl.Starter(
            label="异步编程",
            message="Python中的async/await怎么用？给个实际例子",
            icon="/public/async.svg"
        ),
    ]
    return starters

@cl.on_settings_update
async def on_settings_update(settings):
    # 当设置更新时重新初始化LLM
    llm = ChatOpenAI(
        model="Qwen3-4B-Thinking-2507-GPT-5-Codex-Distill-GGUF",
        temperature=settings["temperature"],
        max_tokens=settings["max_tokens"],
        streaming=True,
    )
    
    memory = ConversationBufferMemory(return_messages=True)
    chain = ConversationChain(llm=llm, memory=memory, verbose=True)
    
    cl.user_session.set("chain", chain)
    cl.user_session.set("settings", settings)
    
    await cl.Message(
        content=f"设置已更新：创造性={settings['temperature']}，最大长度={settings['max_tokens']}"
    ).send()

@cl.on_message
async def on_message(message: cl.Message):
    chain = cl.user_session.get("chain")
    
    if not chain:
        # 如果链不存在，使用默认设置初始化
        llm = ChatOpenAI(
            model="Qwen3-4B-Thinking-2507-GPT-5-Codex-Distill-GGUF",
            temperature=0.3,
            max_tokens=2048,
            streaming=True,
        )
        memory = ConversationBufferMemory(return_messages=True)
        chain = ConversationChain(llm=llm, memory=memory, verbose=True)
        cl.user_session.set("chain", chain)
    
    # 显示思考过程
    msg = cl.Message(content="")
    await msg.send()
    
    # 流式响应
    response = ""
    async for chunk in chain.astream({"input": message.content}):
        if "response" in chunk:
            response += chunk["response"]
            await msg.stream_token(chunk["response"])
    
    await msg.update()

现在我们的应用有了这些增强功能：

可调节的参数设置（创造性、生成长度）
快捷提问按钮，方便常见问题
流式输出，体验更好
设置更新后自动重新初始化

5.3 运行chainlit应用

保存代码为app.py，然后在终端运行：

chainlit run app.py

打开浏览器访问 http://localhost:8000（chainlit默认端口），就能看到我们的编程助手界面了。

6. 实际应用案例与技巧

6.1 案例：多轮编程问题解决

让我们看一个实际的多轮对话例子，展示记忆功能的价值：

用户：我想用Python处理一个JSON文件，该怎么读？

AI：你可以使用Python内置的json模块。这里是一个基本示例：

import json

# 读取JSON文件
with open('data.json', 'r', encoding='utf-8') as f:
    data = json.load(f)

print(data)

用户：文件很大，有10GB，怎么高效读取？

AI：对于大文件，建议使用ijson库进行流式解析，避免一次性加载到内存：

import ijson

# 流式解析大JSON文件
with open('large_data.json', 'rb') as f:
    for item in ijson.items(f, 'item'):
        # 逐项处理
        process_item(item)

用户：那如果我只想读取特定字段呢？

AI：使用ijson可以只解析需要的部分。比如只读取'users'数组：

import ijson

user_ids = []
with open('large_data.json', 'rb') as f:
    # 只解析users字段的id
    for user in ijson.items(f, 'users.item'):
        user_ids.append(user['id'])

注意AI如何理解上下文：当用户说"那如果我只想..."时，AI知道"那"指的是处理大JSON文件的问题，而不是从头开始解释JSON读取。

6.2 优化技巧：提升编程助手的实用性

6.2.1 添加代码格式化

让AI输出的代码更易读：

from langchain.callbacks.base import BaseCallbackHandler

class CodeFormattingCallback(BaseCallbackHandler):
    def on_llm_end(self, response, **kwargs):
        # 检测代码块并格式化
        import re
        
        content = response.generations[0][0].text
        # 查找Python代码块
        python_pattern = r'```python\n(.*?)\n```'
        matches = re.findall(python_pattern, content, re.DOTALL)
        
        for code in matches:
            # 这里可以添加代码格式化逻辑
            # 比如使用black或autopep8格式化代码
            formatted_code = self.format_python_code(code)
            # 替换原代码
            content = content.replace(code, formatted_code)
        
        return content
    
    def format_python_code(self, code):
        # 简单的代码格式化示例
        # 实际可以使用black库
        lines = code.split('\n')
        formatted_lines = []
        indent_level = 0
        
        for line in lines:
            stripped = line.strip()
            if stripped.endswith(':'):
                formatted_lines.append(' ' * indent_level * 4 + stripped)
                indent_level += 1
            elif stripped.startswith(('return', 'break', 'continue', 'pass')):
                indent_level = max(0, indent_level - 1)
                formatted_lines.append(' ' * indent_level * 4 + stripped)
            else:
                formatted_lines.append(' ' * indent_level * 4 + stripped)
        
        return '\n'.join(formatted_lines)

6.2.2 添加错误处理

编程代码经常会有错误，让AI能够处理这些情况：

from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate

# 创建错误处理链
error_handling_prompt = PromptTemplate(
    input_variables=["error", "code", "question"],
    template="""用户遇到了一个编程错误：
错误信息：{error}

相关代码：
{code}

原始问题：{question}

请分析错误原因并提供修正建议。"""
)

error_chain = LLMChain(llm=llm, prompt=error_handling_prompt)

# 在对话中集成错误处理
async def handle_programming_question(question, previous_code=None):
    try:
        # 尝试生成代码
        response = await conversation.apredict(input=question)
        
        # 如果用户反馈错误
        if "错误" in question or "报错" in question or "不行" in question:
            if previous_code:
                # 分析错误
                error_analysis = await error_chain.apredict(
                    error=question,
                    code=previous_code,
                    question=original_question
                )
                return error_analysis
        
        return response
    except Exception as e:
        return f"处理问题时出错：{str(e)}"

6.3 扩展功能：添加工具调用

真正的编程助手应该能执行代码。让我们添加一个安全的代码执行工具：

import subprocess
import tempfile
import os

class SafeCodeExecutor:
    def __init__(self):
        self.timeout = 10  # 10秒超时
    
    def execute_python(self, code):
        """安全执行Python代码"""
        # 创建临时文件
        with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
            f.write(code)
            temp_file = f.name
        
        try:
            # 在受限环境中执行
            result = subprocess.run(
                ['python', temp_file],
                capture_output=True,
                text=True,
                timeout=self.timeout,
                env={**os.environ, 'PYTHONPATH': ''}  # 限制导入
            )
            
            output = result.stdout
            if result.stderr:
                output += f"\n错误：{result.stderr}"
            
            return output
        except subprocess.TimeoutExpired:
            return "执行超时，可能代码有无限循环"
        except Exception as e:
            return f"执行出错：{str(e)}"
        finally:
            # 清理临时文件
            os.unlink(temp_file)

# 在LangChain中作为工具使用
from langchain.tools import Tool

code_executor = SafeCodeExecutor()

tools.append(
    Tool(
        name="PythonCodeRunner",
        func=code_executor.execute_python,
        description="安全执行Python代码并返回结果。输入应该是完整的Python代码。"
    )
)

注意：在实际生产环境中，代码执行需要更严格的安全措施，比如使用Docker沙箱、资源限制等。