)
Agent部分C-898S-989D-82N快点岛币吧组件Tool的问题和工具交给大语言模型然后大语言模型进行1个意图识别决定是否要执行工具如果需要执行工具的话执行工具并返回结果再返回给大语言模型然后代言模型根据结果生成我们的答案SDK这里直接调用openai的sdk结果是ChatCompletionMessage(content, refusalNone, roleassistant, annotationsNone, audioNone, function_callNone, tool_calls[ChatCompletionMessageFunctionToolCall(idcall_579422a0074c40f28d68f1, functionFunction(arguments{city: 北京, date: 2026-05-22}, nameget_weather), typefunction, index0)])from openai import OpenAI import json client OpenAI() # 1. 通过JSON结构定义工具包括工具名称描述参数等 tools [ { type: function, function: { name: get_weather, description: Get todays weather for a location., parameters: { type: object, properties: { city: { type: string, description: 城市名称, e.g. San Francisco, }, date :{ type:string, description:想要查询的天气的日期, e.g. 2023-12-25 } }, required: [city,date], additionalProperties: False, }, strict: True, }, }, ] def get_weather(city,date): return f{city} on {date} is cloudy with a chance of rain. messages [ {role: user, content: What is the weather like in 北京 on 2024-12-25?} ] # 2. Prompt the model with tools defined response client.chat.completions.create( modelgpt-4.1, messagesmessages, toolstools, ) messages.append(response.choices[0].message) for tool_call in response.choices[0].message.tool_calls or []: if tool_call.function.name get_weather: # 3. 执行工具函数的逻辑 args json.loads(tool_call.function.arguments) weather get_weather(args[city],args[date]) # 4. 将工具函数的执行结果添加到消息列表中 messages.append( { role: tool, tool_call_id: tool_call.id, content: json.dumps({weather: weather}), } ) response client.chat.completions.create( modelgpt-4.1, messagesmessages, toolstools, ) # 5. 模型会根据工具函数的执行结果生成最终的回复 print(response.choices[0].message.content)LANGCHAIN方式import json import os from dotenv import load_dotenv from langchain_core.messages import HumanMessage, ToolMessage from langchain_core.tools import tool from langchain_openai import ChatOpenAI from pydantic import BaseModel, Field # 1 使用langchain装饰器定义工具方法和约束 class GetWeatherArgs(BaseModel): # 定义参数city字符串类型带描述大模型会根据这个描述理解参数含义 city: str Field(description城市名称, 比方说 北京市 武汉市需要传递城市的全名) # 定义参数date字符串类型指定日期格式 date: str Field(description天气日期格式是yyyy-MM-dd) tool(description获取城市在特定日期的天气,args_schemaGetWeatherArgs) def get_weather(city: str, date: str): # 模拟天气查询实际项目中这里可以调用真实天气API return f城市{city}在{date}的天气是晴朗的 # 2 创建llm load_dotenv() # 初始化LLM大模型使用ChatOpenAI调用 llm ChatOpenAI( modelqwen-turbo, api_keyos.getenv(OPENAI_API_KEY), # 从.env读取千问的API Key base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, # 千问的OpenAI兼容接口地址 ) # 3 把工具和llm绑定 llm_with_tools llm.bind_tools([get_weather]) # 4 构建message human_message HumanMessage(content北京在2026-05-22的天气是多少) # 5 第一次调用llm意图识别 res llm_with_tools.invoke([human_message]) ai_message res print( 第一次调用大模型决定调用工具) print(res) # 6 从第一次llm返回结果获取工具名称和参数进行工具调用 tool_call res.tool_calls[0][args] # tool_res调用工具之后的结果 tool_res get_weather.invoke(tool_call) # 7 构建message包含问题 第一次调用llm结果 工具执行结果 # 8 第二次调用llm tool_id res.tool_calls[0][id] tool_message ToolMessage(tool_call_idtool_id, contenttool_res) # human_message: 用户问题 # ai_message第一次调用llm结果 # tool_message工具执行结果 final_result llm_with_tools.invoke([human_message, ai_message, tool_message]) # 打印最终回答 print(\n 模型第二次的回复结果最终答案\n, final_result)createagentimport os from dotenv import load_dotenv from langchain.agents import create_agent from langchain_core.messages.tool import tool_call from langchain_core.tools import tool from langchain_openai import ChatOpenAI # 创建llm load_dotenv() llm ChatOpenAI( modelqwen-turbo, # 千问模型 api_keyos.getenv(OPENAI_API_KEY), base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1 ) # 创建工具方法 tool def get_weather(city:str,date:str): 获取城市在特定日期的天气 return f在{city} {date}的天气是晴朗的 # 创建agent agent create_agent( modelllm, tools[get_weather], ) # 调用agent获取结果 res agent.invoke({messages:[{role:user, content:北京在2026-05-22的天气是多少}]}) print(res)MCPMCP的工作流程如下所示 1握手与能力声明Handshake Discovery当AI应用程序Host启动时它会根据配置启动 MCP Server通常是一个子进程。Host 记住了这些工具的名字和用法说明但此时 并没有执行任何代码 。 2用户提问与上下文注入Context InjectionUser提出问题后Host 把用户的这个问题加上刚才 Server 汇报的 工具说明书 一起打包发给 LLM。 3模型决策LLM ReasoningLLM决定是否需要去调用相关工具。 4路由与执行Routing ExecutionHost 收到指令发现是要调用 MCP 工具Host 通过 MCP 协议JSON-RPC给 MCP Server 发消息Server 收到请求 在它自己的进程如果是StreamableHttp方式则是在远端执行里执行相关操作。 所有的驱动依赖、复杂逻辑都在 Server 端。 5结果回传Result FeedbackServer 拿到结果打包成 MCP 响应发回给 HostHost再将相关结果传递给大模型让大模型做下一步的输出。服务端from mcp.server import FastMCP ## stdio传输协议 # 特点本身不是服务器方式使用本地命令行调用 # 不需要启动服务器端在客户端启动和调用 # 定义FastMCP实例 mcp FastMCP() # 创建工具、资源和提示模板 mcp.tool() def get_weather(city:str,date:str) - str: return f在{city}{date}的天气是下雨的 mcp.resource(greeting://default) def get_greeting() - str: return Hello from static resource! mcp.prompt() def greet_user(name: str, style: str friendly) - str: styles { friendly: 写一句友善的问候, formal: 写一句正式的问候, casual: 写一句轻松的问候, } return f为{name}{styles.get(style, styles[friendly])} if __name__ __main__: mcp.run(transportstdio)客户端from mcp.client.streamable_http import streamable_http_client from mcp import ClientSession async def main(): url http://localhost:8000/mcp async with streamable_http_client(urlurl) as (read_stream,write_stream,_): async with ClientSession(read_streamread_stream,write_streamwrite_stream) as session: # 1、初始化 await session.initialize() # 2、获取到server端所有工具 tools await session.list_tools() print(tools) # 3、调用tool result await session.call_tool(nameget_temperature,arguments{city:北京,date:2023-12-12}) print(result) if __name__ __main__: import asyncio asyncio.run(main())
Langchain(下)
发布时间：2026/6/10 23:09:24
Agent部分C-898S-989D-82N快点岛币吧组件Tool的问题和工具交给大语言模型然后大语言模型进行1个意图识别决定是否要执行工具如果需要执行工具的话执行工具并返回结果再返回给大语言模型然后代言模型根据结果生成我们的答案SDK这里直接调用openai的sdk结果是ChatCompletionMessage(content, refusalNone, roleassistant, annotationsNone, audioNone, function_callNone, tool_calls[ChatCompletionMessageFunctionToolCall(idcall_579422a0074c40f28d68f1, functionFunction(arguments{city: 北京, date: 2026-05-22}, nameget_weather), typefunction, index0)])from openai import OpenAI import json client OpenAI() # 1. 通过JSON结构定义工具包括工具名称描述参数等 tools [ { type: function, function: { name: get_weather, description: Get todays weather for a location., parameters: { type: object, properties: { city: { type: string, description: 城市名称, e.g. San Francisco, }, date :{ type:string, description:想要查询的天气的日期, e.g. 2023-12-25 } }, required: [city,date], additionalProperties: False, }, strict: True, }, }, ] def get_weather(city,date): return f{city} on {date} is cloudy with a chance of rain. messages [ {role: user, content: What is the weather like in 北京 on 2024-12-25?} ] # 2. Prompt the model with tools defined response client.chat.completions.create( modelgpt-4.1, messagesmessages, toolstools, ) messages.append(response.choices[0].message) for tool_call in response.choices[0].message.tool_calls or []: if tool_call.function.name get_weather: # 3. 执行工具函数的逻辑 args json.loads(tool_call.function.arguments) weather get_weather(args[city],args[date]) # 4. 将工具函数的执行结果添加到消息列表中 messages.append( { role: tool, tool_call_id: tool_call.id, content: json.dumps({weather: weather}), } ) response client.chat.completions.create( modelgpt-4.1, messagesmessages, toolstools, ) # 5. 模型会根据工具函数的执行结果生成最终的回复 print(response.choices[0].message.content)LANGCHAIN方式import json import os from dotenv import load_dotenv from langchain_core.messages import HumanMessage, ToolMessage from langchain_core.tools import tool from langchain_openai import ChatOpenAI from pydantic import BaseModel, Field # 1 使用langchain装饰器定义工具方法和约束 class GetWeatherArgs(BaseModel): # 定义参数city字符串类型带描述大模型会根据这个描述理解参数含义 city: str Field(description城市名称, 比方说 北京市 武汉市需要传递城市的全名) # 定义参数date字符串类型指定日期格式 date: str Field(description天气日期格式是yyyy-MM-dd) tool(description获取城市在特定日期的天气,args_schemaGetWeatherArgs) def get_weather(city: str, date: str): # 模拟天气查询实际项目中这里可以调用真实天气API return f城市{city}在{date}的天气是晴朗的 # 2 创建llm load_dotenv() # 初始化LLM大模型使用ChatOpenAI调用 llm ChatOpenAI( modelqwen-turbo, api_keyos.getenv(OPENAI_API_KEY), # 从.env读取千问的API Key base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, # 千问的OpenAI兼容接口地址 ) # 3 把工具和llm绑定 llm_with_tools llm.bind_tools([get_weather]) # 4 构建message human_message HumanMessage(content北京在2026-05-22的天气是多少) # 5 第一次调用llm意图识别 res llm_with_tools.invoke([human_message]) ai_message res print( 第一次调用大模型决定调用工具) print(res) # 6 从第一次llm返回结果获取工具名称和参数进行工具调用 tool_call res.tool_calls[0][args] # tool_res调用工具之后的结果 tool_res get_weather.invoke(tool_call) # 7 构建message包含问题 第一次调用llm结果 工具执行结果 # 8 第二次调用llm tool_id res.tool_calls[0][id] tool_message ToolMessage(tool_call_idtool_id, contenttool_res) # human_message: 用户问题 # ai_message第一次调用llm结果 # tool_message工具执行结果 final_result llm_with_tools.invoke([human_message, ai_message, tool_message]) # 打印最终回答 print(\n 模型第二次的回复结果最终答案\n, final_result)createagentimport os from dotenv import load_dotenv from langchain.agents import create_agent from langchain_core.messages.tool import tool_call from langchain_core.tools import tool from langchain_openai import ChatOpenAI # 创建llm load_dotenv() llm ChatOpenAI( modelqwen-turbo, # 千问模型 api_keyos.getenv(OPENAI_API_KEY), base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1 ) # 创建工具方法 tool def get_weather(city:str,date:str): 获取城市在特定日期的天气 return f在{city} {date}的天气是晴朗的 # 创建agent agent create_agent( modelllm, tools[get_weather], ) # 调用agent获取结果 res agent.invoke({messages:[{role:user, content:北京在2026-05-22的天气是多少}]}) print(res)MCPMCP的工作流程如下所示 1握手与能力声明Handshake Discovery当AI应用程序Host启动时它会根据配置启动 MCP Server通常是一个子进程。Host 记住了这些工具的名字和用法说明但此时 并没有执行任何代码 。 2用户提问与上下文注入Context InjectionUser提出问题后Host 把用户的这个问题加上刚才 Server 汇报的 工具说明书 一起打包发给 LLM。 3模型决策LLM ReasoningLLM决定是否需要去调用相关工具。 4路由与执行Routing ExecutionHost 收到指令发现是要调用 MCP 工具Host 通过 MCP 协议JSON-RPC给 MCP Server 发消息Server 收到请求 在它自己的进程如果是StreamableHttp方式则是在远端执行里执行相关操作。 所有的驱动依赖、复杂逻辑都在 Server 端。 5结果回传Result FeedbackServer 拿到结果打包成 MCP 响应发回给 HostHost再将相关结果传递给大模型让大模型做下一步的输出。服务端from mcp.server import FastMCP ## stdio传输协议 # 特点本身不是服务器方式使用本地命令行调用 # 不需要启动服务器端在客户端启动和调用 # 定义FastMCP实例 mcp FastMCP() # 创建工具、资源和提示模板 mcp.tool() def get_weather(city:str,date:str) - str: return f在{city}{date}的天气是下雨的 mcp.resource(greeting://default) def get_greeting() - str: return Hello from static resource! mcp.prompt() def greet_user(name: str, style: str friendly) - str: styles { friendly: 写一句友善的问候, formal: 写一句正式的问候, casual: 写一句轻松的问候, } return f为{name}{styles.get(style, styles[friendly])} if __name__ __main__: mcp.run(transportstdio)客户端from mcp.client.streamable_http import streamable_http_client from mcp import ClientSession async def main(): url http://localhost:8000/mcp async with streamable_http_client(urlurl) as (read_stream,write_stream,_): async with ClientSession(read_streamread_stream,write_streamwrite_stream) as session: # 1、初始化 await session.initialize() # 2、获取到server端所有工具 tools await session.list_tools() print(tools) # 3、调用tool result await session.call_tool(nameget_temperature,arguments{city:北京,date:2023-12-12}) print(result) if __name__ __main__: import asyncio asyncio.run(main())
)
