langchain
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage, AIMessage
from langchain_ollama import ChatOllamafrom tool import PersonInfoTool# 配置Ollama模型
llm = ChatOllama(model="qwen3:4b",temperature=0.6,num_ctx=2048,verbose=True
)# 创建工具
person_tool = PersonInfoTool()
tools = [person_tool]agent_executor = create_agent(model=llm,tools=tools,system_prompt="你是一个会利用工具的助手。",
)for chunk in agent_executor.stream({"messages": [{"role": "user", "content": "帮我查询下王微微的出生年月？"}]},stream_mode="values"):latest_message = chunk["messages"][-1]if latest_message.content:if isinstance(latest_message, HumanMessage):print(f"user:{latest_message.content}")elif isinstance(latest_message, AIMessage):print(f"AI:{latest_message.content}")elif latest_message.tool_calls:print(f"Calling tools:{[tc['name'] for tc in latest_message.tool_calls]}")
langgraph
from typing import TypedDict, Annotatedfrom langchain_core.messages import AIMessage
from langchain_ollama import ChatOllama
from langgraph.graph import START, StateGraph
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_conditionfrom tool import PersonInfoTool# 配置Ollama模型
llm = ChatOllama(model="qwen3:4b",temperature=0.6,num_ctx=2048,verbose=True
)# 创建工具
person_tool = PersonInfoTool()
tools = [person_tool]class State(TypedDict):messages: Annotated[list, add_messages]graph_builder = StateGraph(State)llm_with_tools = llm.bind_tools(tools)def chatbot(state: State):return {"messages": llm_with_tools.invoke(state["messages"])}graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")tool_node = ToolNode(tools=tools)
graph_builder.add_node("tools", tool_node)graph_builder.add_conditional_edges("chatbot", tools_condition)graph_builder.add_edge("tools", "chatbot")
graph = graph_builder.compile()def stream_graph_updates(user_input: str):for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}):for value in event.values():if isinstance(value["messages"], AIMessage):print("Assistant:", value["messages"].content)else:print("Tool:", value["messages"][-1].content)stream_graph_updates("帮我查询下王微微的出生年月?")
# tools/person_info.py
from typing import Anyfrom langchain_core.tools import BaseToolclass PersonInfoTool(BaseTool):name: str = "person_info"description: str = "查询人物出生年份（输入格式：'人名'，例如'肖汪'）"def _run(self, name: str, **kwargs: Any) -> str:return "1980"async def _arun(self, name: str, **kwargs: Any) -> str:return self._run(name)
