Ollama + LangGraph + Agent 用户交互

环境: WSL2 >> Ubuntu 20.04 >> Docker: Docker(Ollama) + Docker(Python 3.12.13)

认知:大模型是过去时

通过调用大模型的API来实现大模型的应用,这种方式拿到的大模型回复不是进行时,而是过去时.我们给大模型法送一条指令,大模型会根据指令进行思考,然后回复.这个回复是大模型已经思考完,执行完的回复,在大模型思考或者执行过程中,我们无法打断它.

比如:我们叫大模型新建一个文本文档,当大模型新建文档时,发现文档已经存在了,他就会停止新建,并且回复我们文档已经存在了,是否需要覆盖.这个时候我们表观上感觉他在和我们进行语音交互,实际上他回复我们后就关闭了思考回到了原始状态,从大模型的角度来说,刚才发生了什么他已经不知道了.我们这个时候再发送:请覆盖文档,其实大模型他是无法理解的.如果对这个认识不到位,我们常常会犯下原则的错误.如:

from langchain.agents import create_agent
from langchain.tools import tool
from langchain.llms import ChatOllama
from langchain.messages import AIMessage, HumanMessage

@tool 
def write_file(filename: str, content: str, to_cover:bool = False) -> str:
try:
        if not filename.endswith('.txt'):
            return f'文件名{filename}不是文本文件'
        # 防止文件名包含特殊字符
        filename = os.path.basename(filename)
        if os.path.exists(filename) and not to_cover:
            return f'文件{filename}已存在, 提醒用户确认是否要覆盖?'
        else:
            with open(filename, 'w', encoding='utf-8') as f:
                f.write(content)
        return f'文件{filename}已成功写入, 写入内容{len(content)}个字符'

    except Exception as e:
        return f'写入文件{filename}时出错: {e}'

tools = [write_file]
llm = ChatOllama(model='llama2')
agent = create_agent(llm, tools)
response = agent.invoke(HumanMessage(content='请新建一个文本文档test.txt, 随机写入内容'))
for message in response['messages']:
    if isinstance(message, AIMessage):
        if '覆盖' in message.content:
            agent.invoke(HumanMessage(content='是'))
#....

表面上看没问题,运行过程中也不报错,但就是无法生成文件.因为在大模型思考结束后,你再发送’是’的时候,大模型已经忘记刚才要做什么了,所以他对你发送的’是’就无法理解了.

常规解决方法:每次invoke的时候,都要发送完整的消息,让大模型从头理解会话记录,然后执行命令.

# ...
tools = [write_file]
llm = ChatOllama(model='llama2')
agent = create_agent(llm, tools)
history_messages = []
history_messages.append(HumanMessage(content='请新建一个文本文档test.txt, 随机写入内容'))
response = agent.invoke(history_messages)
for message in response['messages']:
    if isinstance(message, AIMessage):
        history_messages.append(message)
        if '覆盖' in message.content:
            history_messages.append(HumanMessage(content='是'))
            agent.invoke(history_messages)
# ...

这样就可以成功覆盖文件了.但你仔细观察就会发现,这样也时而成功,时而失败.并且你还会发现大模型最开始给你的随机内容,和真实写入文件的随机内容不一样?因为同样的操作,你让大模型执行了两次,结果是两次的随机内容不一样.这样的问题也可以解决,因为我们在遍历messages的时候,忽略了ToolCall的message就是ToolMessage,所以我们把ToolMessage的判断继续加入for循环中.

这只是一个示例,然而在实际操作中,逻辑会复杂的多,那么for里面的内容就会复杂得多.随着逻辑复杂,我们for的主体就会越来越乱,没有固定的结构,维护起来就会越来越困难,到最后甚至维护不了,直接重写.

LangGraph 解决前面的问题

这里不是教程,而是解决前面的问题.所以LangGraph是什么,就烦请客官自寻官方文档.

"""
    langgraph_confirm.py 通过状态图实现智能体 call function 并且与用户进行交互
    LangGraph(智能体编排框架) 
        StateGraph 状态图,通过状态来定义智能体的运行流程.
    在智能体运行链中定义节点, 并根据节点的状态进行路由.也可以循环执行节点.
    entry_point 启动节点
    node 节点
    edge 边, 用于连接节点. 可以根据节点的状态进行路由. 也可以循环执行节点.
    conditional_edges 条件边, 用于根据节点的状态进行路由.
"""
import uuid, os, pprint
from langchain.messages import HumanMessage, AIMessage, SystemMessage, ToolMessage
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages, BaseMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from typing import Annotated, TypedDict, Literal
from langchain_ollama import ChatOllama

# ==============================================================
# 状态定义
# 状态会在节点中传递, 用于存储节点的状态.所以历史会话也保留在状态中.
# 运行id 用于标识运行,避免不同运行之间的状态冲突.
# ==============================================================
class State(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]        # Annotated定义赋值/更新规则:[列表,追加] 一条一条累加的messages BaseMessage是langchain的消息基类
    need_confirm:bool                                           # 是否需要用户确认
    confirm_text:str                                            # 确认文本
    run_id:str                                                  # 运行id

# ==============================================================
# 工具定义 Tools
# ==============================================================
def write_file(filename:str, content:str, to_cover:bool = False) -> str:
    """
        写入文件
        参数: filename 文件名, content 内容, to_cover 是否覆盖, 默认False,需要用户确认后方可改为True
        返回: 写入结果
    """
    try:
        mode = "w" if to_cover else "x"
        with open(filename, mode, encoding="utf-8") as f:
            f.write(content)
        return f"文件{filename}已创建"
    except FileExistsError as e:
        return f'文件{filename}已存在, 需要用户确认是否要覆盖?(y/n)'
    except Exception as e:
        return f'文件{filename}创建失败: {e}'
    
