项目背景

我们需要维护一批测试数据。该测试数据包括多个时间序列的值，表示从小时x到小时y的数据。因为时间范围比较大，由人工对数据进行比对和编辑比较麻烦。同时，我们希望非技术人员可以对该数据进行核对和直接修改。我们现在希望达到的目标是在本地setup一个环境，使得技术人员和非技术人员都可以对该测试数据进行维护。

解决方案

我们在本地setup Ollama和运行开源模型 - qianwen3:14b。使用AI model的目的是可以处理自然语言，这样使得非技术人员可以同样方便的对数据进行维护。另外，我们setup一个jupyter notebook，用来运行ai agent。使用jupyter notebook的主要原因是借助于matplot lib，我们可以方便的将测试数据的图表化，方便维护人员进行查看。同时我们需要定义一系列的tools，以供ai agent调用来对数据进行真正的处理。整体系统的架构图如下所示：

实现细节

搭建开发环境

这里的实现是基于mac操作系统。对于其他的操作系统区别不大。编程语言选择python。

首先，安装Ollama。过程省略，请参考Ollama网站。然后下载并启动qianwen3:14b：

ollama run qianwen3:14b

然后，我们setup python运行环境。我们创建一个本项目的folder，然后启动vscode。在vscode的terminal里用下面的命令建立virtual environment，并启动：

python -m venv venv
source ./venv/bin/activate

下面在folder的根目录下创建文件requirements.txt，在文件里定义python环境需要包含的package如下：

pandas
matplotlib 
numpy 
pylzma 
ipykernel 
jupyter
langchain
langchain-core
langchain-community
langchain-ollama

然后运行下面的命令安装这些被依赖的package：

pip install -r requirements.txt

最后一步，建立一个空的jupyter notebook。在根目录下，建立一个文件，以".ipynb"结尾，然后双击该文件在编辑框中打开该jupyter notebook。

编码

数据schema

我们使用LangChain作为ai agent的architecture。首先我们定义一下我们数据的schema。这里仅举一个例子，事实上python处理任何schema都是很容易的。

