最近学习langgraph，试试了官方的项目，基于agent的方式构建一个机器人，挺好玩的，分享给大家。
在这个例子中，我们将创建一个聊天机器人，帮助用户生成提示。它将首先从用户那里收集需求，然后生成提示（并根据用户输入对其进行细化）。这些被分为两个独立的状态，LLM决定何时在它们之间转换。系统的图形表示可以在下面找到。
收集信息
首先，让我们定义图中收集用户需求的部分。这将是一个带有特定系统消息的LLM调用。它将可以访问一个工具，当它准备好生成提示时可以调用该工具。
from typing import Listfrom langchain_core.messages import SystemMessagefrom langchain_openai import ChatOpenAIfrom pydantic import BaseModel
模型我们使用qwen2.5 目前刚开源，基于ollama调用
template = """你的工作是从用户那里获取他们想要创建哪种类型的提示模板的信息。您应该从他们那里获得以下信息：-提示的目的是什么-将向提示模板传递哪些变量-输出不应该做什么的任何限制-输出必须遵守的任何要求如果你无法辨别这些信息，请他们澄清！不要试图疯狂猜测。在您能够辨别所有信息后，调用相关工具。"""def get_messages_info(messages):return [SystemMessage(content=template)] + messagesclass PromptInstructions(BaseModel):"""关于如何提示LLM的说明prompt"""objective: strvariables: List[str]constrains:List[str]requirements: List[str]llm =ChatOpenAI(model="qwen2.5", openai_api_key="ollama", openai_api_base="http://localhost:11434/v1/")llm_with_tool = llm.bind_tools([PromptInstructions])# 根据用户的输入 调用 需要的工具def info_chain(state):messages = get_messages_info(state['messages'])response = llm_with_tool.invoke(messages)return {"messages":[response]}
生成提示
我们现在设置将生成提示的状态。这将需要一个单独的系统消息，以及一个在工具调用之前过滤掉所有消息的函数（因为此时之前的状态决定是时候生成提示了
from langchain_core.messages import AIMessage,HumanMessage, ToolMessageprompt_system = """根据以下要求，编写一个好的提示模板:{reqs}"""def get_prompt_messages(messages:list):tool_call = Noneother_msgs = []for m in messages:if isinstance(m, AIMessage) and m.tool_calls:tool_call = m.tool_calls[0]['args']elif isinstance(m, ToolMessage):continueelif tool_call is not None:other_msgs.append(m)return [SystemMessage(content=prompt_system.format(reqs=tool_call))] + other_msgsdef prompt_gen_chain(state):messages = get_prompt_messages(state["messages"])response = llm.invoke(messages)return {"messages":[response]}
定义状态逻辑
这是聊天机器人所处状态的逻辑。如果最后一条消息是工具调用，那么我们处于“提示创建者”（prompt）应该响应的状态。否则，如果最后一条消息不是HumanMessage，那么我们知道人类应该下一条响应，所以我们处于END状态。如果最后一条消息是HumanMessage，那么如果之前有工具调用，我们处于提示状态。否则，我们处于“信息收集”（info）状态。
from typing import Literalfrom langgraph.graph import END# 定义一个函数，用于获取当前状态def get_state(state)-> Literal["add_tool_message","info","__end__"]:messages = state["messages"]if isinstance(messages[-1],AIMessage) and messages[-1].tool_calls:return "add_tool_message"elif not isinstance(messages[-1], HumanMessage):return ENDreturn "info"
创建图形
我们现在可以创建图形。我们将使用SqliteSaver来保存对话历史记录。
# Create the graphfrom langgraph.checkpoint.memoryimport MemorySaverfrom langgraph.graph import StateGraph,STARTfrom langgraph.graph.message import add_messagesfrom typing import Annotatedfrom typing_extensions import TypedDictclass State(TypedDict):messages: Annotated[list, add_messages]memory = MemorySaver()workflow = StateGraph(State)workflow.add_node("info",info_chain)workflow.add_node("prompt",prompt_gen_chain)# 工具调用用来做提示词生成@workflow.add_nodedef add_tool_message(state:State):return {"messages":[ToolMessage(content="Prompt generated!",tool_call_id= state["messages"][-1].tool_calls[0]['id'])]}workflow.add_edge(START,"info")workflow.add_conditional_edges("info",get_state)workflow.add_edge("add_tool_message","prompt")workflow.add_edge("prompt",END)graph = workflow.compile(checkpointer=memory)
打印图
执行图
import uuidconfig = {"configurable":{"thread_id":str(uuid.uuid4())}}while True:user = input("User (q/Q to quit): ")print(f"User (q/Q to quit): {user}")if user in {"q", "Q"}:print("AI: Byebye")breakoutput = Nonefor output in graph.stream({"messages":[HumanMessage(content=user)]}, config=config, stream_mode="updates"):last_message = next(iter(output.values()))["messages"][-1]last_message.pretty_print()if output and "prompt" in output:print("Done!")
测试效果
使用程序，需要输入4个条件
- 提示的目的是什么？- 将向提示模板传递哪些变量？（例如，在这个例子中提到的数组）- 输出不应该做什么的任何限制。- 输出必须遵守的任何要求。
首先测试一个 排序算法
1、写一个快速排序算法 2、“1,3,2,5“ 3、no 4、no
测试一个 ，写故事的
当你输入不完整时，他会自动要求你继续输入，还是很不错的，
哈哈试试吧