def directory_list(directory:str = ".") -> str:
    try:
        files = os.listdir(directory)
        return "\n".join(files)
    except Exception as e:
        return f'目录{directory}不存在: {e}'

def random_txt(length:int = 10)->str:
    """
        随机生成中文字符串
        参数: length 字数, 默认10
        返回: 随机生成的中文字符串
    """
    res = []
    for _ in range(length):
        # 随机中文编码
        code = random.randint(0x4e00, 0x9fa5)
        res.append(chr(code))
    return ''.join(res)

# ==============================================================
# 模型定义
# bind_tools 绑定工具, 这种方式绑定工具,工具的优先级比 langchain中create_agent 绑定工具优先级高, 所以提示词一定要着重强调工具的调用.
# prompt | llm 将提示词绑定到模型中, 并返回一个模型对象(调用模型对象的invoke方法, 可以运行模型),最后将这个对象加入到状态图中, 用于在节点中调用模型
# ==============================================================
prompt = ChatPromptTemplate.from_messages([
    ("system","""你是一名文明的文件助手,你要按照以下规则进行操作:
     1: 你必须使用中文描述你要进行的操作
     2: 当用户要求写文件时,你必须同时**完成两件事**:
        - 在content中说明你将要执行的操作请请求确认
        - 在tool_calls中附上对应的工具调用
     3: 你必须严格调用工具,不能直接执行代码
     4: 操作文件前必须得到用户的确认,确认用语类似:我现在将创建文件,您是否确认继续操作(y/n)? 
     5: 用户输入回答y或者确认或者继续表示用户确认,你必须继续执行操作
     6: 没有用户的确认,你必须拒绝执行函数
     """),
    MessagesPlaceholder("messages"),        # 占位符,添加历史消息       模型接受消息一般是: system, messages
])

tools = [write_file, directory_list, random_txt]
llm = ChatOllama(
    model="qwen2.5:7b",
    temperature=0.2,
    base_url="http://host.docker.internal:11434",
).bind_tools(tools)

tool_llm = prompt | llm     # 绑定prompt和模型  tool_ll就是运行模型的接口(模型对象)

def call_model(state:State):
    messages = state['messages']
    print(f'{'*'*50} messages:')
    pprint.pprint(messages)
    response = tool_llm.invoke(state)
    print(f'{'*'*50} response消息:')
    pprint.pprint(response)
    return {
        'messages': [response],
        'need_confirm': any(key in response.content for key in ['确认', 'y/n', '继续', 'y', '吗？']),
    }

def execute_tool(state:State):
    message = state['messages'][-1]
    calls = message.tool_calls
    results = []
    for call in calls:
        fun = {f.__name__: f for f in tools}[call['name']]
        results.append(ToolMessage(content=fun(**call['args']), tool_call_id=call['id']))
    return {
        'messages': results,
        'need_confirm': False,
    }

def human_confirm(state:State):
    message = state['messages'][-1]
    human_input = input(f'[AI ASK]: {message.content}')
    return {
        'messages': [HumanMessage(content=human_input.strip())],
        'need_confirm': False,
        'confirm_text': human_input.strip(),
        'run_id': message.id,
    }

def judge_node(state:State) -> Literal["human_confirm", "execute_tool", "end"]:
    message = state['messages'][-1]
    print(f'{'*'*50} need_confirm:{state['need_confirm']}')
    
    if state['need_confirm']:
        return 'human_confirm'
    if hasattr(message, 'tool_calls') and message.tool_calls:
        return 'execute_tool'
    return 'end'

# ==============================================================
# 通过状态图(StateGraph)创建代理(agent)
# ==============================================================
def create_confirm_agent():
    # 实例化状态图
    graph = StateGraph(State)
    # 添加节点
    graph.add_node('call_model', call_model)
    graph.add_node('execute_tool', execute_tool)
    graph.add_node('human_confirm', human_confirm)
    # graph.add_node('judge_node', judge_node)

    # 入口
    graph.set_entry_point('call_model')
    # 添加条件边界(可以有多个条件边界)
    graph.add_conditional_edges('call_model', judge_node, {'human_confirm': 'human_confirm', 'execute_tool': 'execute_tool', 'end': END})
    # 添加边
    graph.add_edge('human_confirm', 'call_model')
    graph.add_edge('execute_tool', 'call_model')

    

    return graph.compile()      # 编译状态图, 生成智能体.(agent)

if __name__ == '__main__':
    agent = create_confirm_agent()
    state = {
        "messages":[
            HumanMessage(content="请创建一个txt文件, 文件名test.txt, 并且调用工具生成随机中文内容,写入文件"),
        ],
        "need_confirm": False,
        "confirm_text": "",
        "run_id": str(uuid.uuid4()),
    }

    print(f'{'*'*50} 测试写入文件功能 {'*'*50}')
    apply = agent.invoke(state)
    print(f'{'*'*50} 测试结果:')
    print('\n')
    pprint.pprint(apply)


"""
流程图
Start: 流程开始。
call_model: 调用 LLM 生成响应，判断是否需要确认或调用工具。
judge_node: 根据状态进行路由：
    若 need_confirm 为真，进入 human_confirm。
    若存在 tool_calls，进入 execute_tool。
    否则，进入 END。
human_confirm: 等待用户输入确认信息，完成后返回 call_model。
execute_tool: 执行具体工具函数，完成后返回 call_model。
END: 流程结束
"""

内容优化方向

结构化输出, 例如:json格式,

checkpointer 保存对话历史

将同步invoke改为异步ainvoke