{"level": 5,"hierarchy": {"level_5": "System"},"time": 1735526775,"values": [{"status": "NORMAL","hour": -8,value": 25},{"status": "MODERATE","hour": -7,"value": 60}, ..., {"status": "HIGH","hour": 120,"value": 80}]}
{"level": 4,"hierarchy": {"level_5": "System", "level_4": "Department"},"time": 1735526775,"values": [{"status": "NORMAL","hour": -8,value": 25},{"status": "MODERATE","hour": -7,"value": 60}, ..., {"status": "HIGH","hour": 120,"value": 80}]}
... 

我们的数据文件里分为多行，每行是一条记录。一条记录内部有一个成为"hierarchy"的部分，里面包含从”level_0“到”level_5“的信息。"level_5"为根。每一层仅包含从当前层到根的信息。比如第三层的hierarchy，仅包含从第三层到第五层的信息，而不包含第零层到第二层的信息。另一个重要的部分是"values"。"values"是一只列表，包含多个value的记录。我们目前仅关心"value"的值，可以忽略掉其它的attributes。

普通方法定义

我们首先定义两个方法，它们实现了记录的查找和values的值提取功能。这两个方法作为ai agent的tools的主要功能的实现。

import json


# Search lines in file.
def search_in_file(
    file_path: str,
    keywords: list
) -> list:
    print(f"search file is called, {file_path}, {keywords}")
    matches = []

    with open(file_path) as f:
        for i, line in enumerate(f):
            
            if 'hierarchy' not in line:
                continue
            
            hierarchy = json.loads(line)['hierarchy']
            if len(hierarchy) != len(keywords):
                continue
            
            matched = True
            for keyword in keywords:
                if not keyword in hierarchy.values():
                    matched = False
                    break
                
            if matched:
                matches.append(f"{line}")
    return matches

# Extract pressure values from pressure.
def extract_pressure_values(pressures: list) -> list:
    print("extract_pressure_values is called")
    values = []
    
    for pressure in pressures:
        structured_pressure = json.loads(pressure)
        values = structured_pressure['values']
        value = [value_record['value'] for value_record in values]
        values.append(value)

    return values

Tools

下面基于这两个方法定义"tools"。"tools"可以理解为 Agent 的“外部能力扩展”。这里我们不对LangChain做具体介绍。关于LangChain的用法可以在其官网查看代码示例和文档。

import re
import matplotlib.pyplot as plt
import matplotlib.colors as mcolors
from langchain.tools import tool

def get_color(v):
    if v <= 49:
        return "green"
    elif v <= 65:
        return "yellow"
    elif v <= 74:
        return "orange"
    else:
        return "red"

@tool
def visualize_pressures(file_path: str, hierarchy: list, title: str) -> None:
    """Visualize pressures."""
    print(f"visualize data is called, {file_path}, {hierarchy}, {title}")
    
    if len(hierarchy) == 1:
        hierarchy_items = [hierarchy_item.strip() for hierarchy_item in re.split(r'[^A-Za-z -_]+', hierarchy[0]) if hierarchy_item.strip()]
    else:
        hierarchy_items = hierarchy
    
    pressures = search_in_file(file_path, hierarchy_items)
    for values in extract_pressure_values(pressures):
        hours = range(-8, -8 + len(values))
        
        colors = [get_color(v) for v in values]
        plt.figure(figsize=(15, 5))
        plt.bar(hours, values, color=colors)

        plt.xlabel("Hour")
        plt.ylabel("Value")
        plt.title(title)

        plt.show()

"tools"的方法需要三个参数：

• file_path: 数据文件的路径
• hierarchy: 对应于数据记录里的hierarchy部分信息。我们用hierarchy来定位需要维护的数据记录
• title：我们的tools将使用matplotlib绘图，这个参数是图的title。

代码的逻辑比较简单。我们利用刚才的两个方法找到需要处理的数据记录，然后将value的值提取为一个list，然后用matplotlib绘制柱状图。有两个地方的逻辑需要稍微介绍一下：

1. 关于下面的逻辑：

if len(hierarchy) == 1:
    hierarchy_items = [hierarchy_item.strip() for hierarchy_item in re.split(r'[^A-Za-z -_]+', hierarchy[0]) if hierarchy_item.strip()]

我们传入的hierarchy期望被解析为一个list，也就是["system", "department", ...]，但是ai agent有时将一个hierarchy解析为一个字符串，例如"system,department"。我们判断如果只有一个元素，则尝试将这个元素分解为多个元素。

2. 关于下面的逻辑：

def get_color(v):
    if v <= 49:
        return "green"
    elif v <= 65:
        return "yellow"
    elif v <= 74:
        return "orange"
    else:
        return "red"

我们针对不同的value值绘画成不同的颜色，以利于对值的合法性进行查看。

Agent和model

首先我们如下定义model并将我们定义好的tools捆绑到model上：

from langchain_ollama import ChatOllama
from langchain.agents import create_agent

model = ChatOllama(model='qwen3:14b')

llm_with_tools = model.bind_tools([visualize_pressures])

在这里我们没有定义真正意义上的ai agent，而是直接使用model和model捆绑的tools。其原因如下。正常使用ai agent处理prompt时，处理的流程为：user prompt -> llm model -> tools -> llm model -> user。也就是说在tools返回的值基础上，ai agent会在返回给user之前使用ai model对返回的结果做一次推理。因为我们使用的是本地开源ai model，功能相对较弱，对应输出的内容和格式控制不是很好。为了加强输出格式和内容的稳定性，我们将只使用ai model对调用的tools进行推理，然后直接调用tools并返回结果，省略掉返回给用户之前的推理。调用tools的逻辑如下：
 

tool_map = {
    "visualize_pressures": visualize_pressures,
}

def execute_query(user_query: str, llm_with_tools=llm_with_tools) -> None:
    response = llm_with_tools.invoke(user_query)

    for tc in response.tool_calls:
        tool = tool_map[tc["name"]]
        tool.invoke(tc["args"])

方法execute_query()接收用户的prompt，调用ai model并查看tool_calls。这一部分是ai model对tools调用的推理结果。我们根据这一步的结果，直接调用tools，并且传入ai model生成的调用参数。

Demo和结尾

我们将输入自然语言作为prompt，然后ai model会对我们的prompt进行推理，并生成适当的输入参数。下面是我们的prompt定义：
 

user_query = """
visualize pressure of 'System, CH-MT2500SD' in file ./data/pressure.txt.
"""

execute_query(user_query)

我们会看到tools被正常解析和调用：

visualize data is called, ./data/pressure.txt, ['System', 'CH-MT2500SD'], CH-MT2500SD
search file is called, ./data/pressure.txt, ['System', 'CH-MT2500SD']
extract_pressure_values is called

正确的图表将显示在我们的jupyter notebook里：

在本项目中，我们尝试了使用Ollama的本地ai model解决由自然语言调用代码来解决真实技术问题的案例。我们也总结了一些本地ai model的不足和应对方法。接下来，我们可以进一步丰富我们的tools和ai agent的功能。同时我们可以尝试使用LangGraph或者其他ai agent的framework解决我们现在实现方案的不足。