用LangChain打造一个可以管理日程的智能助手

存储设计
定义工具
创建llm
提示词模板
创建Agent
执行
总结

众所周知，GPT可以认为是一个离线的软件的，对于一些实时性有要求的功能是完全不行，比如实时信息检索，再比如我们今天要实现个一个日程管理的功能，这个功能你纯依赖于ChatGPT或者其他大语言模型（后文简称llm），是完全实现不了的，比如你这次让他帮你记录个日程，你要是和他聊的内容过多，历史聊天记录滚动覆盖了就找不回来了。你要是换个聊天窗口，之前的日程信息你就更找不回来了，其根本原因是目前所有的llm都是无状态的，每轮对话必须携带所有历史聊天记录才能实现多轮对话，而所有的llm都有输入长度限制，比如gpt4目前是128k。

存储设计

所以，如果要实现日程记录永不丢失我们就需要用第三方存储来记录所有的日程信息，这里为了简单，我直接使用了sqlite3（用mysql或者其他存储都是可以的），我创建了一个非常简单的日程表，只有一个时间和描述，整体代码如下：

# 连接到 SQLite 数据库
# 如果文件不存在，会自动在当前目录创建一个名为 'langchain.db' 的数据库文件
import sqlite3
conn = sqlite3.connect('langchain.db')

# 创建一个 Cursor 对象并通过它执行 SQL 语句
c = conn.cursor()
# 创建表
c.execute('''
create table if not exists schedules 
(
    id          INTEGER    primary key autoincrement,
    start_time  TEXT default (strftime('%Y-%m-%d %H:%M:%S', 'now', 'localtime')) not null,
    description text default ''                                                  not null
);
''')

conn.commit()
conn.close()
print("数据库和表已成功创建！")

定义工具

那么接下来的问题就是如何让GPT能够查询和操作这个表了。这里我们直接使用了LangChain的@tool装饰器，讲schedules表的基本操作设置为GPT可以识别的接口，当然使用OpenAI的纯原始接口也是可以实现的（参加我之前的文章OpenAI的多函数调用）,就是代码量相对会多很多。具体的代码如下，这里我定义了对schedules表的增、删、查的功能。


def connect_db():
    """ 连接到数据库 """
    conn = sqlite3.connect('langchain.db')
    return conn
    
@tool
def add_schedule(start_time : str, description : str) -> str: 
    """ 新增日程，比如2024-05-03 20:00:00, 周会 """
    conn = connect_db()
    cursor = conn.cursor()
    cursor.execute("""
        INSERT INTO schedules (start_time, description) VALUES (?, ?);
    """, (start_time, description,))
    conn.commit()
    conn.close()
    return "true"

@tool
def delete_schedule_by_time(start_time : str) -> str:
    """ 根据时间删除日程 """
    conn = connect_db()
    cursor = conn.cursor()
    cursor.execute("""
        DELETE FROM schedules WHERE start_time = ?;
    """, (start_time,))
    conn.commit()
    conn.close()
    return "true"
    
@tool
def get_schedules_by_date(query_date : str) -> str:
    """ 根据日期查询日程，比如 获取2024-05-03的所有日程 """
    conn = connect_db()
    cursor = conn.cursor()
    cursor.execute("""
        SELECT start_time, description FROM schedules WHERE start_time LIKE ?;
    """, (f"{query_date}%",))
    schedules = cursor.fetchall()
    conn.close()
    return str(schedules)

创建llm

到这里，所以依赖的逻辑就已经完成了，接下来就是创建agent了，首先就是想定义好llm，这里我还是选用了OpenAI的gpt3.5，（个人认为这是目前性价比最高的模型），注意llm必须要调用bind_tools方法绑定好我们上面声明好的工具。

## 创建llm
llm = ChatOpenAI(model="gpt-3.5-turbo", max_tokens=4096)
tools = [add_schedule, delete_schedule_by_time, get_schedules_by_date]
llm_with_tools = llm.bind_tools(tools)

提示词模板

然后就是创建提示词模板，这里额外提一下，因为目前所有的llm都不具备对时间的感知能力，所以这里必须在模板里将当前时间传给llm，方便llm去做时间的计算。

## 创建提示词模板  
prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "你是一个日程管理助手",
        ),
        ("placeholder", "{chat_history}"),
        ("user", "{input} \n\n 当前时间为：{current_time}"),
        ("placeholder", "{agent_scratchpad}"),
    ]
)

创建Agent

之后就是创建agent和执行器了，这里自己创建一个一遍，又直接使用了LangChain封装好的方法创建了一遍，二者功能上没有区别，区别就是直接用别人的方法，自己可以少写两行代码。


## agent创建方式1 
from langchain.agents.format_scratchpad.openai_tools import (
    format_to_openai_tool_messages,
)
agent = (
    {
        "current_time": lambda x: x["current_time"],
        "input": lambda x: x["input"],
        "agent_scratchpad": lambda x: format_to_openai_tool_messages(
            x["intermediate_steps"]
        ),
    }
    | prompt
    | llm_with_tools
    | OpenAIToolsAgentOutputParser()
)

## agent创建方式2
agent = create_tool_calling_agent(llm_with_tools, tools, prompt)  
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=False)

执行

用如下的方式就可以执行agent验证功能是否可以正常了。

invoke(
        {
            "input": "查询下我明天有啥安排？",
            "current_time": datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')  # 当前时间必须传
        }
    )

这里我简单实现了一个多轮对话用来验证各功能是否正常。


def ask(question):
    res = agent_executor.invoke(
        {
            "input": question,
            "current_time": datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        }
    )
    return res["output"]

while True:
    question = input(">")
    if question.lower() == '退出':
        break
    print(ask(question))

> 删除今天所有的日程
已成功删除今天所有的日程。
> 创建一套明天晚上6点的日程，开周会
日程已成功创建，明天晚上6点有周会安排。
> 我明天第一条日程是啥？
您明天的第一条日程是沟通会，时间为2024-05-05 09:00:00。祝您顺利！
> 看下我明天早上10点有没有安排？
明天早上10点没有安排，您的日程是：
- 09:00:00 沟通会
- 18:00:00 周会
> 把我明天早上9点的会议改到10点
已成功将您明天早上9点的会议改到10点。

总结

日程管理的能力本质上还是建立在llm的函数调用能力，说白了其实你告诉llm有什么样的函数可以调用，然后让llm自行决策是否需要调用，这也是当下llm智能的体现。使用LangChain其实也只是将函数的定义、调用以及结果返回的流程简化而已。这里额外说下，上面代码中，我并未给llm提供修改日程的方法，但后续测试工程中我让它修改某个日程，它居然修改成功了，你猜它是怎么实现的？